RabbitMQ消息的可靠性投递

缘起

在项目中使用RabbitMQ时，我们可能会遇到这样的问题:如一个订单系统当用户付款成功时我们往消息中间件添加一条记录期望消息消费者修改订单状态，但是最终实际订单状态并没有被修改成功。遇到这种问题我们排查的思路如下:

1.消息是否已经成功发送到消息中间件 2.消息是否有丢失的情况 消息是否已经被消费成功 在生产环境下是不容许出现消息投递/消费错误的情况的，因为这可能会对企业产生巨大的损失，本博客将介绍RabbitMQ如何保证消息的可靠性.

对于RabbitMQ的Message的status，可能会有以下几种情况

• 未接收：由于RabbitMQ所在服务器宕机，客户端的消息发送给RabbitMQ失败
• 未投递：RabbitMQ接收到客户端的消息之后还没来得及给消费者投递消息，结果服务器宕机了
• 投递失败：RabbitMQ把这个消息投递给对应的消费者了，但是消费者宕机了，导致这条消息没能正常消费。

那么对于这三种情况，我们分别要处理的问题也就是以下三个

• 生产者保证消息可靠投递
• RabbitMQ持久化
• 消费者保证消息可靠消费

我们一个一个来解决

生产者保证消息可靠投递

为了保证消息被正确投递到消息中间件，RabbitMQ提供了如下两个配置来保证消息投递的可靠性。

• 在发送消息的时候我们可以设置Mandatory属性。如果设置了Mandatory属性则当消息不能被正确路由到队列中去时将会触发Return Method，这样我们可以在Return Method中进行相关业务处理，如果Mandatory没有设置则当消息不能正确路由到队列中去的时候，Broker将会丢弃该消息
• RabbitMQ还提供了消息确认机制(Publisher Confirm)。生产者将Channel设置成Confirm模式，当设置Confirm模式后所有在该信道上面发布的消息都会被指派一个唯一的ID(从1开始，ID在同个Channel范围是唯一的)，一旦消息被投递到所有匹配的队列之后Broker就会发送一个确认给生产者(包含消息的唯一ID)，这就使得生产者知道消息已经正确到达目的队列了。

如果消息和队列是可持久化的那么确认消息会将消息写入磁盘之后出，Broker回传给生产者的确认消息中DeliverTag域包含了确认消息的序列号，此外Broker也可以设置basic.ack的multiple域，表示到这个序列号之前的所有消息都已经得到了处理(multiple如果为true则表示小于等于deliveryTag的消息都被投递成功，如果为false则表示只有等于deliveryTag的消息已经被投递成功)

除了使用Publisher Confirm方式，RabbitMQ还提供了事务机制保证消息投递，但是使用事务会大大降低系统的吞吐量，就失去了消息中间件存在的意义，本博客不进行探讨。Publisher Confirm模式最大的好处在于他是异步的，一旦发布一条消息生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最终得到确认之后，生产者应用可以通过回调ACK方法来处理该确认消息，如果RabbitMQ因为自身内部错误导致消息丢失，生产者应用可以通过回调NACK方法来处理该确认消息

Publisher Confirm机制在性能上要比事务优越很多，但是Publisher Confirm机制无法进行回滚，一旦服务器崩溃生产者无法得到Confirm信息，生产者其实本身也不知道该消息是否已经被持久化，只有继续重发来保证消息不丢失，但是如果原先已经持久化的消息并不会被回滚，这样队列中就会存在两条相同的消息，系统需要支持去重

Rabbit持久化

假设在运行过程中RabbitMQ服务端宕机了，若此前没有进行持久化操作则消息就会丢失。所以使用RabbitMQ通常建议开启持久化功能

• 交换机持久化 在声明时指定durable为true
• 队列持久化 在声明时指定durable为true
• 消息持久化 在声明时指定delivery_mode为2

消费者保证消息可靠消费

默认情况下，如果Message 已经被某个Consumer正确的接收到了，那么该Message就会被从queue中移除。当然也可以让同一个Message发送到很多的Consumer。如果一个queue没被任何的Consumer Subscribe（订阅），那么，如果这个queue有数据到达，那么这个数据会被cache，不会被丢弃。当有Consumer时，这个数据会被立即发送到这个Consumer，这个数据被Consumer正确收到时，这个数据就被从queue中删除。

如何定义“正确收到”

• 每个Message都要被acknowledged（确认，ack）。我们可以显示的在程序中去ack（Consumer的basic.ack），也可以自动的ack（订阅Queue时指定auto_ack为true）。
• 如果有数据没有被ack，那么RabbitMQ Server会把这个信息发送到下一个Consumer。
• 如果这个app有bug，忘记了ack，那么RabbitMQ Server不会再发送数据给它，因为Server认为这个Consumer处理能力有限。也就是说，在没有收到消费者的ack之前，Server不会再次向这个Consumer投递消息。RabbitMQ会认为这个Consumer还没有处理完上一条消息，没有能力继续接收新消息。
• 而且ack的机制可以起到限流的作用（Benefit to throttling）：在Consumer处理完成数据后发送ack，甚至在额外的延时后发送ack，将有效的balance Consumer的load。

在消费消息的时候手动使用channel.basicAck()进行消息确认，channel.basicNack()进行消息拒绝。

写在最后

本章介绍了RabbitMQ如何保证消息的可靠性投递，看完了这些，想必你已经厉兵秣马，整装待发了，那么下一章，我们就一起来用Java，来做一个RabbitMQ的连接Demo