详解RabbitMQ高级特性之消息确认机制_spring amqp 消息自动确认配置
目录
消息确认机制
RabbitMQ的消息确认机制
自动确认
手动确认
Spring-AMQP的消息确认机制
代码演示
常量类
声明队列和交换机并绑定二者关系
声明RabbitTemplate
编写生产消息代码
AcknowledgeMode.NONE（演示）
编写消费消息代码1
编辑
编写消费消息代码2
AcknowledgeMode.AUTO（演示）
编写消费消息代码1
编写消费消息代码2
AcknowledgeMode.MANUAL（演示）
编写消费消息代码1
编写消费消息代码2
编写消费消息代码3
编写消费消息代码4
消息确认机制
RabbitMQ的消息确认机制
⽣产者发送消息之后,到达消费端之后,可能会有以下情况:
a. 消息处理成功
b. 消息处理异常
RabbitMQ向消费者发送消息之后, 就会把这条消息删掉, 那么第两种情况, 就会造成消息丢失.
那么如何确保消费端已经成功接收了, 并正确处理了呢?
为了保证消息从队列可靠地到达消费者, RabbitMQ提供了消息确认机制（message
acknowledgement)。
消费者在订阅队列时，可以指定 autoAck 参数, 根据这个参数设置, 消息确认机制分为以下两种:
• ⾃动确认: 当autoAck 等于true时, RabbitMQ 会⾃动把发送出去的消息置为确认, 然后从内存(或者磁盘)中删除, ⽽不管消费者是否真正地消费到了这些消息. ⾃动确认模式适合对于消息可靠性要求不⾼的场景.
• ⼿动确认: 当autoAck等于false时，RabbitMQ会等待消费者显式地调⽤Basic.Ack命令, 回复确认信号后才从内存(或者磁盘) 中移去消息. 这种模式适合对消息可靠性要求⽐较⾼的场景.
自动确认
源代码：
/** * Start a non-nolocal, non-exclusive consumer, with * a server-generated consumerTag. * @param queue the name of the queue * @param autoAck true if the server should consider messages * acknowledged once delivered; false if the server should expect * explicit acknowledgements * @param callback an interface to the consumer object * @return the consumerTag generated by the server * @throws java.io.IOException if an error is encountered * @see com.rabbitmq.client.AMQP.Basic.Consume * @see com.rabbitmq.client.AMQP.Basic.ConsumeOk * @see #basicConsume(String, boolean, String, boolean, boolean, Map, Consumer) */String basicConsume(String queue, boolean autoAck, Consumer callback) throwsIOException;代码示例：
DefaultConsumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println(\"接收到消息: \" + new String(body)); }};channel.basicConsume(Constants.TOPIC_QUEUE_NAME1, true, consumer);当autoAck参数置为false, 对于RabbitMQ服务端⽽⾔, 队列中的消息分成了两个部分：
⼀是等待投递给消费者的消息.
⼆是已经投递给消费者, 但是还没有收到消费者确认信号的消息.
如果RabbitMQ⼀直没有收到消费者的确认信号, 并且消费此消息的消费者已经断开连接, 则RabbitMQ会安排该消息重新进⼊队列，等待投递给下⼀个消费者，当然也有可能还是原来的那个消费者.
从RabbitMQ的Web管理平台上, 也可以看到当前队列中Ready状态和Unacked状态的消息数。
Ready: 等待投递给消费者的消息数
Unacked: 已经投递给消费者, 但是未收到消费者确认信号的消息数
手动确认
消费者在收到消息之后, 可以选择确认, 也可以选择直接拒绝或者跳过, RabbitMQ也提供了不同的确认应答的⽅式, 消费者客⼾端可以调⽤与其对应的channel的相关⽅法, 共有以下三种：
1. 肯定确认: Channel.basicAck(long deliveryTag, boolean multiple)
RabbitMQ 已知道该消息并且成功的处理消息. 可以将其丢弃了.
参数说明:
1) deliveryTag: 消息的唯⼀标识，它是⼀个单调递增的64 位的⻓整型值. deliveryTag 是每个通道
（Channel）独⽴维护的, 所以在每个通道上都是唯⼀的. 当消费者确认(ack)⼀条消息时, 必须使⽤对应的通道上进⾏确认。
2) multiple: 是否批量确认. 在某些情况下, 为了减少⽹络流量, 可以对⼀系列连续的 deliveryTag 进
⾏批量确认. 值为 true 则会⼀次性 ack所有⼩于或等于指定 deliveryTag 的消息. 值为false, 则只确认当前指定deliveryTag 的消息。
deliveryTag 是RabbitMQ中消息确认机制的⼀个重要组成部分, 它确保了消息传递的可靠性和顺
序性。
2. 否定确认: Channel.basicReject(long deliveryTag, boolean requeue)
RabbitMQ在2.0.0版本开始引⼊了 Basic.Reject 这个命令, 消费者客⼾端可以调⽤
channel.basicReject⽅法来告诉RabbitMQ拒绝这个消息.
参数说明:
1) deliveryTag: 参考channel.basicAck
2) requeue: 表⽰拒绝后, 这条消息如何处理. 如果requeue 参数设置为true, 则RabbitMQ会重新将这条消息存⼊队列,以便可以发送给下⼀个订阅的消费者. 如果requeue参数设置为false, 则RabbitMQ会把消息从队列中移除, ⽽不会把它发送给新的消费者。
3. 否定确认: Channel.basicNack(long deliveryTag, boolean multiple, boolean requeue)
Basic.Reject命令⼀次只能拒绝⼀条消息,如果想要批量拒绝消息,则可以使⽤Basic.Nack这个命令. 消费者客⼾端可以调⽤ channel.basicNack⽅法来实现。
multiple参数设置为true则表⽰拒绝deliveryTag编号之前所有未被当前消费者确认的消息.
Spring-AMQP的消息确认机制
下面我们基于SpringBoot来演示消息确认机制，不过和RabbitMQ Java Client库有一定差异。
Spring-AMQP 对消息确认机制提供了三种策略.
消息确认机制的三种策略的解读：
1. AcknowledgeMode.NONE
这种模式下, 消息⼀旦投递给消费者, 不管消费者是否成功处理了消息, RabbitMQ 就会⾃动确认消息, 从RabbitMQ队列中移除消息. 如果消费者处理消息失败, 消息可能会丢失.
2. AcknowledgeMode.AUTO(默认)
这种模式下, 消费者在消息处理成功时会⾃动确认消息, 但如果处理过程中抛出了异常, 则不会确认消息.
3. AcknowledgeMode.MANUAL
⼿动确认模式下, 消费者必须在成功处理消息后显式调⽤ basicAck ⽅法来确认消息. 如果消
息未被确认, RabbitMQ 会认为消息尚未被成功处理, 并且会在消费者可⽤时重新投递该消息, 这种模式提⾼了消息处理的可靠性, 因为即使消费者处理消息后失败, 消息也不会丢失, ⽽是可以被重新处理.
代码演示
常量类
public class Constants { public static final String ACK_QUEUE = \"ack.queue\"; public static final String ACK_EXCHANGE = \"ack.exchange\";}声明队列和交换机并绑定二者关系
import org.springframework.amqp.core.*;import org.springframework.beans.factory.annotation.Qualifier;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import rabbitextensionsdemo.constant.Constants;@Configurationpublic class RabbitMQConfig { //消息确认 @Bean(\"ackQueue\") public Queue ackQueue() { return QueueBuilder.durable(Constants.ACK_QUEUE).build(); } @Bean(\"directExchange\") public DirectExchange directExchange() { return ExchangeBuilder.directExchange(Constants.ACK_EXCHANGE).build(); } // @Bean(\"ackBinding\")// public Binding ackBinding(Exchange directExchange, Queue queue){// return BindingBuilder.bind(queue).to(directExchange).with(\"ack\").noargs();// } @Bean(\"ackBinding\") public Binding ackBinding(@Qualifier(\"directExchange\") DirectExchange directExchange, @Qualifier(\"ackQueue\") Queue queue) { return BindingBuilder.bind(queue).to(directExchange).with(\"ack\"); }}声明RabbitTemplate
import org.springframework.amqp.rabbit.connection.ConnectionFactory;import org.springframework.amqp.rabbit.core.RabbitTemplate;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;@Configurationpublic class RabbitTemplateConfig { @Bean public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){ RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory); return rabbitTemplate; }}编写生产消息代码
import jakarta.annotation.Resource;import org.springframework.amqp.rabbit.core.RabbitTemplate;import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.RestController;import rabbitextensionsdemo.constant.Constants;@RequestMapping(\"/producer\")@RestControllerpublic class ProducerController { @Resource(name = \"rabbitTemplate\") private RabbitTemplate rabbitTemplate; @Resource(name = \"confirmRabbitTemplate\") private RabbitTemplate confirmRabbitTemplate; @Resource(name = \"transRabbitTemplate\") private RabbitTemplate transRabbitTemplate; @RequestMapping(\"/ack\") public String ack() { rabbitTemplate.convertAndSend(Constants.ACK_EXCHANGE, \"ack\", \"consumer ack mode test...\"); return \"消息发送成功\"; }}AcknowledgeMode.NONE（演示）
添加配置
spring: application: name: rabbit-extensions-demo rabbitmq: addresses: amqp://study:study@47.98.109.138:5672/extension listener: simple: acknowledge-mode: none
编写消费消息代码1
import com.rabbitmq.client.Channel;import org.springframework.amqp.core.Message;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Component;import rabbitextensionsdemo.constant.Constants;@Componentpublic class AckListener { @RabbitListener(queues = Constants.ACK_QUEUE) public void handMessage(Message message, Channel channel) throws UnsupportedEncodingException { //消费者逻辑 System.out.printf(\"接收到消息: %s, deliveryTag: %d \\n\", new String(message.getBody(),\"UTF-8\"), message.getMessageProperties().getDeliveryTag()); //进行业务逻辑处理 System.out.println(\"业务逻辑处理\"); System.out.println(\"业务处理完成\"); }}发送消息
消费消息
编写消费消息代码2
import com.rabbitmq.client.Channel;import org.springframework.amqp.core.Message;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Component;import rabbitextensionsdemo.constant.Constants;@Componentpublic class AckListener { @RabbitListener(queues = Constants.ACK_QUEUE) public void handMessage(Message message, Channel channel)throws UnsupportedEncodingException { //消费者逻辑 System.out.printf(\"接收到消息: %s, deliveryTag: %d \\n\", new String(message.getBody(),\"UTF-8\"), message.getMessageProperties().getDeliveryTag()); //进行业务逻辑处理 System.out.println(\"业务逻辑处理\"); int num = 3/0; System.out.println(\"业务处理完成\"); }}发送消息
消费消息
此时我们可以看到，消息虽然被消费者收到了，但是因为消费时发生异常导致消息没有被正常消费，而且队列中的消息也已经没有了。
AcknowledgeMode.AUTO（演示）
添加配置
spring: application: name: rabbit-extensions-demo rabbitmq: addresses: amqp://study:study@47.98.109.138:5672/extension listener: simple: acknowledge-mode: auto
编写消费消息代码1
import com.rabbitmq.client.Channel;import org.springframework.amqp.core.Message;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Component;import rabbitextensionsdemo.constant.Constants;@Componentpublic class AckListener { @RabbitListener(queues = Constants.ACK_QUEUE) public void handMessage(Message message, Channel channel) throws UnsupportedEncodingException { //消费者逻辑 System.out.printf(\"接收到消息: %s, deliveryTag: %d \\n\", new String(message.getBody(),\"UTF-8\"), message.getMessageProperties().getDeliveryTag()); //进行业务逻辑处理 System.out.println(\"业务逻辑处理\"); System.out.println(\"业务处理完成\"); }}生产消息
消费消息
编写消费消息代码2
import com.rabbitmq.client.Channel;import org.springframework.amqp.core.Message;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Component;import rabbitextensionsdemo.constant.Constants;@Componentpublic class AckListener { @RabbitListener(queues = Constants.ACK_QUEUE) public void handMessage(Message message, Channel channel)throws UnsupportedEncodingException { //消费者逻辑 System.out.printf(\"接收到消息: %s, deliveryTag: %d \\n\", new String(message.getBody(),\"UTF-8\"), message.getMessageProperties().getDeliveryTag()); //进行业务逻辑处理 System.out.println(\"业务逻辑处理\"); int num = 3/0; System.out.println(\"业务处理完成\"); }}生产消息
消费消息
此时我们可以看到，消费消息时发生异常，导致消息没有被正常消费，而且此时控制台不断地打印日志，deleverTag的值不断增加，观察控制界面可以看到队列中始终有一条消息，是因为AUTO机制下，如果消费消息时发生异常，此时消费方不会确认消息，此时消息就会重新入队，直到消费方能够正常消费消息而不发生异常。
AcknowledgeMode.MANUAL（演示）
添加配置
spring: application: name: rabbit-extensions-demo rabbitmq: addresses: amqp://study:study@47.98.109.138:5672/extension listener: simple: acknowledge-mode: manual
编写消费消息代码1
import com.rabbitmq.client.Channel;import org.springframework.amqp.core.Message;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Component;import rabbitextensionsdemo.constant.Constants;@Componentpublic class AckListener { @RabbitListener(queues = Constants.ACK_QUEUE) public void handMessage(Message message, Channel channel) throws Exception { long deliveryTag = message.getMessageProperties().getDeliveryTag(); //消费者逻辑 System.out.printf(\"接收到消息: %s, deliveryTag: %d \\n\", new String(message.getBody(),\"UTF-8\"), message.getMessageProperties().getDeliveryTag()); //进行业务逻辑处理 System.out.println(\"业务逻辑处理\"); System.out.println(\"业务处理完成\"); }}生产消息
消费消息
编写消费消息代码2
import com.rabbitmq.client.Channel;import org.springframework.amqp.core.Message;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Component;import rabbitextensionsdemo.constant.Constants;@Componentpublic class AckListener { @RabbitListener(queues = Constants.ACK_QUEUE) public void handMessage(Message message, Channel channel) throws Exception { long deliveryTag = message.getMessageProperties().getDeliveryTag(); //消费者逻辑 System.out.printf(\"接收到消息: %s, deliveryTag: %d \\n\", new String(message.getBody(),\"UTF-8\"), message.getMessageProperties().getDeliveryTag()); //进行业务逻辑处理 System.out.println(\"业务逻辑处理\"); int num = 3/0; System.out.println(\"业务处理完成\"); }}发送消息
消费消息
此时我们看到，MANUAL消息确认机制下，消费方消费消息时发生异常，没有手动确认，会导致消息在队列中的状态变为Unacked状态。
编写消费消息代码3
import com.rabbitmq.client.Channel;import org.springframework.amqp.core.Message;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Component;import rabbitextensionsdemo.constant.Constants;@Componentpublic class AckListener { @RabbitListener(queues = Constants.ACK_QUEUE) public void handMessage(Message message, Channel channel) throws Exception { long deliveryTag = message.getMessageProperties().getDeliveryTag(); try { //消费者逻辑 System.out.printf(\"接收到消息: %s, deliveryTag: %d \\n\", new String(message.getBody(),\"UTF-8\"), message.getMessageProperties().getDeliveryTag()); //进行业务逻辑处理 System.out.println(\"业务逻辑处理\"); System.out.println(\"业务处理完成\"); //肯定确认 channel.basicAck(deliveryTag,false); } catch (Exception e) { //否定确认 channel.basicNack(deliveryTag, false, true); } }}生产消息
消费消息
此时我们可以看到，在MANUAL消息确认机制下，如果消息手动确认了，就没问题了。
编写消费消息代码4
import com.rabbitmq.client.Channel;import org.springframework.amqp.core.Message;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Component;import rabbitextensionsdemo.constant.Constants;@Componentpublic class AckListener { @RabbitListener(queues = Constants.ACK_QUEUE) public void handMessage(Message message, Channel channel) throws Exception { long deliveryTag = message.getMessageProperties().getDeliveryTag(); try { //消费者逻辑 System.out.printf(\"接收到消息: %s, deliveryTag: %d \\n\", new String(message.getBody(),\"UTF-8\"), message.getMessageProperties().getDeliveryTag()); //进行业务逻辑处理 System.out.println(\"业务逻辑处理\"); int num = 3/0; System.out.println(\"业务处理完成\"); //肯定确认 channel.basicAck(deliveryTag,false); } catch (Exception e) { //否定确认 channel.basicNack(deliveryTag, false, true); } }}生产消息
消费消息
此时我们可以看到，当消费消息时，因为发生异常，程序会运行到拒绝消息，但是因为设置了重新入队，会导致控制台不停地打印消息，deleverTag的值不断增加。
