RabbitMQ学习笔记

什么是MQ？

Message Queue，消息队列，FIFO，队列中存储的是message，是一种跨进程的通信机制，用于上下游传递消息。在互联网架构中，是一种非常常见的”逻辑解耦 + 物理解耦”的消息通信服务。使用MQ，消息发送只需要依赖MQ，不需要依赖其他服务。

为什么要用MQ？

流量消峰，应用解耦，异步处理

MQ的分类：

• ActiveMQ，
• 优点：单机吞吐量万级，时效性ms级，可用性高，基于主从架构实现高可用性，消息可靠性较低的概率丢失数据
• 缺点：官方社区现在对ActiveMQ 5.x维护越来越少，高吞吐量场景较少使用
• Kafka
• 优点: 性能卓越，单机写入TPS约在百万条/秒，最大的优点，就是吞吐量高。时效性ms级可用性非常高，kafka是分布式的，一个数据多个副本，少数机器宕机，不会丢失数据，不会导致不可用。消费者采用Pull方式获取消息，消息有序，通过控制能够保证所有消息被消费且仅被消费一次；有优秀的第三方Kafka Web管理界面Kafka-Manager；
• 缺点：Kafka单机超过64个队列/分区，Load会发生明显的飙高现象，队列越多，load越高，发送消息响应时间变长，使用短轮询方式，实时性取决于轮询间隔时间，消费失败不支持重试；支持消息顺序，但是一台代理宕机后，就会产生消息乱序，社区更新较慢
• RocketMQ
• RocketMQ出自阿里巴巴的开源产品，用Java语言实现，在设计时参考了 Kafka，并做出了自己的一些改进
• 优点:单机吞吐量十万级,可用性非常高，分布式架构,消息可以做到0丢失,MQ功能较为完善，还是分布式的，扩展性好,支持10亿级别的消息堆积，不会因为堆积导致性能下降,源码是java我们可以自己阅读源码，定制自己公司的MQ
• 缺点：支持的客户端语言不多，目前是java及c++，其中c++不成熟；社区活跃度一般,没有在MQ核心中去实现JMS等接口,有些系统要迁移需要修改大量代码
• RabbitMQ
• 2007年发布，是一个在AMQP(Adviced Message Queue Protocol 高级消息队列协议)基础上完成的，可复用的企业消息系统，是当前最主流的消息中间件之一
• 优点:由于erlang语言的高并发特性，性能较好；吞吐量到万级，MQ功能比较完备,健壮、稳定、易用、跨平台、支持多种语言 如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX文档齐全；开源提供的管理界面非常棒，用起来很好用,社区活跃度高；更新频率相当高
• 缺点：商业版需要收费,学习成本较高

MQ的选择

Kafka主要特点是基于Pull的模式来处理消息消费，追求高吞吐量，一开始的目的就是用于日志收集和传输，适合产生大量数据的互联网服务的数据收集业务。大型公司建议可以选用，如果有日志采集功能，肯定是首选kafka

RocketMQ，天生为金融互联网领域而生，对于可靠性要求很高的场景，尤其是电商里面的订单扣款，以及业务削峰，在大量交易涌入时，后端可能无法及时处理的情况。RoketMQ在稳定性上可能更值得信赖，这些业务场景在阿里双11已经经历了多次考验，如果你的业务有上述并发场景，建议可以选择RocketMQ

RabbitMQ，结合erlang语言本身的并发优势，性能好时效性微秒级，社区活跃度也比较高，管理界面用起来十分方便，如果你的数据量没有那么大，中小型公司优先选择功能比较完备的RabbitMQ

RabbitMQ是一个消息中间件：它接受并转发消息。你可以把它当做一个快递站点，当你要发送一个包裹时，你把你的包裹放到快递站，快递员最终会把你的快递送到收件人那里，按照这种逻辑RabbitMQ是一个快递站，一个快递员帮你传递快件。RabbitMQ与快递站的主要区别在于，它不处理快件而是接收，存储和转发消息数据。

四大核心概念

• 生产者：产生数据发送消息的程序
• 交换机：是RabbitMQ非常重要的一个部件，一方面它接收来自生产者的消息，另一方面它将消息推送到队列中。交换机必须确切知道如何处理它接收到的消息，是将这些消息推送到特定队列还是推送到多个队列，亦或者是把消息丢弃，这个得有交换机类型决定
• 队列：是RabbitMQ内部使用的一种数据结构，尽管消息流经RabbitMQ和应用程序，但它们只能存储在队列中。队列仅受主机的内存和磁盘限制的约束，本质上是一个大的消息缓冲区。许多生产者可以将消息发送到一个队列，许多消费者可以尝试从一个队列接收数据。这就是我们使用队列的方式
• 消费者：消费与接收具有相似的含义。消费者大多时候是一个等待接收消息的程序。请注意生产者，消费者和消息中间件很多时候并不在同一机器上。同一个应用程序既可以是生产者又是可以是消费者

RabbitMQ核心部分

• Hello World
• Work queues
• Publish/Subscribe
• Routing
• Topics
• Publisher Confirms

名词说明

• Broker：接收和分发消息的应用，RabbitMQ Server就是Message Broker
• Virtual host：出于多租户和安全因素设计的，把 AMQP 的基本组件划分到一个虚拟的分组中，类似于网络中的namespace概念。当多个不同的用户使用同一个RabbitMQ server提供的服务时，可以划分出多个vhost，每个用户在自己的 vhost 创建 exchange／queue 等
• Connection：publisher／consumer和broker之间的TCP连接
• Channel：如果每一次访问 RabbitMQ 都建立一个Connection，在消息量大的时候建立 TCP Connection的开销将是巨大的，效率也较低。Channel是在connection内部建立的逻辑连接，如果应用程序支持多线程，通常每个thread创建单独的channel进行通讯，AMQP method包含了channel id 帮助客户端和message broker 识别 channel，所以channel之间是完全隔离的。Channel作为轻量级的Connection极大减少了操作系统建立TCP connection的开销
• Exchange：message 到达 broker 的第一站，根据分发规则，匹配查询表中的 routing key，分发消息到queue 中去。常用的类型有：direct (point-to-point), topic (publish-subscribe) and fanout (multicast)
• Queue：消息最终被送到这里等待consumer取走
• Binding：exchange和queue之间的虚拟连接，binding中可以包含routing key，Binding信息被保存到exchange中的查询表中，用于message的分发依据

安装步骤：

• 下载rabbitmq rpm：https://github.com/rabbitmq/rabbitmq-server/releases
• 下载erlang rpm（zero dependency erlang rpm for rabbitmq）：https://packagecloud.io/rabbitmq/erlang
• 依赖erlang版本关系：https://www.rabbitmq.com/which-erlang.html
• 安装其他依赖： yum install socat logrotate -y
• 安装rabbitmq： rpm -ivh rabbitmq-server-3.10.0-1.el7.noarch.rpm
• 启动服务：
• 设置开机启动： systemctl enable rabbitmq-server.service
• 启动服务： systemctl start rabbitmq-server.service
• 停止服务： systemctl stop rabbitmq-server.service
• 开启web管理插件： rabbitmq-plugins enable rabbitmq_management，之后访问 ip:15672，需要添加用户并赋予权限才能登陆
• 添加用户： rabbitmqctl add_user admin 999999
• 设置用户角色： rabbitmqctl set_user_tags admin administrator
• 设置用户权限： rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"
• 查看当前用户和角色： rabbitmqctl list_users
• 查看命令帮助： rabbitmqctl help，具体命令帮助： rabbitmqctl help xxx
• 关闭应用： rabbitmqctl stop_app
• 清除： rabbitmqctl reset
• 重启： rabbitmqctl start_app

Hello World

    com.rabbitmq
    amqp-client
    5.14.2

Work Queues

基本使用

一个channel发送消息到队列，多个channel消费队列的消息，消息轮询发送给多个channel

注意：获取多个channel消费时，rabbitmq有自己的线程池。一个消息只能发给一个channel

消息应答

• 自动应答：autoAck=true，接收到消息后自动向队列确认
• 手动应答：basicAck（肯定确认），basicNack（否定确认），basicReject（比basicNack少一个参数，直接拒绝可以丢弃）
• multiple参数：批量应答，手动应答的好处是可以批量应答，并且较少网络拥堵。批量应答会将当前channel的所有消息都应答
• 消息自动重写入队：还未向队列发送ack就断开了，那么该channel消费的消息会重新入队，被其他没有断开的channel消费

rabbitmq持久化

• 队列持久化： 生产者在声明队列时，设置durable为true，那么即时rabbitmq重启，也会有该队列，其他没有持久化的队列在重启后会删除。已经声明为非持久化的队列再声明为持久化的会报错
• 消息持久化：队列持久化只能保证队列不丢失，在 生产者发送消息到队列时还需要设置属性 消息持久化，即使消息持久化了，也有可能丢失

不公平分发

• 不公平分发：轮询分发是公平分发，在多个消费者消费能力不同时，轮询分发效率不高，可以采用不公平分发，消费能力强的多接收消息，需要在 所有消费者的channel上设置channel.basicQos(1)
• 预取值：channel.basicQos(prefetch)，表示每个channel缓存的最大没有ack的消息数量，还是轮询分发

发布确认

队列持久化 + 消息持久化也不能保证消息不丢失，当消息发送到队列而又没有保存到磁盘时，rabbitmq宕机，那么消息就会丢失。因此当生产者发送消息到队列，rabbitmq将数据存储到磁盘上之后向生产者确认，保证消息不丢失

发布确认默认是没有开启的，如果要开启需要调用channel.confirmSelect()

• 单个发布确认：简单的确认方式，发一条数据，同步确认发布成功，发布速度特别慢，最多提供每秒不超过百条的吞吐量
• 批量发布确认：发布一批消息后，再同步确认发布成功，提高吞吐量，缺点是当发生故障时不知道是哪个消息出现问题了，如果同步确认失败，发布了批量消息都被消费了吗？有重复消费的问题吗？
• 异步发布确认：生产者只管发布消息，消息确认或失败时调用callback
• 处理异步未确认消息：使用 ConcurrentSkipListMap<long, string></long,>，把未确认的消息放到能被发布线程访问的队列，这个队列在confirm callback与发布线程之间进行消息传递（共享消息，发布时添加消息，确认时删除消息）

对比：

• 单个确认：简单，吞吐量非常有限
• 批量发布：简单，合理的吞吐量，一旦出现问题很难推断是哪条消息出现了问题
• 异步确认：最佳性能和资源利用，在出现错误的情况下可以很好的控制，实现起来稍微复杂

之前都是将消息发送到队列（默认交换机，假设每个消息被一个消费者消费），如果这个队列被多个消费者消费，消费者之间是竞争关系，队列中的消息只能被消费者消费一次。如果要实现同一个消息同时被多个消费者消费（发布、订阅模式），需要使用交换机，为该交换机绑定多个队列，那么消息会同时发送到绑定的所有队列中，这样消费不同队列的消费者就可以消费同一个消息了

exchange概念

rabbitmq消息传递模型的核心思想是：生产者生产的消息不会直接发送给队列，而是只能发送到交换机，交换机的工作内容非常简单，一方面接收来自生产者的消息，另一方面将他们推入队列。交换机必须确切知道如何处理消息，是应该放到特定队列，还是放到许多队列，还是应该丢弃，由交换机的类型决定。

交换机的类型

直接（direct），主题（topic），标题（headers），扇出（fanout）

无名exchange：之前发送消息时交换机为空字符串，使用了默认交换，消息能路由发送到队列中是由routingKey决定的，之前写的队列名，所以消息发送到指定队列。如果使用了交换机，就不能写队列名称了，而是使用routingKey

临时队列

队列没有持久化，一旦消费者断开，队列被自动删除，创建临时队列 channel.queueDeclare().getQueue()，feature标识为 AD，Excl

绑定队列

定义一个交换机，可以为该交换机绑定队列，绑定时是通过routingKey绑定的，不同队列可以指定相同的routingKey，交换机的消息是根据routingKey确认发送到哪个队列

fanout exchange（发布/订阅）

这种类型的交换机非常简单，将接收到的消息 广播到所有已知的队列中，系统有一个默认的 amq.fanout交换机

消费时，交换机和队列必须都创建好了，否则报错。生产者声明临时队列后（durable=false，即使有名字也是临时队列），消费者连接后断开，那么该队列就会被删除。

direct exchange（直接交换机）

直接交换机在发送消息时需要指定routingKey，消息发送到指定的routingKey绑定的队列，而队列绑定哪些routingKey就消费哪个routingKey的消息，如果多个队列绑定了同一个routingKey，那么消息会发送到每一个绑定的队列

topic exchange（主题交换机）

发送到topic交换机的routingKey需要满足一定的格式，多个单词使用 .隔开，可以使用 *表示任意一个单词，使用 #表示零个或多个单词，在绑定队列时更灵活。

topic交换机中使用 #绑定时相当于fanout，可以接收所有消息。没有使用 .和 #绑定时相当于direct

死信队列

没有被消费的消息，可以放入死信队列。死信的来源：消息ttl过期，队列达到最大长度（先进先出），消息被拒绝且requeue=false

生产者和队列转发到死信队列时都可以设置ttl，参数 x-message-ttl

延迟队列

希望在指定时间到了之后或之前取出和处理，延迟队列就是用来存放需要在指定时间被处理元素的队列。

使用场景

• 订单30分钟内未支付自动取消
• 预定会议后，在预定时间点前10分钟通知参会人员

rabbitmq中的ttl

ttl是rabbitmq中的一个消息或一个队列的属性，表示一个消息或队列中的所有消息的最大存活时间。如果一个消息或一个队列设置了ttl属性，那么这条消息如果在ttl时间内没有被消费，则会称为死信，如果同时配置，则使用较小的值。

• 消息设置ttl：setExpiration(ttlTime)
• 队列设置ttl：args.put(“x-message-ttl”, ttlTime)
• 两者的区别：如果消息设置了ttl，那么消息一旦过期就会被队列丢弃（如果配置了死信交换机则转发给死信交换机），如果队列设置了ttl，即使过期，也不一定会被马上丢弃，因为消息是否过期是 投递给消费者之前判定的，如果队列有积压情况，即使过期也会存活一段时间。如果不设置ttl，则消息永远不会过期，如果设置为0，表示除非可以直接投递给消费者，否则消息将马上丢弃。

死信队列中，消息ttl过期的情况下，永远不使用正常情况下的队列消费，那么该消息总是在ttl时间后发送到死信交换机，对于生产者和死信交换机的消费者而言，刚好延迟了ttl时间。

当需要延迟的时间有很多时，不能通过无限制的增加队列来满足需求，设置一个普通队列，可以接收所有延迟时间的消息，在生产者发送消息时指定失效时间，但是有一个问题，如果有前后两个消息，后面的消息延迟时间比前一个消息延迟时间小，那么后面的消息依然需要等待前面的消息失效时才会将后面的消息发送到死信交换机（队列的先进先出特性，rabbitmq只会监听队列最前面的消息的ttl是否到期）

解决办法：使用rabbitmq的延迟插件，使用死信交换机，设置直接在交换机处监听延迟，达到延迟时间时直接发送到队列。

延迟交换机

将下载后的插件放在 /usr/lib/rabbitmq/lib/rabbitmq_server-3.10.0/plugins目录下

开启插件： rabbitmq-plugins enable rabbitmq_delayed_message_exchange

15672页面交换机页面在添加交换机时就可以看到多了一个 x-delayed-message类型的交换机

配置延迟交换机需要通过设置定制交换机，设置类型为 x-delayed-message，设置参数 Map.of("x-delayed-type", "direct")，表示延迟类型为direct

注意：在发送消息时，要设置发送消息的delay而不是expiration

延迟队列再需要延时处理的场景下非常有用，使用rabbitmq来实现延迟队列可以很好的利用rabbitmq的特性，如消息可靠发送、消息可靠投递、死信队列保障消息被消费一次以及未被正确处理的消息不会被丢弃。

另外，通过rabbitmq集群的特性可以很好的解决单点故障的问题，不会因为单个节点挂掉而导致延迟队列不可用或者消息丢失。

延迟队列还有很多其他选择，比如利用java的DelayQueue，Redis的zset，Quartz或者kafka时间轮，这些方式各有特点，需要根据场景选用。

在rabbitmq重启时，生产者投递的消息会丢失，交换机有问题或者队列有问题都会导致消息丢失

解决方法：

• 消息发送到交换机后，调用确认回调接口，通知生产者消息确认状态
• 交换机转发消息到队列时，如果消息没有无法转发到队列，调用消息回退接口，通知生产者消息消费状态

交换机接收消息确认回调

实现接口RabbitTemplate.ConfirmCallback，并将该实现类对象设置到rabbitTemplate，每一条消息是否被交换机接收都会调用该回调方法

交换机不可路由消息时回退消息给生产者

实现接口RabbitTemplate.ReturnCallback，并将该实现类对象设置到rabbitTemplate，当交换机的转发该消息到队列时，如果没有可以路由的队列，那么会调用消息回退回调方法

备份交换机

当交换机无法路由消息到队列时，如果该交换机配置了备份交换机，那么消息会转发给备份交换机，即使生产者配置了消息回退，那么此时因为消费被转发给了备份交换机，不会再进行消息回退。

使用备份交换机的原因：即使配置了交换机回调、消息回退，我们也并不知道如何处理这些消息，最多打印下日志，再来手动处理，通过日志处理无法路由的消息很不优雅，特别是当生产者有多台服务器的时候，而且消息回退会增加系统的复杂性，需要增加处理这些被退回消息的逻辑。如果既不想丢失消息，又不想增加生产者的复杂性，那么可以使用备份交换机。前面提到队列中没有被消费的消息可以通过死信队列处理，那么这里的消息都没有进入到队列，那么就可以使用备份交换机处理

备份交换机一般使用fanout类型，这样所有的队列都可以消费到消息，还可以建立一个报警队列，用独立的消费者来进行监测和报警。

消息没有被正确处理：队列中的消息可以配置死信交换机、死信队列。交换机无法路由的消息可以使用备份交换机。

幂等性

一次请求或多次请求的结果是一致的。

重复消费：消费者消费rabbitmq中的消息时，当已经把消息发送给了消费者，但是消费者的ack返回时因为网络中断导致mq未收到确认消息，该条消息会重新发送给其他消费者，或者在网络重连后再次发送给该消费者，但是实际上该消费者已经成功消费了该消息，造成消费者重复消费。

如果利用mq做支付，当支付后向mq发送ack时网络中断，那么mq会将消息重新发送给消费者，造成重复消费，用户重复支付。

解决思路：幂等性问题一般使用全局id，可以使用唯一标识符，比如时间戳或者uuid或者mq中的消息的id，或者按自己的规则生成一个全局唯一id，每次消费时用该id判断是否消费过

在海量订单生成的业务高峰期，生产端可能会重复发送消息，这时消费端就要实现幂等性，以保证消息永远不会被消费多次。主流的幂等性有两种操作：1，唯一id+指纹码机制，利用数据库主键去重，2，利用redis的原子性实现

第一种方式，实现简单，拼接id和指纹码后查询判断是否重复，劣势是高并发时，如果是单个数据库就会有写入瓶颈，当然也可以分库分表提升性能，这种方式不推荐。

第二种方式，利用redis执行setnx命令，天然具有幂等性，从而实现不重复消费。

优先级队列

可以为队列设置优先级，在发送消息的时候，设置消息的优先级，消息在到达队列时，会进行优先级排序，优先处理优先级高的消息，优先级最大值可为0-255，实际使用时不要设置太大，比较浪费cpu和内存。

惰性队列

正常情况下，队列中的消息会尽可能的存放在内存中，即使是持久化的消息，在被写入磁盘的同时也会在内存中驻留一份备份，当rabbitmq需要释放内存的时候，会将内存中的消息换页至磁盘，这个操作会耗费较长的时间，也会阻塞队列的操作，进而无法接收新的消息。

惰性队列的设计目标是支持更长的队列，即支持更长时间内的消息存储。当消费者因为各种各样的原因长时间无法消费数据造成消息堆积时，惰性队列就很有必要。

惰性队列，消息会先写入磁盘，如果需要消费，再从磁盘中取出。

队列有两种模式：default和lazy，lazy就是惰性模式，可以通过policy的方式设置，也可以声明队列时设置 x-queue-mode设置，policy具有更高的优先级。注意：如果要改变已有队列的模式，需要先删除队列，再声明一个新的。

内存开销对比：存储100万条消息，每条消息1kb，普通队列需要占用1.2GB内存，惰性队列占用1.5MB

clustering

拷贝主节点 /var/lib/rabbitmq/.erlang.cookie到其他所以要加入的节点

加入新节点到集群：

• 先关闭服务 systemctl stop rabbitmq-server.service，（可能是因为直接覆盖了.erlang.cookie导致的无法stop，直接停止服务）
• 启动rabbitmq的虚拟机 rabbitmq-server -detached
• 先停止 rabbitmqctl stop_app
• 再加入 rabbitmqctl join_cluster rabbit@localhost102，加入集群会隐式删除节点存在的资源和数据
• 启动节点 rabbitmqctl start_app
• 查看集群状态 rabbitmqctl cluster_status

从集群中删除节点localhost103：

• 先停止要删除的节点 rabbitmqctl -n rabbit@localhost103 stop_app，远程删除，也可以在被删除节点执行 rabbitmqctl stop_app
• 再在集群节点执行 rabbitmqctl forget_cluster_node rabbit@localhost103
• 此时103仍然认为是集群中的一个节点，启动时会报错，需要重置才能重新启动 rabbitmqctl reset， rabbitmqctl start_app，或者重新加入集群 rabbitmqctl join_cluster rabbit@localhost102再启动

总结：如果要加入集群，需要停止后加入，然后启动。如果用从集群中删除节点，需要先停止，在forget。被删除的节点如果要重写加入集群，重写join即可，如果要单独启动，需要先reset，再start，因为它被forget后仍然认为是集群的一个节点。

注意： rabbitmqctl stop_app只是停止rabbitmq服务， rabbitmqctl stop是关闭erlang虚拟机和rabbitmq服务，如果关闭了虚拟机，需要执行 rabbitmq-server -detached启动虚拟机

原来节点添加的用户也可以用来登录集群的其他节点

镜像队列

在某个节点上创建的队列，只能在该节点上访问，当该节点宕机时，队列不再可用。

即使消息持久化，并且队列也持久化，但依然无法避免由于缓存导致的问题，因为消息发送之后和被写入磁盘之间如果出现问题，那么消息依然会丢失，通过publish/confirm机制能够确保客户端知道哪些消息已经存入磁盘，尽管如此，一般不希望遇到因单点故障导致的服务不可用。

镜像队列机制可以将队列镜像到集群的其他节点上，如果集群中的一个节点失效，队列能够自动的切换到镜像的另外一个节点以保证服务可用。

设置镜像队列：在集群的任意节点admin菜单下添加policy，设置策略名称、策略pattern（ ^mirrior表示以mirrior开头的队列）、配置 ha-mode=exactly， ha-params=2， ha-sync-mode=automatic就会报mirrior开头的队列在集群中保存两份

Haproxy + keepalived实现高可用负载均衡

rabbitmq连接时只能连接一个ip，当该节点宕机时，即使队列在别的节点有镜像队列，那么消费者也无法连接到该节点。

haproxy提供高可用、负载均衡及基于tcp、http应用的代理，支持虚拟主机，免费且可靠的一种解决方案。haproxy实现了一种事件驱动、单一进程模型，此模型支持非常大的并发连接数。

nginx、lvs、haproxy直接的区别：www.ha97.com/5646.html

搭建haproxy

• 在两台节点安装： yum install haproxy -y
• 修改配置文件： /etc/haproxy/haproxy.cfg

Keepalived实现双机(主备)热备

后续有时间研究下haproxy和keepalived

Federation Exchange

当两个地区的网络延迟较高时，可以配置Federation Exchange，将两个地区的交换机连接起来，一个作为上游，一个作为下游

使用：

• 开启插件 rabbitmq-plugins enable rabbitmq_federation
• 开启插件管理 rabbitmq-plugins enable rabbitmq_federation_management
• 配置：在下游配置添加一个上游，设置名称和上游的uri如 name=localhost102-upstream， uri=amqp://admin:xxx@localhost102

Shovel

与federation具备的数据转发功能类似，shovel能够可靠、持续的从一个broker中的交换机或队列拉取数据并转发到另一个broker的交换机或队列中，如果source中是一个队列的话，消费者无法消费source队列中的数据，该队列的数据都会被转发到destination

使用：

• 开启插件 rabbitmq-plugins enable rabbitmq_shovel
• 开启插件管理 rabbitmq-plugins enable rabbitmq_shovel_management
• 配置

Original: https://www.cnblogs.com/bingmous/p/16537455.html
Author: Bingmous
Title: RabbitMQ学习笔记