RabbitMQ消费端ACK与重回队列机制

消费端的手工ACK与NACK

当我们设置 autoACK=false 时，就可以使用手工ACK方式了，那么其实手工方式包括了手工ACK与NACK。
当我们手工 ACK 时，会发送给Broker一个应答，代表消息成功处理了，Broker就可以回送响应给生产端了。NACK 则表示消息处理失败了，如果设置重回队列，Broker端就会将没有成功处理的消息重新发送。

使用方式

1. 消费端进行消费的时候，如果由于业务异常我们可以手工 NACK 并进行日志的记录，然后进行补偿！
   方法：void basicNack(long deliveryTag, boolean multiple, boolean requeue)
2. 如果由于服务器宕机等严重问题，那我们就需要手工进行 ACK 保障消费端消费成功！
   方法：void basicAck(long deliveryTag, boolean multiple)

消费端的重回队列

• 消费端重回队列是为了对没有处理成功的消息，把消息重新会递给Broker！
• 重回队列，会把消费失败的消息重新添加到队列的尾端，供消费者继续消费。
• 一般我们在实际应用中，都会关闭重回队列，也就是设置为false

演示重回队列

生产端

对消息设置自定义属性以便进行区分

public class Producer {

    public static void main(String[] args) throws Exception {
        //1 创建ConnectionFactorys
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.11.76");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        //2 获取Connection
        Connection connection = connectionFactory.newConnection();
        //3 通过Connection创建一个新的Channel
        Channel channel = connection.createChannel();
        
        String exchange = "test_ack_exchange";
        String routingKey = "ack.save";
        
        for(int i =0; i<5; i ++){
            //设置消息属性
            Map<String, Object> headers = new HashMap<String, Object>();
            headers.put("num", i);
            AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                    .deliveryMode(2)
                    .contentEncoding("UTF-8")
                    .headers(headers)
                    .build();
            //发送消息
            String msg = "Hello RabbitMQ ACK Message " + i;
            channel.basicPublish(exchange, routingKey, true, properties, msg.getBytes());
        }   
    }
}

自定义消费者

对第一条消息进行NACK，并设置重回队列

public class MyConsumer extends DefaultConsumer {

    private Channel channel ;

    public MyConsumer(Channel channel) {
        super(channel);
        this.channel = channel;
    }

    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        System.err.println("-----------consume message----------");
        System.err.println("body: " + new String(body));
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        if((Integer)properties.getHeaders().get("num") == 0) {
            //NACK，参数三requeue：是否重回队列
            channel.basicNack(envelope.getDeliveryTag(), false, true);
        } else {
            channel.basicAck(envelope.getDeliveryTag(), false);
        }
    }
}

消费端

关闭自动签收功能

public class Consumer {
    
    public static void main(String[] args) throws Exception {
        
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.11.76");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();
        
        String exchangeName = "test_ack_exchange";
        String queueName = "test_ack_queue";
        String routingKey = "ack.#";
        //声明交换机和队列，然后进行绑定设置路由Key
        channel.exchangeDeclare(exchangeName, "topic", true, false, null);
        channel.queueDeclare(queueName, true, false, false, null);
        channel.queueBind(queueName, exchangeName, routingKey);
        
        //手工签收 必须要设置 autoAck = false
        channel.basicConsume(queueName, false, new MyConsumer(channel));
    }
}

运行说明

先启动消费端，然后启动生产端，消费端打印如下，显然第一条消息由于我们调用了NACK，并且设置了重回队列，所以会导致该条消息一直重复发送，消费端就会一直循环消费。

-----------consume message----------
body: Hello RabbitMQ ACK Message 0
-----------consume message----------
body: Hello RabbitMQ ACK Message 1
-----------consume message----------
body: Hello RabbitMQ ACK Message 2
-----------consume message----------
body: Hello RabbitMQ ACK Message 3
-----------consume message----------
body: Hello RabbitMQ ACK Message 4
-----------consume message----------
body: Hello RabbitMQ ACK Message 0
-----------consume message----------
body: Hello RabbitMQ ACK Message 0
-----------consume message----------

一般工作中不会设置重回队列这个属性，我们都是自己去做补偿或者投递到延迟队列里的，然后指定时间去处理即可。