RabbitMQ学习（五）消息确认机制

男娘i 2022-03-16 11:42 235阅读 0赞

# 转载：[https://blog.csdn.net/wqc19920906/article/details/82227164][https_blog.csdn.net_wqc19920906_article_details_82227164] #

[https://blog.csdn.net/anumbrella/article/details/81321701][https_blog.csdn.net_anumbrella_article_details_81321701]

2018年08月30日 20:18:52 [jcpp9527][] 阅读数：187

原文：[RabbitMQ学习（二）：Java使用RabbitMQ要点知识][RabbitMQ_Java_RabbitMQ]

附带的demo：[https://download.csdn.net/download/leixiaotao\_java/10177767][https_download.csdn.net_download_leixiaotao_java_10177767]

一、**Producer消息确认机制**

### **1、什么是生产者消息确认机制？** ###

没有消息确认模式时，生产者不知道消息是不是已经到达了Broker[服务器][Link 1]，这对于一些业务严谨的系统来说将是灾难性的。消息确认模式可以采用AMQP协议层面提供的事务机制实现（此文没有这种实现方式），但是会降低RabbitMQ的吞吐量。RabbitMQ自身提供了一种更加高效的实现方式：confirm模式。

消息生产者通过调用Channel.confirmSelect()方法将Channel信道设置成confirm模式。一旦信道被设置成confirm模式，该信道上的所有消息都会被指派一个唯一的ID（从1开始），一旦消息被对应的Exchange接收，Broker就会发送一个确认给生产者（其中deliveryTag就是此唯一的ID），这样消息生产者就知道消息已经成功到达Broker。

confirm模式最大的好处在于他是异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最终得到确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果RabbitMQ因为自身内部错误导致消息丢失，就会发送一条nack消息，生产者应用程序同样可以在回调方法中处理该nack消息。

在channel 被设置成 confirm 模式之后，所有被 publish 的后续消息都将被 confirm（即 ack） 或者被nack一次。但是没有对消息被 confirm 的快慢做任何保证，并且同一条消息不会既被 confirm又被nack 。

### **2、开启confirm模式** ###

如上所说生产者通过调用Channel.confirmSelect()方法将Channel信道设置成confirm模式。

**注意**：已经在transaction事务模式的channel是不能再设置成confirm模式的，即这两种模式是不能共存的。

### **3、普通confirm模式** ###

普通confirm模式是串行的，即每次发送了一次消息，生产者都要等待Broker的确认消息，然后根据确认标记权衡消息重发还是继续发下一条。由于是串行的，在效率上是比较低下的。

（1）重点方法

1.  `/**`
2.  `* 等待Broker返回消息确认标记`
3.  `* 注意，在非确定的通道，waitforconfirms抛出IllegalStateException。`
4.  `* @return 是否发送成功`
5.  `* @throws java.lang.IllegalStateException`
6.  `*/`
7.  `boolean waitForConfirms() throws InterruptedException;`

（2）部分使用代码如下：

1.  `//注意：返回的时候Return在前，Confirm在后`
2.  `channel.confirmSelect();`
3.  `int i=1;`
4.  `while (i<=50) {`
5.  `//发布消息`
6.  `channel.basicPublish("",queueName,true,MessageProperties.TEXT_PLAIN,SerializationUtils.serialize(object));`
7.  `//等待Broker的确认回调`
8.  `if(channel.waitForConfirms())`
9.  `System.out.println("send success!");`
10. `else`
11. `System.out.println("send error!");`
12. `i++;`
13. `}`

### **3.4、批量confirm模式** ###

批量confirm模式是异步的方式，效率要比普通confirm模式高许多，但是此种方式也会造成线程阻塞，想要进行失败重发就必须要捕获异常。网络上还有采用waitForConfirms()实现批量confirm模式的，但是只要一条失败了，就必须把这批次的消息统统再重发一次，非常的消耗性能，因此此文不予考虑。

（1）重点代码

1.  `/**`
2.  `* 等待直到所有消息被确认或者某个消息发送失败。如果消息发送确认失败了，`
3.  `* waitForConfirmsOrDie 会抛出IOException异常。当在非确认通道上调用时`
4.  `* ，会抛出IllegalStateException异常。`
5.  `* @throws java.lang.IllegalStateException`
6.  `*/`
7.  `void waitForConfirmsOrDie() throws IOException, InterruptedException;`

（2）部分代码如下：

1.  `//注意：返回的时候Return在前，Confirm在后`
2.  `channel.confirmSelect();`
3.  `int i=1;`
4.  `while (i<=50) {`
5.  `//发布消息`
6.  `channel.basicPublish("",queueName,true,MessageProperties.TEXT_PLAIN,SerializationUtils.serialize(object));`
7.  `i++;`
8.  `}`
9.  `channel.waitForConfirmsOrDie();`

### **3.5、ConfirmListener[监听器][Link 2]模式** ###

RabbitMQ提供了一个ConfirmListener接口专门用来进行确认监听，我们可以实现ConfirmListener接口来创建自己的消息确认监听。ConfirmListener接口中包含两个回调方法：

1.  `/**`
2.  `* 生产者发送消息到exchange成功的回调方法`
3.  `*/`
4.  `void handleAck(long deliveryTag, boolean multiple) throws IOException;`
5.  `/**`
6.  `* 生产者发送消息到服务器broker失败的回调方法，服务器丢失了此消息。`
7.  `* 注意，丢失的消息仍然可以传递给消费者，但broker不能保证这一点。`
8.  `*/`
9.  `void handleNack(long deliveryTag, boolean multiple) throws IOException;`

其中deliveryTag是Broker给每条消息指定的唯一ID（从1开始）；multiple表示是否接收所有的应答消息，比如multiple=true时，发送100条消息成功过后，我们并不会收到100次handleAck方法调用。

（1）重要方法

1.  `//注册消息确认监听器`
2.  `channel.addConfirmListener(new MyConfirmListener());`

（2）部分使用代码如下：

1.  `//注意：返回的时候Return在前，Confirm在后`
2.  `channel.confirmSelect();`
3.  `//注册消息确认监听器`
4.  `channel.addConfirmListener(new MyConfirmListener());`
5.  `//注册消息结果返回监听器`
6.  `channel.addReturnListener(new MyReturnListener());`
7.  `int i=1;`
8.  `while (i<=50) {`
9.  `//发布消息`
10. `channel.basicPublish("",queueName,true,MessageProperties.TEXT_PLAIN,SerializationUtils.`
11. `serialize(object));`
12. `i++;`
13. `}`

1.  `//自定义的消息确认监听器`
2.  `public class MyConfirmListener implements ConfirmListener{`
3.  `/**`
4.  `* 生产者发送消息到exchange成功的回调方法`
5.  `* 消息被Exchange接受以后，如果没有匹配的Queue，则会被丢弃。但是可以设置ReturnListener监听来监听有没有匹配的队列。`
6.  `* 因此handleAck执行了，并不能完全表示消息已经进入了对应的队列，只能表示对应的exchange成功的接收了消息。`
7.  `* 消息被exchange接收过后，还需要通过一定的匹配规则分发到对应的队列queue中。`
8.  `*/`
9.  `public void handleAck(long deliveryTag, boolean multiple) throws IOException {`
10. `//注意：deliveryTag是broker给消息指定的唯一id（从1开始）`
11. `System.out.println("Exchange接收消息："+deliveryTag+"（deliveryTag）成功！multiple="+multiple);`
12. `}`
13. `/**`
14. `* 生产者发送消息到服务器broker失败的回调方法，服务器丢失了此消息。`
15. `* 注意，丢失的消息仍然可以传递给消费者，但broker不能保证这一点。（不明白，既然丢失了，为啥还能发送）`
16. `*/`
17. `public void handleNack(long deliveryTag, boolean multiple) throws IOException {`
18. `System.out.println("Exchange接收消息："+deliveryTag+"（deliveryTag）失败！服务器broker丢失了消息");`
19. `}`
20. `}`

1.  `//自定义的结果返回监听器`
2.  `/**`
3.  `* 实现此接口以通知交付basicpublish失败时，“mandatory”或“immediate”的标志监听（源代码注释翻译）。`
4.  `* 在发布消息时设置mandatory等于true，监听消息是否有相匹配的队列，`
5.  `* 没有时ReturnListener将执行handleReturn方法，消息将返给发送者`
6.  `*/`
7.  `public class MyReturnListener implements ReturnListener {`
8.  `public void handleReturn(int replyCode, String replyText, String exchange, String routingKey,`
9.  `BasicProperties properties, byte[] body) throws IOException {`
10. `System.out.println("消息发送到队列失败：回复失败编码："+replyCode+";回复失败文本："+replyText+";失败消息对象："+SerializationUtils.deserialize(body));`
11. `}`
12. `}`

## 4、**Consumer消息确认机制** ##

为了保证消息从队列可靠地到达消费者，RabbitMQ提供消息确认机制(message acknowledgment)。消费者在注册消费者时，可以指定noAck参数，当noAck=false时，RabbitMQ会等待消费者显式发回ack信号后才从内存（或磁盘，如果是持久化消息的话）中移去消息。否则，RabbitMQ会在队列中消息被消费后立即删除它。

当noAck=false时，对于RabbitMQ服务器端而言，队列中的消息分成了两部分：一部分是等待投递给消费者的消息（web管理界面上的Ready状态）；一部分是已经投递给消费者，但是还没有收到消费者ack信号的消息（web管理界面上的Unacked状态）。如果服务器端一直没有收到消费者的ack信号，并且消费此消息的消费者已经断开连接，则服务器端会安排该消息重新进入队列，等待投递给下一个消费者（也可能还是原来的那个消费者）。

（1）重要方法

1.  `/**`
2.  `*1. 开始一个非局部、非排他性消费， with a server-generated consumerTag.`
3.  `* 注意：执行这个方法会回调handleConsumeOk方法,在此方法中处理消息。`
4.  `* @param queue 队列名称`
5.  `* @param autoAck 是否自动应答。false表示consumer在成功消费过后必须要手动回复一下服务器，如果不回复，服务器就将认为此条消息消费失败，继续分发给其他consumer。`
6.  `* @param callback 回调方法类`
7.  `* @return 由服务器生成的consumertag`
8.  `* @throws java.io.IOException if an error is encountered`
9.  `*/`
10. `String basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException;`
11.  
12.  
13. `/**`
14. `*2`
15. `consumer处理成功后，通知broker删除队列中的消息，如果设置multiple=true，表示支持批量确认机制以减少网络流量。`
16. `例如：有值为5,6,7,8 deliveryTag的投递`
17. `如果此时channel.basicAck(8, true);则表示前面未确认的5,6,7投递也一起确认处理完毕。`
18. `如果此时channel.basicAck(8, false);则仅表示deliveryTag=8的消息已经成功处理。`
19. `*/`
20. `void basicAck(long deliveryTag, boolean multiple) throws IOException;`
21.  
22. `/**3`
23. `consumer处理失败后，例如：有值为5,6,7,8 deliveryTag的投递。`
24. `如果channel.basicNack(8, true, true);表示deliveryTag=8之前未确认的消息都处理失败且将这些消息重新放回队列中。`
25. `如果channel.basicNack(8, true, false);表示deliveryTag=8之前未确认的消息都处理失败且将这些消息直接丢弃。`
26. `如果channel.basicNack(8, false, true);表示deliveryTag=8的消息处理失败且将该消息重新放回队列。`
27. `如果channel.basicNack(8, false, false);表示deliveryTag=8的消息处理失败且将该消息直接丢弃。`
28. `*/`
29. `void basicNack(long deliveryTag, boolean multiple, boolean requeue) throws IOException;`
30.  
31. `/**4`
32. `相比channel.basicNack，除了没有multiple批量确认机制之外，其他语义完全一样。`
33. `如果channel.basicReject(8, true);表示deliveryTag=8的消息处理失败且将该消息重新放回队列。`
34. `如果channel.basicReject(8, false);表示deliveryTag=8的消息处理失败且将该消息直接丢弃。`
35. `*/`
36. `void basicReject(long deliveryTag, boolean requeue) throws IOException;`

（2）部分使用代码如下：

1.  `//this表示自己的Consumer`
2.  `channel.basicConsume(queueName, false, this);`
3.  `...`
4.  `@Override`
5.  `public void handleDelivery(String arg0, Envelope envelope, BasicProperties arg2, byte[] body) throws IOException {`
6.  `if (body == null)`
7.  `return;`
8.  `Map<String, Object> map = (Map<String, Object>) SerializationUtils.deserialize(body);`
9.  `/**`
10. `* 专门处理奇数消息的消费者`
11. `*/`
12. `int tagId = (Integer) map.get("tagId");`
13. `if (tagId % 2 != 0) {`
14. `//处理消息`
15. `System.out.println("接收并处理消息："+tagId);`
16. `//通知服务器此消息已经被处理了`
17. `channel.basicAck(envelope.getDeliveryTag(), false);`
18. `}else{`
19. `//通知服务器消息处理失败，重新放回队列。false表示处理失败消息不放会队列，直接删除`
20. `channel.basicReject(envelope.getDeliveryTag(), true);`
21. `}`
22. `}`
23.

[https_blog.csdn.net_wqc19920906_article_details_82227164]: https://blog.csdn.net/wqc19920906/article/details/82227164
[https_blog.csdn.net_anumbrella_article_details_81321701]: https://blog.csdn.net/anumbrella/article/details/81321701
[jcpp9527]: https://me.csdn.net/wqc19920906
[RabbitMQ_Java_RabbitMQ]: https://blog.csdn.net/leixiaotao_java/article/details/78924863
[https_download.csdn.net_download_leixiaotao_java_10177767]: https://download.csdn.net/download/leixiaotao_java/10177767
[Link 1]: https://www.baidu.com/s?wd=%E6%9C%8D%E5%8A%A1%E5%99%A8&tn=24004469_oem_dg&rsv_dl=gh_pl_sl_csd
[Link 2]: https://www.baidu.com/s?wd=%E7%9B%91%E5%90%AC%E5%99%A8&tn=24004469_oem_dg&rsv_dl=gh_pl_sl_csd