集成RabbitMQ+MQ常用操作

1.环境搭建

1.Docker安装RabbitMQ

1.拉取镜像

docker pull rabbitmq:3.8-management

2.安装命令

docker run -e RABBITMQ_DEFAULT_USER=sun \-e RABBITMQ_DEFAULT_PASS=mq \-v mq-plugins:/plugins \--name mq \--hostname mq \-p 15672:15672 \-p 5672:5672 \-d 699038cb2b96 # 注意这里是镜像id，需要替换

3.开启5672和15672端口

4.登录控制台

15672端口

2.整合Spring AMQP

1.sun-common模块下创建新模块

2.引入amqp依赖和fastjson

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><!-- 继承父模块的版本和通用依赖 --><parent><groupId>com.sunxiansheng</groupId><artifactId>sun-common</artifactId><version>1.0-SNAPSHOT</version></parent><artifactId>sun-common-rabbitmq</artifactId><!-- 子模块的version,如果不写就默认跟父模块的一样 --><version>${children.version}</version><!-- 自定义依赖,无需版本号 --><dependencies><!--AMQP依赖，包含RabbitMQ--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><!-- 用于传递消息时的序列化操作 --><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId></dependency></dependencies></project>

3.新建一个mq-demo的模块

1.在sun-frame下创建mq-demo

2.然后在mq-demo下创建生产者和消费者子模块

3.查看是否交给父模块管理了

4.在mq-demo模块引入sun-common-rabbitmq依赖

    <dependencies><!-- 引入sun-common-rabbitmq --><dependency><groupId>com.sunxiansheng</groupId><artifactId>sun-common-rabbitmq</artifactId><version>1.0-SNAPSHOT</version></dependency></dependencies>

5.publisher引入sun-common-test依赖

    <dependencies><!-- sun-common-test --><dependency><groupId>com.sunxiansheng</groupId><artifactId>sun-common-test</artifactId><version>1.0-SNAPSHOT</version></dependency></dependencies>

6.将sun-common-rabbitmq clean-install一下

7.给consumer和publisher都创建主类

1.ConsumerApplication.java

package com.sunxiansheng.consumer;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.ComponentScan;@SpringBootApplication
@ComponentScan("com.sunxiansheng")
public class ConsumerApplication {public static void main(String[] args) {SpringApplication.run(ConsumerApplication.class, args);}
}

2.PublisherApplication.java

package com.sunxiansheng.publisher;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplication
public class PublisherApplication {public static void main(String[] args) {SpringApplication.run(PublisherApplication.class);}
}

4.测试MQ

1.application.yml mq的最基本配置

spring:# RabbitMQ 配置rabbitmq:# 服务器地址host: ip# 用户名username: sunxiansheng# 密码password: rabbitmq# 虚拟主机virtual-host: /# 端口port: 5672

2.consumer

1.TestConfig.java MQ配置

package com.sunxiansheng.consumer.config;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** Description: 最基本的MQ测试* @Author sun* @Create 2024/8/2 14:34* @Version 1.0*/
@Configuration
public class TestConfig {/*** 创建一个fanout类型的交换机* @return*/@Beanpublic FanoutExchange fanoutExchange() {return new FanoutExchange("fanout.exchange.test");}/*** 创建一个队列* @return*/@Beanpublic Queue fanoutQueueTest() {return new Queue("fanout.queue.test");}/*** 交换机和队列绑定*/@Beanpublic Binding binding() {return BindingBuilder.bind(fanoutQueueTest()).to(fanoutExchange());}}

2.TestConfigListener.java 监听队列

package com.sunxiansheng.consumer.listener;import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;/*** Description: 最基本的MQ测试* @Author sun* @Create 2024/8/2 14:34* @Version 1.0*/
@Component
public class TestConfigListener {@RabbitListener(queues = "fanout.queue.test")public void receive(String message) {System.out.println("接收到的消息：" + message);}}

3.publisher

1.TestConfig.java 测试(注意指定启动类)

package com.sunxiansheng.consumer.config;import com.sunxiansheng.publisher.PublisherApplication;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.boot.test.context.SpringBootTest;import javax.annotation.Resource;/*** Description: 最基本的MQ测试* @Author sun* @Create 2024/8/2 14:34* @Version 1.0*/
@SpringBootTest(classes = PublisherApplication.class) // 指定启动类
public class TestConfig {@Resourceprivate AmqpTemplate amqpTemplate;@Testpublic void send() {// 交换机String exchange = "fanout.exchange.test";// 路由键String routingKey = "";// 消息String message = "hello fanout";// 发送消息amqpTemplate.convertAndSend(exchange, routingKey, message);}}

2.结果

2.基本交换机

1.Fanout

1.FanoutConfig.java 交换机配置

package com.sunxiansheng.consumer.config;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** Description: Fanout交换机* @Author sun* @Create 2024/7/29 15:06* @Version 1.0*/
@Configuration
public class FanoutConfig {@Beanpublic FanoutExchange fanoutExchange1() {// 创建一个fanout类型的交换机return new FanoutExchange("fanout.exchange");}@Beanpublic Queue fanoutQueue1() {// 创建一个队列return new Queue("fanout.queue1");}@Beanpublic Queue fanoutQueue2() {// 创建一个队列return new Queue("fanout.queue2");}// 两个队列绑定到交换机上@Beanpublic Binding bindingFanoutQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange1) {return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange1);}@Beanpublic Binding bindingFanoutQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange1) {return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange1);}
}

2.FanoutConfigListener.java 监听者

package com.sunxiansheng.consumer.listener;import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;/*** Description: Fanout交换机* @Author sun* @Create 2024/7/29 15:06* @Version 1.0*/
@Component
public class FanoutConfigListener {@RabbitListener(queues = "fanout.queue1")public void receive1(String message) {System.out.println("fanout.queue1接收到的消息：" + message);}@RabbitListener(queues = "fanout.queue2")public void receive2(String message) {System.out.println("fanout.queue2接收到的消息：" + message);}}

3.FanoutConfig.java 生产者

package com.sunxiansheng.consumer.config;import com.sunxiansheng.publisher.PublisherApplication;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.boot.test.context.SpringBootTest;import javax.annotation.Resource;/*** Description: Fanout交换机* @Author sun* @Create 2024/7/29 15:06* @Version 1.0*/
@SpringBootTest(classes = PublisherApplication.class) // 指定启动类
public class FanoutConfig {@Resourceprivate AmqpTemplate amqpTemplate;@Testpublic void send() {// 交换机String exchange = "fanout.exchange";// 路由键String routingKey = "";// 消息String message = "hello fanout";// 发送消息amqpTemplate.convertAndSend(exchange, routingKey, message);}}

2.Direct

1.DirectConfig.java 交换机配置

package com.sunxiansheng.consumer.config;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** Description: Direct交换机* @Author sun* @Create 2024/7/29 15:06* @Version 1.0*/
@Configuration
public class DirectConfig {@Beanpublic DirectExchange directExchange() {// 创建一个direct类型的交换机return new DirectExchange("direct.exchange");}@Beanpublic Queue directQueue1() {// 创建一个队列return new Queue("direct.queue1");}@Beanpublic Queue directQueue2() {// 创建一个队列return new Queue("direct.queue2");}// 两个队列绑定到交换机上，这里需要指定routingKey@Beanpublic Binding bindingDirectQueue1(Queue directQueue1, DirectExchange directExchange) {return BindingBuilder.bind(directQueue1).to(directExchange).with("black");}@Beanpublic Binding bindingDirectQueue2(Queue directQueue2, DirectExchange directExchange) {return BindingBuilder.bind(directQueue2).to(directExchange).with("green");}}

2.DirectConfigListener.java 监听者

package com.sunxiansheng.consumer.listener;import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;/*** Description: Direct交换机* @Author sun* @Create 2024/7/29 15:06* @Version 1.0*/
@Component
public class DirectConfigListener {@RabbitListener(queues = "direct.queue1")public void receive1(String message) {System.out.println("direct.queue1接收到的消息：" + message);}@RabbitListener(queues = "direct.queue2")public void receive2(String message) {System.out.println("direct.queue2接收到的消息：" + message);}}

3.DirectConfig.java 生产者

package com.sunxiansheng.consumer.config;import com.sunxiansheng.publisher.PublisherApplication;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.boot.test.context.SpringBootTest;import javax.annotation.Resource;/*** Description: Direct交换机* @Author sun* @Create 2024/7/29 15:06* @Version 1.0*/
@SpringBootTest(classes = PublisherApplication.class) // 指定启动类
public class DirectConfig {@Resourceprivate AmqpTemplate amqpTemplate;@Testpublic void send() {// 交换机String exchange = "direct.exchange";// 路由键String routingKey = "black";// 消息String message = "hello direct";// 发送消息amqpTemplate.convertAndSend(exchange, routingKey, message);}}