[大数据]rabbitmq 发布确认及死信队列

发布确认

发布确认原理

开启发布确认的方法

发布确认默认是没有开启的，如果要开启需要调用方法 confirmSelect，每当你要想使用发布确认，都需要在channel 上调用该方法

Connection connection = RabbitmqUtil.getConnection();
Channel channel = connection.createChannel();
//开启发布确认
channel.confirmSelect();

单个发布确认

这是一种简单的确认方式，它是一种同步确认发布的方式，也就是发布一个消息之后只有它被确认发布，后续的消息才能继续发布, waitForConfirmsOrDie(long)这个方法只有在消息被确认的时候才返回，如果在指定时间范围内这个消息没有被确认那么它将抛出异常。

这种确认方式有一个最大的缺点就是:发布速度特别的慢，因为如果没有确认发布的消息就会阻塞所有后续消息的发布，这种方式最多提供每秒不超过数百条发布消息的吞吐量。当然对于某些应用程序来说这可能已经足够了。

  /**
     * 单个确认
     */
    public static void publicMessageSingle() throws IOException, InterruptedException {
        Connection connection = RabbitmqUtil.getConnection();
        Channel channel = connection.createChannel();
        //队列声明
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,true,false,false,null);
        //开启发布确认
        channel.confirmSelect();
        //开始时间
        long begin = System.currentTimeMillis();
        //批量发消息
        for (int i = 0; i < 1000; i++) {
            String message = i+"";
            channel.basicPublish("",queueName,null,message.getBytes());
            //单个消息立马确认
            boolean flag = channel.waitForConfirms();
            if (flag) {
                System.out.println("消息发送成功");
            }
        }
        //结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布1000条单独确认消息,耗时"+(end - begin)+"ms");
    }

批量确认发布

上面那种方式非常慢，与单个等待确认消息相比，先发布一批消息然后一起确认可以极大地提高吞吐量，当然这种方式的缺点就是:当发生故障导致发布出现问题时，不知道是哪个消息出现问题了，我们必须将整个批处理保存在内存中，以记录重要的信息而后重新发布消息。当然这种方案仍然是同步的，也一样阻塞消息的发布。

 /**
     * 批量确认
     */
    public static void publicMessageBatch() throws IOException, InterruptedException {
        Connection connection = RabbitmqUtil.getConnection();
        Channel channel = connection.createChannel();
        //队列声明
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,true,false,false,null);
        //开启发布确认
        channel.confirmSelect();
        //开始时间
        long begin = System.currentTimeMillis();

        //批量确认消息大小
        int batchSize = 100;

        //批量发消息 批量发布确认
        for (int i = 1; i <= 1000; i++) {
            String message = i+"";
            channel.basicPublish("",queueName,null,message.getBytes());
            //判断达到100条消息的时候，批量确认一次
            if (i % batchSize == 0) {
                //发布确认
                channel.waitForConfirms();
            }
        }

        //结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布1000条批量确认消息,耗时"+(end - begin)+"ms");
        RabbitmqUtil.closeConnectionAndChannel(channel,connection);
    }

异步发布确认

异步确认虽然编程逻辑比上两个要复杂，但是性价比最高，无论是可靠性还是效率都没得说，他是利用回调函数来达到消息可靠性传递的,这个中间件也是通过函数回调来保证是否投递成功，下面就让我们来详细讲解异步确认是怎么实现的。

/**
     * 异步确认发布
     */
    public static void publicMessageAsync() throws IOException, InterruptedException {
        Connection connection = RabbitmqUtil.getConnection();
        Channel channel = connection.createChannel();
        //队列声明
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,true,false,false,null);
        //开启发布确认
        channel.confirmSelect();
        //开始时间
        long begin = System.currentTimeMillis();


        //消息确认成功 回调函数
        // 参数1消息的标记 2. 是否为批量确认
        ConfirmCallback ackCallBack = (deliveryTag,multiple)->{
            System.out.println("确认的消息"+deliveryTag);
        };

        //消息确认失败 回调函数
        ConfirmCallback nackCallBack = (deliveryTag,multiple)->{
            System.out.println("未确认的消息"+deliveryTag);
        };
        //准备消息的监听器 监听哪些消息成功了 哪些消息失败了 异步通知
        /**
         * 1.第一个监听哪些成功
         * 2.监听哪些消息失败
         */
        channel.addConfirmListener(ackCallBack,nackCallBack);

        //批量发消息 批量发布确认
        for (int i = 1; i <= 1000; i++) {
            String message = i+"";
            channel.basicPublish("",queueName,null,message.getBytes());
        }



        //结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布1000条批量确认消息,耗时"+(end - begin)+"ms");
//        RabbitmqUtil.closeConnectionAndChannel(channel,connection);
    }

如何处理异步未确认的消息

最好的解决的解决方案就是把未确认的消息放到一个基于内存的能被发布线程访问的队列，比如说用ConcurrentLinkedQueue这个队列在confirm callbacks与发布线程之间进行消息的传递。

/**
     * 异步确认发布
     */
    public static void publicMessageAsync() throws IOException, InterruptedException {
        Connection connection = RabbitmqUtil.getConnection();
        Channel channel = connection.createChannel();
        //队列声明
        String queueName = UUID.randomUUID().toString();
        channel.queueDeclare(queueName,true,false,false,null);
        //开启发布确认
        channel.confirmSelect();

        //线程安全有序的哈希表，适用于高并发的情况下
        /**
         * 1.轻松的奖序号与消息进行关联
         * 2.轻松批量删除条目，只要给到序号
         * 3.支持高并发。
         */
        ConcurrentSkipListMap<Long,Object> outStandingConfirms = new ConcurrentSkipListMap<>();

        //消息确认成功 回调函数
        // 参数1消息的标记 2. 是否为批量确认
        ConfirmCallback ackCallBack = (deliveryTag,multiple)->{
            if (multiple) {
                //获取已确认的消息
                ConcurrentNavigableMap<Long, Object> confirmed =
                        outStandingConfirms.headMap(deliveryTag);
                //删除已确认的消息
                confirmed.clear();
            } else {
                outStandingConfirms.remove(deliveryTag);
            }
            System.out.println("确认的消息"+deliveryTag);
        };

        //消息确认失败 回调函数
        ConfirmCallback nackCallBack = (deliveryTag,multiple)->{
            String msg = outStandingConfirms.get(deliveryTag).toString();
            System.out.println("未确认的消息"+msg);
        };
        //准备消息的监听器 监听哪些消息成功了 哪些消息失败了 异步通知
        /**
         * 1.第一个监听哪些成功
         * 2.监听哪些消息失败
         */
        channel.addConfirmListener(ackCallBack,nackCallBack);

        //开始时间
        long begin = System.currentTimeMillis();
        //批量发消息 批量发布确认
        for (int i = 1; i <= 1000; i++) {
            String message = i+"";
            channel.basicPublish("",queueName,null,message.getBytes());
            //此处记录下所有要发送的消息 消息的总和
            outStandingConfirms.put(channel.getNextPublishSeqNo(),message);
        }

        //结束时间
        long end = System.currentTimeMillis();
        System.out.println("发布1000条批量确认消息,耗时"+(end - begin)+"ms");
//        RabbitmqUtil.closeConnectionAndChannel(channel,connection);
    }

死信队列

死信的概念

先从概念解释上搞清楚这个定义，死信，顾名思义就是无法被消费的消息，字面意思可以这样理解，一般来说,producer将消息投递到 broker或者直接到queue里了，consumer 从 queue取出消息进行消费，但某些时候由于特定的原因导致queue中的某些消息无法被消费，这样的消息如果没有后续的处理，就变成了死信，有死信自然就有了死信队列。

应用场景:为了保证订单业务的消息数据不丢失，需要使用到RabbitMQ的死信队列机制，当消息消费发生异常时，将消息投入死信队列中.还有比如说:用户在商城下单成功并点击去支付后在指定时间未支付时自动失效

死信的来源

• 消息TIL过期
• 队列达到最大长度(队列满了，无法再添加数据到mq.中)
• 消息被拒绝(basic.reject或 basic.nack)并且requeue=false.

死信实战

消息TTL过期

消息过期后满足死信要求，消费者1若在10s内未能接受消息，则会将消息转移到死信队列，由消费者2来消费。

生产者

public class Producer {
   private static final String NORMAL_EXCHANGE = "normal_exchange";
   public static void main(String[] args) throws IOException {
       Connection connection = RabbitmqUtil.getConnection();
       Channel channel = connection.createChannel();
       //死信队列 设置TTL时间 单位ms
       AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();

       for (int i = 1; i <= 10; i++) {
           String message = "info" + i;
           channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",properties,message.getBytes());
       }
       RabbitmqUtil.closeConnectionAndChannel(channel,connection);
   }
}

消费者1

public class Consumer1 {
    //普通交换机名称
    public static final String NORMAL_EXCHANGE = "normal_exchange";
    //死信交换机名称
    public static final String DEAD_EXCHANGE = "dead_exchange";
    //普通队列名称
    public static final String NORMAL_QUEUE = "normal_queue";
    //死信队列名称
    public static final String DEAD_QUEUE = "dead_queue";


    public static void main(String[] args) throws IOException {
        Connection connection = RabbitmqUtil.getConnection();
        Channel channel = connection.createChannel();
        //声明死信和普通交换机，类型为direct
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);
        //声明普通队列
        Map<String,Object> arguments = new HashMap<>();
        //过期时间 10s = 100000ms
//        arguments.put("x-dead-letter-ttl",100000);
        //正常队列设置死信交换机
        arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);
        //设置死信RoutingKey
        arguments.put("x-dead-letter-routing-key","lisi");
        channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);

        // 声明死信队列
        channel.queueDeclare(DEAD_QUEUE,false,false,false,null);

        //绑定普通的交换机与队列
        channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");
        //绑定死信交换机与死信队列
        channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");
        System.out.println("等待接受消息...");

        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("consumer1接受的消息是"+new String(message.getBody(),"UTF-8"));
        };
        channel.basicConsume(NORMAL_QUEUE, true, deliverCallback,consumerTag -> {});
    }
}

消费者2

public class Consumer2 {
    //死信队列名称
    public static final String DEAD_QUEUE = "dead_queue";


    public static void main(String[] args) throws IOException {
        Connection connection = RabbitmqUtil.getConnection();
        Channel channel = connection.createChannel();
        System.out.println("等待接受消息...");
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            System.out.println("consumer2接受的消息是"+new String(message.getBody(),"UTF-8"));
        };
        channel.basicConsume(DEAD_QUEUE, true, deliverCallback,consumerTag -> { });
    }
}

队列达到最大长度

c1消费者修改如下代码

//正常队列设置死信交换机
arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);
//设置死信RoutingKey
arguments.put("x-dead-letter-routing-key","lisi");
//设置正常队列长度的限制
arguments.put("x-max-length",6);

生产者修改代码 去掉TTL限制

//死信队列 设置TTL时间 单位ms
//AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();
channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",null,message.getBytes());