RabbitMQ-死信队列

Posted 2023-03-03 weixin_43956692

  一、介绍

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

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

  二、来源

         1、消息TTL过期

         2、队列达到最大长度（队列满了，无法再添加数据到mq中）

         3、消息被拒绝（basic.reject或basic.nack）并且requeue=false

  三、代码

        生产者

public class Producer 
    //普通交换机的名称
    public static final String NORMAL_EXCHANGE= "normal_exchange";

    public static void main(String[] args) throws Exception 
        Channel channel = RabbitMqUtil.getChannel();
        //延迟消息 设置TTL
        AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().expiration("10000").build();
        for (int i = 0; i < 11; i++) 
            String message = i+"";
            channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",properties,message.getBytes());
        
    

        正常消费者

public class Consumer01 
    //普通交换机的名称
    public static final String NORMAL_EXCHANGE= "normal_exchange";
    //死信交换机
    public static final String DEAL_EXCHANGE= "deal_exchange";
    //普通队列名称
    public static final String NORMAL_QUEUE = "normal_queue";
    //死信队列名称
    public static final String DEAL_QUEUE= "deal_queue";

    public static void main(String[] args) throws Exception 
        Channel channel = RabbitMqUtil.getChannel();
        //声明死信和普通交换机类型为direct
        channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        channel.exchangeDeclare(DEAL_EXCHANGE, BuiltinExchangeType.DIRECT);
        //声明普通和死信队列
        Map<String, Object> arguments = new HashMap<>();
        //过期时间
        //arguments.put("x-message-ttl",10000);
        //正常队列设置死信交换机
        arguments.put("x-dead-letter-exchange",DEAL_EXCHANGE);
        //设置死信routingKey
        arguments.put("x-dead-letter-routing-key","lisi");
        channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);
        //============================================
        channel.queueDeclare(DEAL_QUEUE,false,false,false,null);

        //绑定普通的交换机和队列
        channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");
        //绑定死信的交换机与队列
        channel.queueBind(DEAL_QUEUE,DEAL_EXCHANGE,"lisi");
        System.out.println("等待接收消息......");
        //接收消息
        channel.basicConsume(NORMAL_QUEUE,false,(consumerTag,message)->
            System.out.println("consumer1接收的消息内容是："+new String(message.getBody()));
        ,(consumerTag)->

        );
    

        死信消费者

public class Consumer02 
    //死信队列名称
    public static final String DEAL_QUEUE= "deal_queue";

    public static void main(String[] args) throws Exception 
        Channel channel = RabbitMqUtil.getChannel();

        System.out.println("等待接收消息......");
        //接收消息
        channel.basicConsume(DEAL_QUEUE,false,(consumerTag,message)->
            System.out.println("consumer2接收的消息内容是："+new String(message.getBody()));
        ,(consumerTag)->

        );
    

  四、队列达到最大长度

       声明正常队列时添加如下参数，并删掉旧队列重新创建 

 //设置正常队列长度限制
 arguments.put("x-max-length",6);

  五、消息被拒

        将自动应答修改为手动应答

//接收消息
        channel.basicConsume(NORMAL_QUEUE,false,(consumerTag,message)->
            String msg = new String(message.getBody());
            if(msg.equals("5"))
                System.out.println("consumer1拒绝的消息内容是："+msg);
                channel.basicReject(message.getEnvelope().getDeliveryTag(),false);
            else
                System.out.println("consumer1接收的消息内容是："+msg);
                channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
            

        ,(consumerTag)->

        );