16-RabbitMQ高级特性-消费端的消息ACK与重回队列

标签：false 16 ACK RabbitMQ 队列 import 重回 channel

消费端的消息ACK与重回队列

消费端的手工ACK和NACK

ACK分为自动和手动
消费端进行消费的时候, 如果由于业务异常我们可以进行日志的记录, 然后进行补偿
如果由于服务器宕机等严重问题, 那我们就需要手工进行ACK保障消费端消费成功

消费端的重回队列

消费端重回队列是为了对没有处理成功的消息, 把消息重新会递给Broker
一般我们在实际应用中, 都会关闭重回队列, 也就是设置为FALSE
为什么不使用重回队列的功能呢, 因为消息重回队列会加入到队列的尾部, 也会造成一条甚至大量消息一直重复投递在队列中死循环
说道这里, 其实我是真实碰到过的, 当时正是双11, 我们的失败策略就是用的重回队列, 导致有大量的消息一直因为业务的异常, 重回队列, 导致了4000万的订单MQ消息, 一直压力下不去, 差点被领导骂死~, 后面还做了重大事故回顾会议, 哎

消息重回队列代码实现

消费者

package com.dance.redis.mq.rabbit.rqueue;

import com.dance.redis.mq.rabbit.RabbitMQHelper;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeUnit;
 
public class Receiver {
 
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQHelper.getChannel();
        String queueName = "test001";
        RabbitMQHelper.queueDeclare(channel,queueName,true);
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body)
                    throws IOException {
                try {
                    System.out.println("receive message:" + new String(body) + ", RoutingKey: " + envelope.getRoutingKey());
                    /**
                     * 手工ACK
                     * 消费失败, 但是消息不重回队列
                     *  channel.basicNack(envelope.getDeliveryTag(), false, false);
                     * 消费失败, 将消息重新丢回消息队列尾部
                     *  channel.basicNack(envelope.getDeliveryTag(), false, true);
                     * 消费成功
                     *  channel.basicAck(envelope.getDeliveryTag(), false);
                     */
                    if((Integer)properties.getHeaders().get("flag") == 0) {
                        //throw new RuntimeException("异常");
                        // 设置为false表示关闭重回队列
                        channel.basicNack(envelope.getDeliveryTag(), false, false);
                        // 设置为true表示开启重回队列 将这条消息重回放入队列
//                        channel.basicNack(envelope.getDeliveryTag(), false, true);
                    } else {
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        //    参数：队列名称、是否自动ACK、Consumer
        channel.basicConsume(queueName, false, consumer);
        // 等待回调函数执行完毕之后，关闭资源。
        TimeUnit.SECONDS.sleep(50);
        channel.close();
        RabbitMQHelper.closeConnection();
    }
}

生产者

package com.dance.redis.mq.rabbit.rqueue;

import com.dance.redis.mq.rabbit.RabbitMQHelper;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;

import java.util.HashMap;
import java.util.Map;

public class Sender {

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQHelper.getChannel();
        //4 声明
        String queueName = "test001";
        RabbitMQHelper.queueDeclare(channel, queueName, true);
        for (int i = 0; i < 5; i++) {
            String msg = "Hello World RabbitMQ " + i;
            Map<String, Object> headers = new HashMap<>();
            headers.put("flag", i);
            AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                    .deliveryMode(2)
                    .contentEncoding("UTF-8")
                    .headers(headers).build();
            channel.basicPublish("", queueName, props, msg.getBytes());
        }
    }

}