RabbitMQ——消息确认
一、消息确认机制
生产者发送的消息，可能有以下两种情况：
1> 消息消费成功
2> 消息消费失败
为了保证消息可靠的到达消费者（！！！注意：消息确认机制和前面的工作模式中的publisher confirms模式有很大区别，消息确认保证的是消息可靠的到达消费者，而publisher confirms保证的是消息可靠的到达RabbitMQServer），RabbitMQ引入了消息确认机制：
消费者在消费消息时，可以指定autoAck参数，对应着两种确认方式：
（1）自动确认：消息只要到达消费者就会自动确认，不会考虑消费者是否正确消费了这些消息，直接从 内存/磁盘 中删除消息；
（2）手动确认：消息到达消费者，不会自动确认，会等待消费者调用Basic.Ack命令，才会从内存/磁盘 移除这条消息。
DefaultConsumer consumer = new DefaultConsumer(channel) {@Overridepublic 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);
如果将basicConsume中的true改为false,对于RabbitMQ服务器来说，消息分成两个部分：
1> 未发送给消费者的消息；
2> 已经发送给消费者，但还没有被确认的消息。
对应管理界面：
二、手动确认方法
RabbitMQ提供了两种手动确认，分别是肯定确认和否定确认，对应着3个方法:
（1）肯定确认：Channel.basicAck(long deliveryTag, boolean multiple)
其中，deliveryTag为消息的唯一表示，在每一个channel中都是唯一的，相当于TCP协议中的序号的作用，用来确认哪条消息已经收到，multiple为true表示是否批量确认，同样与TCP中确认序号的作用类似，表示在这个deliveryTag前的消息都已收到，为false则表示一次只确认一条消息。
（2）否定确认： Channel.basicReject(long deliveryTag, boolean requeue)
如果消息到达消费者后，消费者未正确处理这条消息（如发生异常），就可以通过这个方法进行否定确认，其中，参数requeue表示这条消息是否需要重新入队，这个方法一次只能确认一条消息。
（3）否定确认：Channel.basicNack(long deliveryTag, boolean multiple, boolean requeue)
与basicAck一样，这个方法可以通过multiple来实现批量确认
三、使用Spring Boot演示消息确认机制
演示之前，需要先了解Spring-AMQP的确认机制，它和RabbitMQ JDK Client库的确认机制有些许不同：
1. AcknowledgeMode.NONE
和RabbitMQ JDK Client库的自动确认机制一样，只要消息到达消费者，这条消息就会从队列中移除。
2. AcknowledgeMode.AUTO（和JDK Client库的区别）
消息到达消费者且处理成功，才会自动确认，如果消息在处理过程中发生了异常，则不会自动确认。
3. AcknowledgeMode.MANUAL
和JDK Client库一样，属于手动确认机制，同样分为肯定确认和否定确认。
接下来，进行准备工作，创建一个Spring Boot项目，添加RabbitMQ依赖：
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>添加RabbitMQ相关配置：
spring:rabbitmq:addresses: amqp://study:study@110.41.17.130:5672/extensionlistener:simple:acknowledge-mode: none #表示当前确认机制未AcknowledgeMode.NONE添加下图的包，创建对应的类：
3.1 AcknowledgeMode.NONE
（1）编写常量类（由于只是验证消息确认机制，可以随便使用一种工作模式）
public class Constants {public static final String ACK_QUEUE = "ack.queue";public static final String ACK_EXCHANGE = "ack.exchange"; }
（2）配置声明队列、交换机、交换机与队列绑定关系
@Configuration public class RabbitMQConfig {//1.声明队列@Bean("ackQueue")public Queue ackQueue(){return QueueBuilder.durable(Constants.ACK_QUEUE).build();}//2.声明交换机@Bean("ackExchange")public FanoutExchange ackExchange(){return ExchangeBuilder.fanoutExchange(Constants.ACK_EXCHANGE).build();}//3.声明队列与交换机的绑定关系@Beanpublic Binding ackBinding(@Qualifier("ackExchange") FanoutExchange fanoutExchange,@Qualifier("ackQueue") Queue queue){return BindingBuilder.bind(queue).to(fanoutExchange);} }
（3）编写生产者代码
@RestController @RequestMapping("/producer") public class ProducerController {@Resourceprivate RabbitTemplate rabbitTemplate;@RequestMapping("/ack")public String ack(){rabbitTemplate.convertAndSend(Constants.ACK_EXCHANGE,"","consumer ack mode test...");return "发送消息成功";} }
（4）编写消费者代码
@Component public class AckListener {@RabbitListener(queues = Constants.ACK_QUEUE)public void ackListener(Message message, Channel channel) throws UnsupportedEncodingException {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 = 10/0;System.out.println("业务处理完成");} }
（5）测试正常消费消息的情况
（6）测试消息消费异常情况（将消费者代码中得 int num = 10/0 注解打开）
（7）总结
经过确认，可以发现AcknowledgeMode.NONE机制，无论消费者是否正确消费消息，都会自动确认，不会保留异常消费的消息
3.2 AcknowledgeMode.AUTO
（1） 修改配置文件中的 acknowledge-mode: none 为 acknowledge-mode: auto
（2）演示消息正常消费的情况（将 int num = 10/0 注释）
@RabbitListener(queues = Constants.ACK_QUEUE)public void ackListener(Message message, Channel channel) throws UnsupportedEncodingException {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 = 10/0;System.out.println("业务处理完成");}运行程序，访问 producer/ack 接口发送消息：
（3）演示消息处理异常情况（取消 int num = 10/0 的注释）
@RabbitListener(queues = Constants.ACK_QUEUE)public void ackListener(Message message, Channel channel) throws UnsupportedEncodingException {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 = 10/0;System.out.println("业务处理完成");}运行程序，访问生产者端口：
消息没有成功消费，如果此时我们修改代码（将 int num = 10/0 注释，使程序不再发生异常），再次运行程序，队列中保存的消息就会被正常消费：
3.3 AcknowledgeMode.MANUAL
(1) 修改配置文件中 acknowledge-mode: auto 为 acknowledge-mode: manul
（2）修改消费者代码（由于是手动确认，需要在代码中添加正常消费时的 “肯定确认” 和 消费异常时的 “否定确认”）
@RabbitListener(queues = Constants.ACK_QUEUE)public void ackListener(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 = 10/0;System.out.println("业务处理完成");//消息正常消费，肯定确认channel.basicAck(deliveryTag,false);//单条确认}catch (Exception e){//消息消费异常，否定确认channel.basicNack(deliveryTag,false,true);//单条否定，异常后重新入队}}
（3）消息正常消费时的情况（注释 int num = 10/0）
接下来注释掉手动肯定确认的这行代码，看看会发生什么情况：
//消息正常消费，肯定确认//channel.basicAck(deliveryTag,false);//单条确认再次运行程序并通过接口 producer/ack 发送消息：
可以看到，在AcknowledgeMode.MANUAL机制下，如果不手动确认，队列不会移除已经被正常消费的消息：
（3）消息消费异常的情况下（取消 int num = 10/0 的注释）
接下来注释掉channel.basicNack,运行程序，访问接口发送消息：
如果将参数requeue置为false，会怎么样？
channel.basicNack(deliveryTag,false,false);//单条否定，重新发送消息运行程序：
（4）总结
1> 无论是否正常处理消息都要进行手动确认；
2> 正常处理消息但未手动确认，管理界面中的队列会有 一条/多条 Unacked 的消息（重新启动程序后会重新消费）；
3> 异常处理消息且未手动确认，也会有 一条/多条 Unacked 的消息（重启程序同样重新消费）；
4> 如果异常处理，将requeue置为false，队列不会保存 这条/多条 异常消费的消息
置为true，队列会不断重新发送 这条/多条 消息
