SpringBoot整合Rabbitmq:超详细(包括docker配置Rabbitmq的详细过程)
一、什么是mq
MQ(message queue)，从字面意思上看就个 FIFO 先入先出的队列，只不过队列中存放的内容是 message 而已，它是一种具有接收数据、存储数据、发送数据等功能的技术服务。
在互联网架构中，MQ 是一种非常常见的上下游“逻辑解耦+物理解耦”的消息通信服务，用于上下游传递消息。使用了 MQ 之后，消息发送上游只需要依赖 MQ，不用依赖其他服务
常见的MQ消息中间件有很多，例如ActiveMQ、RabbitMQ、Kafka、RocketMQ等等。
二、什么是Rabbitmq
2007 年发布，是一个在AMQP(高级消息队列协议)基础上完成的，可复用的企业消息系统，是当前最主流的消息中间件之一。
-
优点：由于 erlang 语言的高并发特性，性能较好；吞吐量到万级，MQ 功能比较完备、健壮、稳定、易用、跨平台、支持多种语言如Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持 AJAX 文档齐全；开源提供的管理界面非常棒，用起来很好用，社区活跃度高；更新频率相当高。
-
缺点：商业版需要收费，学习成本较高。
-
选用场景：结合 erlang 语言本身的并发优势，性能好时效性微秒级，社区活跃度也比较高，管理界面用起来十分方便，如果你的数据量没有那么大，中小型公司优先选择功能比较完备的 RabbitMQ。
三、在linux系统上用docker部署Rabbitmq详细步骤
-
前提条件
确保你的系统已经安装了 Docker。如果尚未安装，可以参考 如何部署Docker 使用 Docker 部署 RabbitMQ 的详细指南_docker部署rabbitmq-CSDN博客
-
拉取 RabbitMQ 镜像
从 Docker Hub 拉取 RabbitMQ 的镜像，推荐使用带有管理界面的版本：
docker pull rabbitmq:management -
创建挂载目录
创建一个目录用于持久化 RabbitMQ 的数据，避免容器重启后数据丢失：
mkdir -p /usr/local/docker/rabbitmq -
启动 RabbitMQ 容器
运行以下命令启动 RabbitMQ 容器：
docker run -d --name rabbitmq -v /usr/local/docker/rabbitmq:/var/lib/rabbitmq -p 5672:5672 -p 15672:15672 -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin rabbitmq:management-
-d：后台运行容器。 -
--name rabbitmq：为容器指定名称。 -
-v：将宿主机的目录挂载到容器内，用于数据持久化。 -
-p：映射容器端口到宿主机端口，5672用于消息通信，15672用于管理界面。 -
-e：设置默认的用户名和密码
-
-
访问管理界面
打开浏览器，访问
http://localhost:15672，使用默认用户名和密码（guest/guest）登录,出现以下界面则启动成功
**提示：RabbitMQ是采用 Erlang语言开发的，所以系统环境必须提供 Erlang环境，需要是安装 Erlang，但是如果是通过docker拉取的RabbitMQ，RabbitMQ 的官方 Docker 镜像已经预装了 Erlang。当你从 Docker Hub 拉取并运行 RabbitMQ 镜像时，Erlang 环境已经内置在镜像中，无需额外安装。**
此时，你的RabbitMQ便配置成功了
四、RabbitMQ管控台使用
参考文章：RabbitMQ的管控台的使用说明-腾讯云开发者社区-腾讯云
五、RabbitMQ四大核心概念
-
生产者：产生数据发送消息的程序是生产者。
-
交换机：交换机是 RabbitMQ 非常重要的一个部件，一方面它接收来自生产者的消息，另一方面它将消息推送到队列中。交换机必须确切知道如何处理它接收到的消息，是将这些消息推送到特定队列还是推送到多个队列，亦或者是把消息丢弃，这个是由交换机类型决定的。
-
队列：队列是 RabbitMQ 内部使用的一种数据结构，尽管消息流经 RabbitMQ 和应用程序，但它们只能存储在队列中。队列仅受主机的内存和磁盘限制的约束，本质上是一个大的消息缓冲区。许多生产者可以将消息发送到一个队列，许多消费者可以尝试从一个队列接收数据。
-
消费者：消费与接收具有相似的含义。消费者大多时候是一个等待接收消息的程序。请注意生产者，消费者和消息中间件很多时候并不在同一机器上。同一个应用程序既可以是生产者又是可以是消费者。
六、springboot简单整合RabbitMQ
-
在pom.xml文件中导入RabbitMQ依赖
org.springframework.boot spring-boot-starter-amqp -
在application.yml配置RabbitMQ
spring: rabbitmq: #rabbitmq的ip host: 你的rabbit配置的linuxip username: guest password: guest #虚拟主机 stream: virtual-host: / port: 5672 listener: simple: #消费者的最小数量 concurrency: 10 #消费者的最大数量 max-concurrency: 10 #限制消费者，每次只能处理一条消息，处理完才能继续下一条消息 prefetch: 1 #启动时，是否默认启动容器，默认true auto-startup: true #被拒绝后，重新进入队列 default-requeue-rejected: true template: retry: #启用重试机制，默认为false enabled: true #设置初始化的重试时间间隔 initial-interval: 1000ms #重试最大次数，默认是3 max-attempts: 3 #重试最大时间间隔，默认是10s max-interval: 10000ms #重试时间间隔的乘数 multiplier: 1 -
创建RibbitMQ的配置类 RabbitMQConfig.java
@Configuration public class RabbitMQConfig { private static final String QUEUE = "queue"; //创建队列 //true:表示持久化 //队列在默认情况下放到内存，rabbitmq重启后就丢失了，如果希望重启后，队列 //数据还能使用，就需要持久化 @Bean public Queue queue1(){ return new Queue(QUEUE,true); } } -
创建消息的生产者和消费者
-
创建生产者MQSender
@Slf4j @Service public class MQSender { //装配RabbitTemplate ->操作RabbitMQ @Resource private RabbitTemplate rabbitTemplate; //方法:发送消息 public void send(Object msg){ log.info("发送消息-->" + msg); //指定你队列的名字 rabbitTemplate.convertAndSend("queue",msg); } } -
创建消费者MQReceiver
@Service @Slf4j public class MQReceiver { //方法：接收消息 @RabbitListener(queues = "queue") public void receive(Object msg){ log.info("接收到的消息-->" + msg); } }
5、创建controller层调用消息生产者
@RequestMapping("/rabbit")
@RestController
public class RabbitMQConteroller {
@Resource
private MQSender mqSender;
@GetMapping("/mq")
//方法：调用消息生产者，发送消息
public void mq(){
mqSender.send("hello,消费者~");
}
}此时，看到控制台输出
表示你已经成功整合了RabbitMQ
七、RabbitMQ交换机介绍
-
RabbitMQ 消息传递模型的核心思想是：生产者生产的消息从不会直接发送到队列。实际上，通常生产者甚至都不知道这些消息传递传递到了哪些队列中。
-
相反，生产者只能将消息发送到交换机(exchange)，交换机工作的内容非常简单，一方面它接收来自生产者的消息，另一方面将它们推入队列。交换机必须确切知道如何处理收到的消息。是应该把这些消息放到特定队列还是说把他们到许多队列中还是说应该丢弃它们。这就的由交换机的类型来决定。
-
交换机的类型有四种：直接(direct)、主题(topic)、标题(headers)、扇出(fanout)，每一个都有，以下我们对四种交换机根据简单例子介绍
八、RabbitMQ交换机的在springboot中使用的简单例子
fanout模式
-
fanout就是广播模式，就是把交换机（Exchange）里的消息队列发送给所有绑定该交换机的队列，忽略routingKey（也就是路由)
-
示意图
-
生产者把消费发送给指定的交换机
-
再把交换机的消息发送给所有绑定该交换机的队列，忽略路由
代码示例：
-
创建--fanout--广播模式队列和交换机示例,在RibbitMQConfig中配置示例
//--fanout--广播模式队列示例
private static final String QUEUE1 = "queue1_fanout01";
private static final String QUEUE2 = "queue2_fanout02";
private static final String EXCHANGE = "fanoutExchange";
//创建队列
//true:表示持久化
//队列在默认情况下放到内存，rabbitmq重启后就丢失了，如果希望重启后，队列
//数据还能使用，就需要持久化
@Bean
public Queue queue1(){
return new Queue(QUEUE1,true);
}
@Bean
public Queue queue2(){
return new Queue(QUEUE2,true);
}
//创建Fanout交换机
@Bean
public FanoutExchange exchange(){
return new FanoutExchange(EXCHANGE);
}
//将queue1_fanout01,2队列绑定到交换机EXCHANGE，此时如果消息发送者发送消息到交换机，因为
//是广播模式，与其交换机绑定的所有队列都可以接收到消息
@Bean
public Binding binding01(){
return BindingBuilder.bind(queue1()).to(exchange());
}
@Bean
public Binding binding02(){
return BindingBuilder.bind(queue2()).to(exchange());
}2.在MQSender消息生产者配置方法(消息生产者和消费者都是服务提供者，所以记得加@Service注解)
//方法：发送消息到交换机
//这里的空串是因为fanout要屏蔽路由
public void sendFanout(Object msg){
log.info("发送消息->" + msg);
rabbitTemplate.convertAndSend("fanoutExchange","",msg);
}
3.在MQReceiver消息消费者接收消息
@RabbitListener(queues = "queue1_fanout01") //表示监听的队列是哪个
public void receive1(Object msg){
log.info("queue1_fanout01接收到的消息-->" + msg);
}
@RabbitListener(queues = "queue2_fanout02")
public void receive2(Object msg){
log.info("queue2_fanout02接收到的消息-->" + msg);
}
4.在controller中调用消息生产者方法
//调用生产者，发送消息到交换机
@GetMapping("/mq/fanout")
public void fanout(){
mqSender.sendFanout("hello,交换机");
}5.观看控制台输出如下表示成功
direct模式
-
direct就是路由模式，路由模式是在使用交换机的同时，生产者指定路由发送数据，消费者绑定路由接收数据。
-
与广播模式不同的是，广播模式只要是绑定了交换机的队列都会收到生产者向交换机推送过来的数据，而路由模式下加了一个路由设置，生产者向交换机发送数据时，会声明发送到交换机下的哪个路由，并且只有当消费者的队列绑定了交换机并且声明了路由，才会收到数据
-
示意图
代码示例
-
创建direct路由模式示例
//--direct--路由模式示例 private static final String QUEUE3 = "queue_direct01"; private static final String QUEUE4 = "queue_direct02"; private static final String EXCHANGE2 = "directExchange"; //路由模式，定义路由 private static final String ROUTING_KEY01 = "queue.red"; private static final String ROUTING_KEY02 = "queue.green"; //direct----- @Bean public Queue queue3(){ return new Queue(QUEUE3,true); } @Bean public Queue queue4(){ return new Queue(QUEUE4,true); } //配置direct交换机 @Bean public DirectExchange exchange_direct(){ return new DirectExchange(EXCHANGE2); } //direct绑定交换机 //.with：指定路由 @Bean public Binding binding_direct1(){ return BindingBuilder.bind(queue3()).to(exchange_direct()).with(ROUTING_KEY01); } @Bean public Binding binding_direct2(){ return BindingBuilder.bind(queue4()).to(exchange_direct()).with(ROUTING_KEY02); } -
在消息生产者配置发送消息方法
//方法：发送消息到direct交换机，并且指定路由 //这时我们如果调用sendDirect1方法，会将消息发送到交换机的queue.red路由，在配置文件中queue3绑定了 //路由queue.red，所有只有与queue3队列绑定的消费者才能接收到消息 public void sendDirect1(Object msg){ log.info("发送消息->" + msg); rabbitTemplate.convertAndSend("directExchange","queue.red",msg); } public void sendDirect2(Object msg){ log.info("发送消息->" + msg); rabbitTemplate.convertAndSend("directExchange","queue.green",msg); } -
配置消息接收者
@RabbitListener(queues = "queue_direct01") public void queue_direct01(Object msg){ log.info("queue_direct01接收到的消息-->" + msg); } @RabbitListener(queues = "queue_direct02") public void queue_direct02(Object msg){ log.info("queue_direct02接收到的消息-->" + msg); } -
配置controller
//调用生产者，发送消息到交换机
@GetMapping("/mq/direct1")
public void direct1(){
mqSender.sendDirect1("hello,jack");
}
@GetMapping("/mq/direct2")
public void direct2(){
mqSender.sendDirect2("hello,tom");
}5.此时，调用direct1控制台输出
调用direct2输出
topic模式
-
direct模式会造成路由RoutingKey堆积太多，而实际开发往往是按照某个规则来进行路由匹配的，RabbitMQ提供了Topic模式来适应这种需求
-
Topic模式是direct模式上的一种扩展叠加，扩展叠加了模糊路由RoutingKey的模式，可以理解为是模糊的路由匹配模式
-
*（星号）：可以（只能匹配到一个单词）
-
#（井号）：可以匹配多个单词
-
这里是指一个完整的英语单词，而不是一个字母
-
-
示意图：
代码示例：
-
配置topic路由模式示例
package com.yw.config; import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.Queue; import org.springframework.amqp.core.TopicExchange; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class RabbitMQTopicConfig { private static final String QUEUE1 = "queue_topic01"; private static final String QUEUE2 = "queue_topic02"; private static final String EXCHANGE = "topicExchange"; private static final String KEY01 = "#.queue.#"; private static final String KEY02 = "*.queue.#"; @Bean public Queue topic_queue1(){ return new Queue(QUEUE1); } @Bean public Queue topic_queue2(){ return new Queue(QUEUE2); } @Bean public TopicExchange topicExchange(){ return new TopicExchange(EXCHANGE); } @Bean public Binding binding_topic1(){ return BindingBuilder.bind(topic_queue1()).to(topicExchange()).with(KEY01); } @Bean public Binding binding_topic2(){ return BindingBuilder.bind(topic_queue2()).to(topicExchange()).with(KEY02); } } -
配置消息生产者方
//方法：发送消息到topic交换机，并且指定路由 //此时调用sendTopic1根据路由KEY02前面为星号，所以此路由的队列不能接收到消息 public void sendTopic1(Object msg){ log.info("发送消息->" + msg); rabbitTemplate.convertAndSend("topicExchange","hello.hello.queue.green",msg); } //两个队列都满足路由，都可以接收消息 public void sendTopic2(Object msg){ log.info("发送消息->" + msg); rabbitTemplate.convertAndSend("topicExchange","hello.queue.green",msg); } -
配置消息消费者方法
@RabbitListener(queues = "queue_topic01") public void queue_topic01(Object msg){ log.info("queue_topic01接收到的消息-->" + msg); } @RabbitListener(queues = "queue_topic02") public void queue_topic02(Object msg){ log.info("queue_topic02接收到的消息-->" + msg); } -
配置controller
@GetMapping("/mq/topic1") public void topic1(){ mqSender.sendTopic1("hello,tom"); } @GetMapping("/mq/topic2") public void topic2(){ mqSender.sendTopic2("hello,tom"); } -
调用topic1控制台输出
-
调用topic2控制台输出
headers模式
-
headers交换机是一种比较复杂且少见的交换机，不同于direct和topic,它不关心路由key是否匹配，而只关心header中的key-value是否匹配，有点类似与http里面的请求头
-
headers头路由模型中，消息是根据prop，即请求头中的k-v来匹配的
-
绑定的队列指定的headers中必须包含一个“x-match"的键
-
“x-match"的键有两个：all和any
-
all:表示绑定对队列/消费方 指定的所有的k-v都必须在消息header中出现并匹配
-
any:表示绑定对队列/消费方 指定的的k-v至少有一个在消息header中出现并匹配
-
代码示例
-
headers模式的示例
package com.yw.config; import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.HeadersExchange; import org.springframework.amqp.core.Queue; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import java.util.HashMap; import java.util.Map; @Configuration public class RabbitMQHeadersConfig { private static final String QUEUE1 = "queue_headers01"; private static final String QUEUE2 = "queue_headers02"; private static final String EXCHANGE = "headersExchange"; //创建队列 @Bean public Queue headers_queue1() { return new Queue(QUEUE1); } @Bean public Queue headers_queue2() { return new Queue(QUEUE2); } //创建交换机 @Bean public HeadersExchange headersExchange(){ return new HeadersExchange(EXCHANGE); } //创建连接 @Bean public Binding headersBinding01(){ //先定义k-v因为可以有多个，所以将其放入map Map -
创建消息生产者方法
//方法：发送消息到headers交换机，并且指定需要匹配的k-v //调用sendHeader01，因为其setHeader了两个属性值，与配置的两个队列都匹配，所以两个队列都可以收到消息
public void sendHeader01(String msg){ log.info("发送消息->" + msg); //创建消息属性MessageProperties MessageProperties properties = new MessageProperties(); properties.setHeader("color","red"); properties.setHeader("speed","fast"); //创建message对象【包含了发送的消息本身和属性】 Message message = new Message(msg.getBytes(), properties); rabbitTemplate.convertAndSend("headersExchange","",message); } public void sendHeader02(String msg){ log.info("发送消息->" + msg); //创建消息属性MessageProperties MessageProperties properties = new MessageProperties(); properties.setHeader("color","red"); properties.setHeader("speed","normal"); //创建message对象【包含了发送的消息本身和属性】 Message message = new Message(msg.getBytes(), properties); rabbitTemplate.convertAndSend("headersExchange","",message); } -
创建消息消费者接收消息
@RabbitListener(queues = "queue_headers01") public void queue_headers01(Message msg){ log.info("queue_topic01接收到的消息对象-->" + msg); log.info("queue_topic01接收到的消息-->" + new String(msg.getBody())); } @RabbitListener(queues = "queue_headers02") public void queue_headers02(Message msg){ log.info("queue_topic02接收到的消息对象-->" + msg); log.info("queue_topic02接收到的消息-->" + new String(msg.getBody())); } -
配置controller
@GetMapping("/mq/headers1") public void headers1(){ mqSender.sendHeader01("hello,red"); } @GetMapping("/mq/headers2") public void headers2(){ mqSender.sendHeader02("hello,great"); } -
调用headers1方法控制台输出
-
调用headers2方法控制台输出
至此，文章结束。
