RabbitMq 手动ACK

1、消费者消费消息

如果我们设置的是自动确认的话，那么就会有当接收到消息，去处理消息的时候生成异常，那么就会有消息丢失的情况。这种异常可能是网络波动，服务器宕机等情况都会发生。

并且如果没有ACK 的话会导致mq 中的unacked 的数量飙升，可能会撑爆内存

所以大部分项目中都会去使用手动ACK。下面介绍具体步骤。

因为这里的ACK说的是消费者向服务端返回的ACK确认，所以下面是消费方的配置和程序

1、关键依赖，

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
            <version>2.3.9.RELEASE</version>
        </dependency>

2.消费者yml配置开启手动ACK的开关

spring:
  rabbitmq:
    host: 192.26.196.131
    port: 5672
    username: admin
    password: admin
    listener:
      simple:
        acknowledge-mode: manual #开启手动ACK
        prefetch: 1

3.消费方的程序编写

package com.my.test;
 
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;
 
import java.io.IOException;
 
@Component
public class MqConsumer {
    @RabbitListener(queues = "test_queue")
    public void processHandler(String msg, Channel channel, Message message) {
 
        try {
 
            System.out.println("收到消息"+msg);
             //具体的业务逻辑处理，数据处理，入库等操作

            System.out.println("确认消息");
            //手动签收
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }  catch (Exception e) {
            //是否是重新回到队列的消息
            //这里有一个问题就是，代码错误的话回无限的重新接受消息，从而是cpu急剧上升
            System.out.println("消息即将再次返回队列处理...");
        try {
             //第一个boolean：是否允许多条处理  
            //第二个boolean：是否重新回到队列
        channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
                } catch (IOException e) {
                    e.printStackTrace();
                }
 
        }
    }
 
}
 

3、问题

业务代码一旦出现BUG，多数情况不会自动修复，一条消息会被无限投递进队列，消费端一直接受处理--->然后处理失败--->重新回到队列，导致了死循环。

解决思路

1、 将失败的消息拒绝之后，重新发送到队列的尾部，这样可以保证其他的正常的消息是可以被处理的。

2.设置消息重试次数，达到了重试上限以后，队列删除此消息，并将消息持久化到数据库并推送报警，然后人工处理和定时任务补偿。

//确认消息
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
//channel对象重新发送消息到队尾
channel.basicPublish(message.getMessageProperties().getReceivedExchange(),
                    message.getMessageProperties().getReceivedRoutingKey(), MessageProperties.PERSISTENT_TEXT_PLAIN,
                    JSON.toJSONBytes(msg));