一文了解消息中间件RabbitMQ

消息中间件作用有三个&＃xff1a;异步处理消息、降低耦合度、流量削峰。

当前使用较多的消息队列有RabbitMQ、RocketMQ、ActiveMQ、Kafka、ZeroMQ、MetaMQ等&＃xff0c;而部分数据库如Redis、MySQL以及phxsql也可实现消息队列的功能

3.1 RabbitMQ介绍

RabbitMQ是采用erlang语言开发的&＃xff0c;必须有erlang环境才可以运行。RabbitMQ是一个遵循AMQP协议的消息中间件&＃xff0c;它从生产者接收消息并递送给消费者&＃xff0c;在这个过程中&＃xff0c;根据规则进行路由&＃xff0c;缓存与持久化。

3.2 RabbitMQ的基本介绍

1. producer&＃xff1a;消息生产者&＃xff0c;就是投递消息的程序&＃xff1b;

2. Message&＃xff1a;由生产者通过RabbitMQ发送给消费者的信息&＃xff08;路由键&＃xff08;Routing Key&＃xff09;是供交换机查看并根据键来决定如何分发消息到列队的一个键。路由键可以说是消息的目的地址&＃xff09;&＃xff1b;

3. 连接&＃xff08;Connection&＃xff09;&＃xff1a;连接RabbitMQ和应用服务器的TCP连接&＃xff1b;

4. 信道&＃xff08;Channel&＃xff09;&＃xff1a;消息通道&＃xff0c;在客户端的每个连接里&＃xff0c;可建立多个channel&＃xff0c;每个channel代表一个会话任务&＃xff1b;

   为什么不通过TCP直接发送命令&＃xff1a; 创建和销毁TCP会话是非常昂贵的开销&＃xff0c;信道的原理是一条线程一条信道&＃xff0c;多条线程多条信道共同使用一条TCP连接&＃xff0c;每个信道的ID唯一&＃xff1b;

5. 绑定&＃xff08;Binding&＃xff09;&＃xff1a;绑定是队列和交换机的一个关联连接&＃xff1b;

6. Exchange&＃xff1a;通过routingkey和队列绑定在一起的&＃xff0c;如果消息拥有的“路由键”跟队列和交换器的“路由键”匹配&＃xff0c;那么消息就会被路由到该绑定的队列当中去&＃xff1b;

   Exchange的类型&＃xff1a;
   &＃xff08;1&＃xff09;fanout&＃xff1a;消息会被转发到所有与交换机绑定队列&＃xff1b;
   &＃xff08;2&＃xff09;direct&＃xff1a;精确匹配Routing key和routing key&＃xff1b;
   &＃xff08;3&＃xff09;topic&＃xff1a;模糊匹配Routing key和routing key

7. 队列&＃xff08;Queue&＃xff09;&＃xff1a;存储消息的缓存&＃xff1b;

8. Consumer&＃xff1a;处理消息&＃xff1b;
   消费模式&＃xff1a;

&＃xff08;1&＃xff09;pull&＃xff1a;消费者主动拉取消息

优点&＃xff1a;客户端可以依据自己的消费能力进行消费 &＃xff1b;传输失败时不需要重试&＃xff0c;反正数据还在服务端。
缺点&＃xff1a;消息的实时性难以实现&＃xff1b;
场景&＃xff1a;服务端生产消息数据比较大时&＃xff0c;而消费端处理比较复杂&＃xff0c;消费能力相对较低。


&＃xff08;2&＃xff09;push&＃xff1a;服务端主动推送消息给客户端

优点&＃xff1a;服务端主动推送给客户端&＃xff0c;实时性很高&＃xff1b;
缺点&＃xff1a;当客户端消费能力远低于服务端生产能力&＃xff0c;那么一旦服务端推送大量消息到 客户端时&＃xff0c;就会导致客户端消息堆积&＃xff0c;处理缓慢&＃xff0c;甚至服务崩溃&＃xff1b;
服务端需要维护每次传输状态&＃xff0c;以防消息传递失败进行重试。
场景&＃xff1a;适合用于数据实时性要求高的场景


3.3 RabbitMQ的队列模式

1. 普通队列
   非持久化消息&＃xff1a;消息存储在内存之中&＃xff1b;持久化的消息&＃xff1a;存储在磁盘中&＃xff1b;
   适用场景&＃xff1a;消息量不是很大的情况
2. 镜像队列
   消息发送到队列时&＃xff0c;如果队列是镜像队列&＃xff0c;会将队列镜像到cluster中其他的节点之上&＃xff1b;
   对于持久化的消息&＃xff0c;会在内存和磁盘中各存一份。
   缺点&＃xff1a;镜像队列不能实现负载均衡&＃xff1b;对exclusive队列设置镜像并不会有任何作用。

队列结构&＃xff1a;是一个特殊的BackingQueue&＃xff0c;它内部包裹了一个普通的BackingQueue做本地消息持久化处理&＃xff0c;增加了将消息和ack复制到所有镜像的功能。所有对mirror_queue_master的操作&＃xff0c;会通过组播GM的方式同步到各slave节点。GM负责消息的广播&＃xff0c;mirror_queue_slave负责回调处理&＃xff0c;而master上的回调处理是由coordinator负责完成。mirror_queue_slave中包含了普通的BackingQueue进行消息的存储&＃xff0c;master节点中BackingQueue包含在mirror_queue_master中由AMQQueue进行调用。

3. exclusive队列

   只对首次声明它的连接可见&＃xff1b;
   会在其连接断开的时候自动删除&＃xff1b;

4. lazy队列
   优点&＃xff1a;尽可能的将消息存储到磁盘中&＃xff0c;在消费者消费到相应的消息时才会被加载到内存中&＃xff0c;占用内存少&＃xff1b;
   缺点&＃xff1a;存在队列的单点故障&＃xff1b;
   使用场景&＃xff1a;发送端过快或消费端宕机&＃xff0c;导致消息大量积压&＃xff0c;在消息大爆发的时候&＃xff0c;MQ服务器会撑不住&＃xff0c;影响其他队列的消息收发。

5. 死信队列
   死信消息如何形成的&＃xff1a;
   &＃xff08;1&＃xff09;消息被否定确认&＃xff1b;
   &＃xff08;2&＃xff09;消息在队列的存活时间超过设置的TTL时间&＃xff1b;
   &＃xff08;3&＃xff09;消息队列的消息数量已经超过最大队列长度。
   死信队列&＃xff1a;如果配置了死信队列信息&＃xff0c;那么该消息将会被丢进死信队列中&＃xff1b;如果没有配置&＃xff0c;则该消息将会被丢弃。
   场景&＃xff1a;故障发生时&＃xff0c;让未正确处理的消息存放到死信队列中&＃xff0c;待后续排查清楚问题后&＃xff0c;处理死信消息。

6. 延迟队列
   场景&＃xff1a;实现定时任务
   实现方式&＃xff1a;
   &＃xff08;1&＃xff09;延迟插件&＃xff1b;
   &＃xff08;2&＃xff09;ttl&＃43;死信队列&＃xff1a;消息过期时间/队列过期时间&＃xff0c;如果同时设置队列和消息过期时间&＃xff0c;较小的生效。

3.3 RabbitMQ的工作模式

1. 简单模式&＃xff1a;一个生产者&＃xff0c;一个消费者&＃xff0c;一个队列。
2. 工作模式&＃xff1a;一个生产者对应多个消费者&＃xff0c;但是一条消息只能被一个消费者获取。
3. 发布订阅模式&＃xff1a;一个消费者将消息首先发送到交换器&＃xff0c;交换器绑定到多个队列&＃xff0c;然后被监听该队列的消费者所接收并消费。
4. 路由模式&＃xff1a;使用的是type为direct的交换机&＃xff0c;让消费者有选择性的接收消息。
5. 主题模式&＃xff1a;使用type为topic的交换机

4.1 节点类型

磁盘节点&＃xff1a;配置信息和元信息存储在磁盘上&＃xff0c;至少一个磁盘节点。如果唯一磁盘的磁盘节点崩溃&＃xff0c;集群是可以保持运行的&＃xff0c;但不能更改任何东西&＃xff0c;集群中最好设置两个磁盘节点。

内存节点&＃xff1a;配置信息和元信息存储在内存中&＃xff0c; 性能好&＃xff1b;

4.2 普通集群

默认普通队列&＃xff0c;集群仅采用元数据同步。仅同步元数据的原因&＃xff1a;存储空间和性能。

特点&＃xff1a;
&＃xff08;1&＃xff09;master存储队列的元数据信息和消息&＃xff1b;
&＃xff08;2&＃xff09;slave只同步队列元数据信息&＃xff0c;不会同步存储消息&＃xff1b;
&＃xff08;3&＃xff09;master&＃xff08;读写&＃xff09;&＃xff0c;slave&＃xff08;不提供读写&＃xff09;作为路由节点起到转发作用&＃xff1b;

缺点&＃xff1a;master节点故障后&＃xff0c;slave节点无法取到master节点中还未消费的消息实体。如果做了消息持久化&＃xff0c;等master节点恢复&＃xff0c;然后才可被消费&＃xff1b;如果没有持久化的话&＃xff0c;就会产生消息丢失的现象&＃xff1b;

4.3 镜像模式

1. 把需要的队列做成镜像队列&＃xff0c;master会将队列镜像到cluster中其他的slave节点之上。逻辑上的master和slave&＃xff0c;相对于主队列而言。

2. 特点&＃xff1a;

   &＃xff08;1&＃xff09;消息的发布与消费都是通过master节点完成&＃xff0c;master通过 mirror 队列把消息数据同步到其他的 slave 节点&＃xff0c;对于持久化的消息会在磁盘和内存中各存一份&＃xff1b;
   &＃xff08;2&＃xff09;除publish外所有动作都只会向master发送&＃xff0c;然后由master将命令执行的结果广播给slave&＃xff1b;
   &＃xff08;3&＃xff09;master节点故障后&＃xff0c;集群中最老的slave被提升为master&＃xff1b;slave节点故障后&＃xff0c;仅仅是相邻节点感知&＃xff0c;然后重新调整邻居节点信息。

3. 镜像队列 gm&＃xff1a;

   通过将所有 gm 进程形成一个循环链表&＃xff0c;每个 gm 都会监控位于自己左右两边的 gm&＃xff0c;当有 gm 新增时&＃xff0c;相邻的 gm 保证当前广播的消息会通知到新的 gm 上&＃xff1b;当有 gm 失效时&＃xff0c;相邻的 gm 会接管保证本次广播消息会通知到所有 gm。

4. 镜像队列的消息广播流程&＃xff1a;

   &＃xff08;1&＃xff09;Master 节点发出消息&＃xff0c;顺着镜像队列循环列表发送&＃xff1b;
   &＃xff08;2&＃xff09;所有 Slave 节点收到消息会对消息进行缓存&＃xff08;Slave 节点缓存消息用于在广播过程中&＃xff0c;有节点失效或者新增节点&＃xff0c;这样左侧节点感知变化后会重新将消息推送给右侧节点&＃xff09;&＃xff1b;
   &＃xff08;3&＃xff09;当 Master 节点收到自己发送的消息后意味着所有节点都收到了消息&＃xff0c;会再次广播 Ack 消息&＃xff1b;
   &＃xff08;4&＃xff09;Ack 消息同样会顺着循环列表经过所有 Slave 节点&＃xff0c;通知 Slave 节点可以清除缓存消息&＃xff1b;
   &＃xff08;5&＃xff09;当 Ack 消息回到 Master 节点&＃xff0c;对应消息的广播结束。

5. 优点&＃xff1a;&＃xff08;1&＃xff09;保证 100% 数据不丢失&＃xff1b; &＃xff08;2&＃xff09;消除普通模式中队列消息单点带来的风险&＃xff1b;

6. 缺点&＃xff1a;降低系统性能&＃xff0c;集群内部的同步通讯会占用大量的网络带宽

4,4 远程模式&＃xff08;shovel&＃xff09;

远程模式可以实现双活的一种模式&＃xff0c;简称 shovel 模式&＃xff0c;所谓的 shovel 就是把消息进行不同数据中心的复制工作&＃xff0c;可以跨地域的让两个 MQ 集群互联&＃xff0c;远距离通信和复制。
Shovel 把消息进行数据中心的复制工作&＃xff0c;可以跨地域的让两个 MQ 集群互联。

原理&＃xff1a;使用shovel插件实现&＃xff0c;通过定义RabbitMQ上一个队列和另外一个RabbitMQ上的交换机之间的复制关系来实现远程复制。它会在主服务上建立一个队列来监听交换机&＃xff0c;所有到交换机的消息会投递到该队列&＃xff0c;并且在从服务中订阅这个队列&＃xff0c;使队列中的消息复制到从服务的交换机中。优点&＃xff1a;异地的 MQ 集群&＃xff0c;近端同步确认&＃xff0c;远端异步确认&＃xff1b;缺点&＃xff1a;配置比较复杂

4.5 多活模式

federation 插件是一个不需要构建 cluster &＃xff0c;而在 brokers 之间传输消息的高性能插件&＃xff0c;federation 插件可以在 brokers 或者 cluster 之间传输消息&＃xff0c;连接的双方可以使用不同的 users 和 virtual hosts&＃xff0c;双方也可以使用不同版本的 rabbitMQ 和 erlang。federation 插件使用 AMQP 协议通信&＃xff0c;可以接受不连续的传输。

federation 插件实现&＃xff0c;只配置两个集群中两个单点队列之间的复制&＃xff0c;每个集群内部自己会完成多节点的同步&＃xff08;镜像队列来完成&＃xff09;。

优点&＃xff1a;Federation插件可以在Brokers或者Cluster之间传输消息&＃xff0c;连接双方可以使用不同的users和vistual hosts&＃xff1b; 双方也可以使用版本不同的RabbitMQ和Erlang。

rabbitmq发出消息后会等待consumer端应答&＃xff0c;只有收到ack确定信息后才会将消息在rabbitmq清除掉。

5.1 自动ack

&＃xff08;1&＃xff09;autoAck&＃61;true&＃xff1b;
&＃xff08;2&＃xff09;消息一旦被接收&＃xff0c;消费者自动发送ACK&＃xff1b;

对于自动确认来说&＃xff0c;当方法没有异常执行完毕后&＃xff0c;会对MQ发出ACK&＃xff1b;若方法出现异常&＃xff0c;会对MQ发出nack&＃xff0c;消息会重回队列。

场景&＃xff1a;消息不太重要&＃xff0c;丢失也没有影响

5.2 手动ack

保证数据不被重复消费&＃xff0c;同时数据也不能少&＃xff0c;也就是数据一致性。
&＃xff08;1&＃xff09;关闭自动ack&＃xff1a;autoAck&＃61;false&＃xff1b;
&＃xff08;2&＃xff09;消息接收后&＃xff0c;不会自动发送ACK&＃xff0c;需要手动调用

场景&＃xff1a;消息非常重要&＃xff0c;异常发生不会丢数据。

如果重复ack&＃xff08;自动ack的同时开启手动ack&＃xff09;会造成造成消息丢失