SpringBoot 整合 RabbitMQ
概念
2007 年发布，是一个在 AMQP(高级消息队列协议)基础上完成的，可复用的企业消息系统，是当前最主流的消息中间件之一。
RabbitMQ是一个由erlang开发的AMQP（Advanced Message Queue 高级消息队列协议 ）的开源实现，由于erlang 语言的高并发特性，性能较好，本质是个队列，FIFO 先入先出，里面存放的内容是message。
RabbitMQ 是一个消息中间件：它接收消息并且转发，就类似于一个快递站，卖家把快递通过快递站，送到我们的手上，MQ也是这样，接收并存储消息，再转发。
MQ典型应用场景：
- 异步处理。把消息放入消息中间件中，等到需要的时候再去处理。
- 流量削峰。例如秒杀活动，在短时间内访问量急剧增加，使用消息队列，当消息队列满了就拒绝响应，跳转到错误页面，这样就可以使得系统不会因为超负载而崩溃。
- 日志处理
- 应用解耦。假设某个服务A需要给许多个服务（B、C、D）发送消息，当某个服务（例如B）不需要发送消息了，服务A需要改代码再次部署；当新加入一个服务（服务E）需要服务A的消息的时候，也需要改代码重新部署；另外服务A也要考虑其他服务挂掉，没有收到消息怎么办？要不要重新发送呢？是不是很麻烦，使用MQ发布订阅模式，服务A只生产消息发送到MQ，B、C、D从MQ中读取消息，需要A的消息就订阅，不需要了就取消订阅，服务A不再操心其他的事情，使用这种方式可以降低服务或者系统之间的耦合。
首先先介绍一个简单的一个消息推送到接收的流程，提供一个简单的图：
黄色的圈圈就是我们的消息推送服务，将消息推送到 中间方框里面也就是 rabbitMq的服务器，然后经过服务器里面的交换机、队列等各种关系（后面会详细讲）将数据处理入列后，最终右边的蓝色圈圈消费者获取对应监听的消息。
常用的交换机有以下三种，因为消费者是从队列获取信息的，队列是绑定交换机的（一般），所以对应的消息推送/接收模式也会有以下几种：
Direct Exchange
直连型交换机，根据消息携带的路由键将消息投递给对应队列。
大致流程，有一个队列绑定到一个直连交换机上，同时赋予一个路由键 routing key 。
然后当一个消息携带着路由值为X，这个消息通过生产者发送给交换机时，交换机就会根据这个路由值X去寻找绑定值也是X的队列。
Fanout Exchange
广播式交换机，这个交换机没有路由键概念，就算你绑了路由键也是无视的。 这个交换机在接收到消息后，会直接转发到绑定到它上面的所有队列。
Topic Exchange
主题交换机，这个交换机其实跟直连交换机流程差不多，但是它的特点就是在它的路由键和绑定键之间是有规则的。
其路由规则为：routingkey必须为单词列表，单词之间以点号分隔开，*号代表一个单词，#号可以替代零个或多个单词。
举例：
路由*.*.key1绑定队列Q1，如：red.blue.key1、black.green.key1等，通过这些路由都可以发送消息给队列Q1
路由#.key2.*绑定队列Q2，如：a.b.c.key2.red、orange.key2.blue、key2.red等，通过这些路由都可以发送消息给队列Q2
路由key3.#绑定队列Q2，如：key3.red.blue、key3、key3.green、key3.black.blue.red等，通过这些路由都可以将消息发送给Q2
主流的消息队列对比
| 特性 | ActiveMQ | RabbitMQ | RocketMQ | Kafka |
|---|---|---|---|---|
| 单机吞吐量 | 万级，比 RocketMQ、Kafka 低一个数量级 | 同 ActiveMQ | 10 万级，支撑高吞吐 | 10 万级，高吞吐，一般配合大数据类的系统来进行实时数据计算、日志采集等场景 |
| topic 数量对吞吐量的影响 | topic 可以达到几百/几千的级别，吞吐量会有较小幅度的下降，这是 RocketMQ 的一大优势，在同等机器下，可以支撑大量的 topic | topic 从几十到几百个时候，吞吐量会大幅度下降，在同等机器下，Kafka 尽量保证 topic 数量不要过多，如果要支撑大规模的 topic，需要增加更多的机器资源 | ||
| 时效性 | ms 级 | 微秒级，这是 RabbitMQ 的一大特点，延迟最低 | ms 级 | 延迟在 ms 级以内 |
| 可用性 | 高，基于主从架构实现高可用 | 同 ActiveMQ | 非常高，分布式架构 | 非常高，分布式，一个数据多个副本，少数机器宕机，不会丢失数据，不会导致不可用 |
| 消息可靠性 | 有较低的概率丢失数据 | 基本不丢 | 经过参数优化配置，可以做到 0 丢失 | 同 RocketMQ |
| 功能支持 | MQ 领域的功能极其完备 | 基于 erlang 开发，并发能力很强，性能极好，延时很低 | MQ 功能较为完善，还是分布式的，扩展性好 | 功能较为简单，主要支持简单的 MQ 功能，在大数据领域的实时计算以及日志采集被大规模使用 |
SpringBoot整合RabbitMQ
RabbitMQ是MQ产品中的目前较为流行的产品之一，它遵从AMQP协议。RabbitMQ的底层实现语言使用的是Erlang，所以安装RabbitMQ需要先安装Erlang。
Erlang安装
windows版安装包下载地址：https😕/www.erlang.org/downloads
下载完毕后得到exe安装文件，一键傻瓜式安装，安装完毕需要重启，需要重启，需要重启。
安装的过程中可能会出现依赖Windows组件的提示，根据提示下载安装即可，都是自动执行的，如下：
Erlang安装后需要配置环境变量，否则RabbitMQ将无法找到安装的Erlang。需要配置项如下，作用等同JDK配置环境变量的作用。
- ERLANG_HOME
- PATH
安装 rabbitMQ
csdn 安装教程：https://blog.csdn.net/tirster/article/details/121938987
windows版安装包下载地址：https://rabbitmq.com/install-windows.html
下载完毕后得到exe安装文件，一键傻瓜式安装，安装完毕后会得到如下文件
启动服务器
rabbitmq-service.bat start # 启动服务
rabbitmq-service.bat stop # 停止服务
rabbitmqctl status # 查看服务状态
运行sbin目录下的rabbitmq-service.bat命令即可，start参数表示启动，stop参数表示退出，默认对外服务端口15672。
注意：启动rabbitmq的过程实际上是开启rabbitmq对应的系统服务，需要管理员权限方可执行。
说明：有没有感觉15672的服务端口很熟悉？activemq与rabbitmq有一个端口冲突问题，学习阶段无论操作哪一个？请确保另一个处于关闭状态。
说明：不喜欢命令行的小伙伴可以使用任务管理器中的服务页，找到RabbitMQ服务，使用鼠标右键菜单控制服务的启停。
访问web管理服务
RabbitMQ也提供有web控制台服务，但是此功能是一个插件，需要先启用才可以使用。
rabbitmq-plugins.bat list # 查看当前所有插件的运行状态
rabbitmq-plugins.bat enable rabbitmq_management # 启动rabbitmq_management插件
启动插件后可以在插件运行状态中查看是否运行，运行后通过浏览器即可打开服务后台管理界面
http://localhost:15672
web管理服务默认端口15672，访问后可以打开RabbitMQ的管理界面，如下：
[图片加载失败]
首先输入访问用户名和密码，初始化用户名和密码相同，均为：guest，成功登录后进入管理后台界面，如下：
[图片加载失败]
整合(direct模型)
RabbitMQ满足AMQP协议，因此不同的消息模型对应的制作不同，先使用最简单的direct模型开发。
步骤①：导入springboot整合amqp的starter，amqp协议默认实现为rabbitmq方案
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
步骤②：配置RabbitMQ的服务器地址
spring:rabbitmq:host: localhostport: 5672
步骤③：初始化直连模式系统设置
由于RabbitMQ不同模型要使用不同的交换机，因此需要先初始化RabbitMQ相关的对象，例如队列，交换机等
@Configuration
public class RabbitConfigDirect {// 声明队列@Beanpublic Queue directQueue(){return new Queue("direct_queue");}@Beanpublic Queue directQueue2(){return new Queue("direct_queue2");}// 声明交换机@Beanpublic DirectExchange directExchange(){return new DirectExchange("directExchange");}// 声明交换机和队列的绑定关系@Beanpublic Binding bindingDirect(){return BindingBuilder.bind(directQueue()).to(directExchange()).with("direct");}@Beanpublic Binding bindingDirect2(){return BindingBuilder.bind(directQueue2()).to(directExchange()).with("direct2");}
}
队列Queue与直连交换机DirectExchange创建后，还需要绑定他们之间的关系Binding，这样就可以通过交换机操作对应队列。
步骤④：使用AmqpTemplate操作RabbitMQ
@Service
public class MessageServiceRabbitmqDirectImpl implements MessageService {@Autowiredprivate AmqpTemplate amqpTemplate;@Overridepublic void sendMessage(String id) {System.out.println("待发送短信的订单已纳入处理队列（rabbitmq direct），id："+id);amqpTemplate.convertAndSend("directExchange","direct",id);}
}
amqp协议中的操作API接口名称看上去和jms规范的操作API接口很相似，但是传递参数差异很大。
步骤⑤：使用消息监听器在服务器启动后，监听指定位置，当消息出现后，立即消费消息
@Component
public class MessageListener {@RabbitListener(queues = "direct_queue")public void receive(String id){System.out.println("已完成短信发送业务(rabbitmq direct)，id："+id);}
}
使用注解@RabbitListener定义当前方法监听RabbitMQ中指定名称的消息队列。
整合(topic模型)
步骤①：同上
步骤②：同上
步骤③：初始化主题模式系统设置
@Configuration
public class RabbitConfigTopic {@Beanpublic Queue topicQueue(){return new Queue("topic_queue");}@Beanpublic Queue topicQueue2(){return new Queue("topic_queue2");}@Beanpublic TopicExchange topicExchange(){return new TopicExchange("topicExchange");}@Beanpublic Binding bindingTopic(){return BindingBuilder.bind(topicQueue()).to(topicExchange()).with("topic.*.id");}@Beanpublic Binding bindingTopic2(){return BindingBuilder.bind(topicQueue2()).to(topicExchange()).with("topic.orders.*");}
}
主题模式支持routingKey匹配模式，*表示匹配一个单词，#表示匹配任意内容，这样就可以通过主题交换机将消息分发到不同的队列中，详细内容请参看RabbitMQ系列课程。
| 匹配键 | topic.*.* | topic.# |
|---|---|---|
| topic.order.id | true | true |
| order.topic.id | false | false |
| topic.sm.order.id | false | true |
| topic.sm.id | false | true |
| topic.id.order | true | true |
| topic.id | false | true |
| topic.order | false | true |
步骤④：使用AmqpTemplate操作RabbitMQ
@Service
public class MessageServiceRabbitmqTopicImpl implements MessageService {@Autowiredprivate AmqpTemplate amqpTemplate;@Overridepublic void sendMessage(String id) {System.out.println("待发送短信的订单已纳入处理队列（rabbitmq topic），id："+id);amqpTemplate.convertAndSend("topicExchange","topic.orders.id",id);}
}
发送消息后，根据当前提供的routingKey与绑定交换机时设定的routingKey进行匹配，规则匹配成功消息才会进入到对应的队列中。
步骤⑤：使用消息监听器在服务器启动后，监听指定队列
@Component
public class MessageListener {@RabbitListener(queues = "topic_queue")public void receive(String id){System.out.println("已完成短信发送业务(rabbitmq topic 1)，id："+id);}@RabbitListener(queues = "topic_queue2")public void receive2(String id){System.out.println("已完成短信发送业务(rabbitmq topic 22222222)，id："+id);}
}
使用注解@RabbitListener定义当前方法监听RabbitMQ中指定名称的消息队列。
总结
- springboot整合RabbitMQ提供了AmqpTemplate对象作为客户端操作消息队列
- 操作ActiveMQ需要配置ActiveMQ服务器地址，默认端口5672
- 企业开发时通常使用监听器来处理消息队列中的消息，设置监听器使用注解@RabbitListener
- RabbitMQ有5种消息模型，使用的队列相同，但是交换机不同。交换机不同，对应的消息进入的策略也不同
