消息中间件RabbitMQ可靠性投递与生产实践

约定不等于承诺〃 2022-10-01 08:47 164阅读 0赞

# 本章重点： #

> 可靠性投递
> 
> 1.  确保消息发送到RabbitMQ服务器
> 2.  确保消息被正确的路由
> 3.  确保消息在队列正确地存储
> 4.  确保消息从队列正确地投递到消费者
> 5.  消费者回调
> 6.  补偿机制
> 7.  消息幂等性
> 8.  消息的顺序性

# 可靠性投递 #

首先需要明确，效率和可靠性是无法兼得的，如果要保证每一个环节都成功，势必会对消息的收发效率造成影响，如过是一些业务实时性要求不是特别高的场合，可以牺牲可靠性来换取效率。

1.  ①代表消息从生产者发送到Exchange
2.  ②代表消息从Exchange路由到Queue
3.  ③ 代表消息在Queue中存储；
4.  ④ 代表消费者订阅Queue并消费消息。

## 1.确保消息发送到RabbitMQ服务器 ##

可能因为网络或者Broker的问题导致①失败，而生产者是无法得知消息是否正确发送到Broker的。

有两种解决方案：

第一种是Transaction事务模式

第二种是Confirm确认模式

> 1.在通过channel.txSelect方法开启事务之后，我们便可以发布消息给RabbitMQ了，如果事务提交成功，则消息一定 到达了RabbitMQ中，如果在事务提交执行之前由于RabbitMQ异常崩溃或者其他原因抛出异常，这个时候我们便可以将其捕获，进而通过执行channel.txRollback方法来实现事务回滚。使用事务机制的话会“吸干”RabbitMQ的性 能，一般不建议使用。 2.生产者通过调用channel.conﬁrmSelect方法（即Conﬁrm.Select命令）将信道设置为conﬁrm模式。一旦消息被投递到所有匹配的队列之后，RabbitMQ就会发送一个确认（Basic.Ack）给生产者（包含消息的唯一ID），这就使得生产者知晓消息已经正确到达了目的地了。

## 2.确保消息被正确的路由 ##

可能因为路由关键字错误，或者队列不存在，或者队列名称错误导致②失败。

1.  使用mandatory参数和ReturnListener，可以实现消息无法路由的时候返回给生产者。
2.  另一种方式就是使用备份交换机（alternate-exchange），无法路由的消息会发送到这个交换机上。

Map<String,Object> arguments = new HashMap<String,Object>();
    // 指定交换机的备份交换机
    arguments.put("alternate-exchange","ALTERNATE_EXCHANGE"); 
    channel.exchangeDeclare("TEST_EXCHANGE","topic", false, false, false, arguments);
    复制代码复制代码

## 3.确保消息在队列正确地存储 ##

可能因为系统宕机、重启、关闭等等情况导致存储在队列的消息丢失，即③出现问题。

解决方案：

1.队列持久化

// String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments 
    channel.queueDeclare(QUEUE_NAME, true, false, false, null);
    复制代码复制代码

2.交换机持久化

// String exchange, boolean durable 
    channel.exchangeDeclare("MY_EXCHANGE","true");
    复制代码复制代码

3.消息持久化

AMQP.BasicProperties properties = new AMQP.BasicProperties
    .Builder() 
    // 2代表持久化，其他代表瞬态 
    .deliveryMode(2) 
    .build(); 
    channel.basicPublish("", QUEUE_NAME, properties, msg.getBytes());
    复制代码复制代码

## 4.确保消息从队列正确地投递到消费者 ##

> 如果消费者收到消息后未来得及处理即发生异常，或者处理过程中发生异常，会导致④失败。 为了保证消息从队列可靠性到达消费者，RabbitMQ提供了消息确认机制（message acknowledgement）,消费者在订阅队列时，可以指定autoAck参数，当autoAck等于false时，RabbitMQ会等待消费者显示地回复确认消息才从队列中删除该消息。 如果消息消费失败，也可以调用Basic.Reject或者BasicNack来拒绝当前消息而不是确认，如果requere参数为true，可以把这条消息重新存入队列，以便发送给下一个消费者。

## 5.消费者回调 ##

消费者处理消息之后，可以再发送一条消息给生产者，或者调用生产者地API，告知消息处理完毕。

## 6.补偿机制 ##

对于一定时间没有响应地消息，可以设置一个定时重发地机制，但是要控制次数，比如最多重复三次，否则会造成消息堆积。

## 7.消息幂等性 ##

服务端是没有这种控制的，只能在消费端控制。

如何避免消息的重复消费？

消息重复消费可能会有两个原因：

1.  生产者的问题。环节①重复发送消息，比如在开启Confirm模式但未收到确认
2.  环节④出了问题，由于消费者未发送ACK或者其它原因，消息重复投递

对于重复发送的消息，可以对每一条消息生成一个唯一的业务id，通过日志或者建表来做重复控制。

## 8.消息的顺序性 ##

消息的顺序性是指消费者消费消息的顺序跟生产者投递消息的顺序是一致的。