RabbitMQ 再深入

问即是学

什么是RabbitMQ ?

RabbitMQ是一款开源的，Erlang编写的，基于AMQP协议的消息中间件

什么是消息中间件 ?

是关注于数据的发送和接收，利用高效可靠的异步消息传递机制集成分布式系统

RabbitMQ的使用场景 ?

• 1. 服务间异步通信
• 2. 顺序消费
• 3. 定时任务
• 4. 请求削峰

RabbitMQ基本概念 ?

• Broker: 简单来说就是消息队列服务器实体
• Exchange: 消息交换机，它指定消息按什么规则，路由到哪个队列
• Queue: 消息队列载体，每个消息都会被投入到一个或多个队列
• Binding: 绑定，它的作用就是把exchange和queue按照路由规则绑定起来
• Routing Key: 路由关键字，exchange根据这个关键字进行消息投递
• VHost: vhost 可以理解为虚拟 broker ，即 mini-RabbitMQ server。其内部均含有独立的 queue、exchange 和 binding 等，但最最重要的是，其拥有独立的权限系统，可以做到 vhost 范围的用户控制。当然，从 RabbitMQ 的全局角度，vhost 可以作为不同权限隔离的手段
• Producer: 消息生产者，就是投递消息的程序
• Consumer: 消息消费者，就是接受消息的程序
• Channel: 消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务

RabbitMQ的工作模式 ?

Simple模式 (即最简单的收发模式)

• 1. 消息产生消息，将消息放入队列
• 2. 消息的消费者(consumer) 监听 消息队列,如果队列中有消息,就消费掉,消息被拿走后,自动从队列中删除(隐患 消息可能没有被消费者正确处理,已经从队列中消失了,造成消息的丢失，这里可以设置成手动的ack,但如果设置成手动ack，处理完后要及时发送ack消息给队列，否则会造成内存溢出)。

Work工作模式 (资源的竞争)

• 1. 消息产生者将消息放入队列消费者可以有多个,消费者1,消费者2同时监听同一个队列,消息被消费。
• 2. C1 C2共同争抢当前的消息队列内容,谁先拿到谁负责消费消息(隐患：高并发情况下,默认会产生某一个消息被多个消费者共同使用,可以设置一个开关(synchronize) 保证一条消息只能被一个消费者使用)。

Publish/Subscribe发布订阅 (共享资源)

• 1. 每个消费者监听自己的队列；
• 2. 生产者将消息发给broker，由交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接收到消息。

Routing路由模式

• 1. 消息生产者将消息发送给交换机按照路由判断,路由是字符串(info) 当前产生的消息携带路由字符(对象的方法),交换机根据路由的key,只能匹配上路由key对应的消息队列,对应的消费者才能消费消息;
• 2. 根据业务功能定义路由字符串
• 3. 从系统的代码逻辑中获取对应的功能字符串,将消息任务扔到对应的队列中。
• 4. 业务场景:error 通知;EXCEPTION;错误通知的功能;传统意义的错误通知;客户通知;利用key路由,可以将程序中的错误封装成消息传入到消息队列中,开发者可以自定义消费者,实时接收错误;

Topic主题模式 (路由模式的一种)

• 1. 星号井号代表通配符
• 2. 星号代表多个单词,井号代表一个单词
• 3. 路由功能添加模糊匹配
• 4. 消息产生者产生消息,把消息交给交换机
• 5. 交换机根据key的规则模糊匹配到对应的队列,由队列的监听消费者接收消息消费

如何保证RabbitMQ消息的顺序性 ?

如何确保消息正确地发送至 RabbitMQ ?

发送方确认模式

• 将信道设置成 confirm 模式（发送方确认模式），则所有在信道上发布的消息都会被指派一个唯一的 ID。
• 一旦消息被投递到目的队列后，或者消息被写入磁盘后（可持久化的消息），信道会发送一个确认给生产者（包含消息唯一 ID）。
• 如果 RabbitMQ 发生内部错误从而导致消息丢失，会发送一条 nack（not acknowledged，未确认）消息。
• 发送方确认模式是异步的，生产者应用程序在等待确认的同时，可以继续发送消息。当确认消息到达生产者应用程序，生产者应用程序的回调方法就会被触发来处理确认消息。

接收方确认机制

• 消费者接收每一条消息后都必须进行确认（消息接收和消息确认是两个不同操作）。只有消费者确认了消息，RabbitMQ 才能安全地把消息从队列中删除。
• 这里并没有用到超时机制，RabbitMQ 仅通过 Consumer 的连接中断来确认是否需要重新发送消息。也就是说，只要连接不中断，RabbitMQ 给了 Consumer 足够长的时间来处理消息。保证数据的最终一致性；

下面罗列几种特殊情况

• 如果消费者接收到消息，在确认之前断开了连接或取消订阅，RabbitMQ 会认为消息没有被分发，然后重新分发给下一个订阅的消费者。（可能存在消息重复消费的隐患，需要去重）
• 如果消费者接收到消息却没有确认消息，连接也未断开，则 RabbitMQ 认为该消费者繁忙，将不会给该消费者分发更多的消息。

如何保证RabbitMQ消息的可靠传输 ?

消息不可靠的情况可能是消息丢失，劫持等原因；

丢失又分为：生产者丢失消息、消息列表丢失消息、消费者丢失消息；

生产者丢失消息

从生产者弄丢数据这个角度来看，RabbitMQ提供transaction和confirm模式来确保生产者不丢消息；

• transaction机制

发送消息前，开启事务（channel.txSelect()）,然后发送消息，如果发送过程中出现什么异常，事务就会回滚（channel.txRollback()）,如果发送成功则提交事务（channel.txCommit()）。然而，这种方式有个缺点：吞吐量下降；

• confirm模式

一旦channel进入confirm模式，所有在该信道上发布的消息都将会被指派一个唯一的ID（从1开始），一旦消息被投递到所有匹配的队列之后；
rabbitMQ就会发送一个ACK给生产者（包含消息的唯一ID），这就使得生产者知道消息已经正确到达目的队列了；
如果rabbitMQ没能处理该消息，则会发送一个Nack消息给你，你可以进行重试操作。

消息队列丢失消息

消息持久化。处理消息队列丢数据的情况，一般是开启持久化磁盘的配置。

如何持久化?

• 1. 将queue的持久化标识durable设置为true,则代表是一个持久的队列
• 2. 发送消息的时候将deliveryMode=2

消费者丢失消息

消费者丢数据一般是因为采用了自动确认消息模式，改为手动确认消息即可！
解决方案：处理消息成功后，手动回复确认消息。

RabbitMQ 上的一个 queue 中存放的 message 是否有数量限制 ?

可以认为是无限制，因为限制取决于机器的内存，但是消息过多会导致处理效率的下降。

消息基于什么传输 ?

RabbitMQ 使用信道的方式来传输数据。信道是建立在真实的 TCP 连接内的虚拟连接，且每条 TCP 连接上的信道数量没有限制。

消息如何分发 ?

若该队列至少有一个消费者订阅，消息将以循环（round-robin）的方式发送给消费者。每条消息只会分发给一个订阅的消费者（前提是消费者能够正常处理消息并进行确认）。通过路由可实现多消费的功能

消息怎么路由 ?

消息提供方->路由->一至多个队列消息发布到交换器时，消息将拥有一个路由键（routing key），在消息创建时设定。通过队列路由键，可以把队列绑定到交换器上。消息到达交换器后，RabbitMQ 会将消息的路由键与队列的路由键进行匹配（针对不同的交换机有不同的路由规则）；

常用的交换机主要分为一下三种：

• fanout: 如果交换机收到消息，将会广播到所有绑定的队列上
• direct: 如果路由键完全匹配，消息就被投递到相应的队列
• topic: 可以使来自不同源头的消息能够到达同一个队列。 使用 topic 交换机时，可以使用通配符

RabbitMQ 的优缺点 ?

优点

• 异步处理 - 相比于传统的串行、并行方式，提高了系统吞吐量。
• 应用解耦 - 系统间通过消息通信，不用关心其他系统的处理。
• 流量削锋 - 可以通过消息队列长度控制请求量；可以缓解短时间内的高并发请求。
• 日志处理 - 解决大量日志传输。
• 消息通讯 - 消息队列一般都内置了高效的通信机制，因此也可以用在纯的消息通讯。比如实现点对点消息队列，或者聊天室等。

缺点

• 系统可用性降低
• 系统复杂度提高
• 一致性问题