【java开发笔记】RabbitMQ--超时队列和死信队列_java 死信队列

死信队列（Dead Letter Queue, DLQ）是一种特殊的消息队列，用于存储无法被正常消费的消息。

使用场景：

• 当消息满足某些条件（如被拒绝、TTL过期、队列满等）而无法被正常消费时，这些消息将被发送到死信队列中进行处理。
• 死信队列常用于处理消费者无法正确处理的消息，避免消息丢失，以及实现延迟消息处理或消息重试机制。

超时队列是一种用于处理需要在指定时间内被处理的消息队列。如果消息在设定的时间内没有被消费，则会被视为超时消息。

使用场景：

• 超时队列常用于处理有时间限制的任务，例如用户在商城下单后未在规定时间内支付，则该订单消息可以被放入超时队列进行处理。

RabbitMQ配置： 

常量定义队列的名称和交换机名称等。

    //延迟时间 单位毫秒
    private static final long DELAY_TIME = 1000 * 30;
    //超时队列名称
    public static final String STROKE_OVER_QUEUE = "STROKE_OVER_QUEUE";
    //死信队列名称
    public static final String STROKE_DEAD_QUEUE = "STROKE_DEAD_QUEUE";
    //超时队列交换器名称
    public static final String STROKE_OVER_QUEUE_EXCHANGE = "STROKE_OVER_QUEUE_EXCHANGE";
    //死信队列交换器名称
    public static final String STROKE_DEAD_QUEUE_EXCHANGE = "STROKE_DEAD_QUEUE_EXCHANGE";
    //超时交换器 ROUTINGKEY
    public static final String STROKE_OVER_KEY = "STROKE_OVER_KEY";
    //死信交换器 ROUTINGKEY
    public static final String STROKE_DEAD_KEY = "STROKE_DEAD_KEY";

声明超时队列，使用HashMap存储。

配置死信队列通常涉及设置x-dead-letter-exchange（死信交换机）和x-dead-letter-routing-key（死信路由键）

 @Bean
    public Queue strokeOverQueue() {
        Map<String, Object> args = new HashMap<>();
        // x-dead-letter-exchange    这里声明当前队列绑定的死信交换机
        args.put("x-dead-letter-exchange", STROKE_DEAD_QUEUE_EXCHANGE);
        // x-dead-letter-routing-key  这里声明当前队列的死信路由key
        args.put("x-dead-letter-routing-key", STROKE_DEAD_KEY);
        // x-message-ttl  声明队列的TTL
        args.put("x-message-ttl", DELAY_TIME);
       //return QueueBuilder.durable(STROKE_OVER_QUEUE).withArguments(args).build();
       return new Queue(STROKE_OVER_QUEUE, true, false, false, args);
    }

 队列声明包括队列名称、是否持久化、是否独占、是否自动删除等

声明死信队列

   @Bean
    public Queue strokeDeadQueue() {
        return new Queue(STROKE_DEAD_QUEUE,true);
    }

创建超时交换机和死信交换机

    @Bean
    DirectExchange strokeOverQueueExchange() {
        return new DirectExchange(STROKE_OVER_QUEUE_EXCHANGE);
    }
    @Bean
    DirectExchange strokeOverQueueExchange() {
        return new DirectExchange(STROKE_OVER_QUEUE_EXCHANGE);
    }

 超时队列和死信队列绑定

    @Bean
    Binding bindingStrokeOverDirect() {
        return BindingBuilder.bind(strokeOverQueue()).to(strokeOverQueueExchange()).with(STROKE_OVER_KEY);
    }
    @Bean
    Binding bindingStrokeDeadDirect() {
            return BindingBuilder.bind(strokeDeadQueue()).to(strokeDeadQueueExchange()).with(STROKE_DEAD_KEY);
    }

生产者发送信息

@Component
public class MQProducer {
    @Autowired
    RabbitTemplate rabbitTemplate;
 
    /**
     * 发送延时订单MQ
     *
     * @param strokeVO
     */
    public void sendOver(StrokeVO strokeVO) {
        String mqMessage = JSON.toJSONString(strokeVO);
        rabbitTemplate.convertAndSend( RabbitConfig.STROKE_OVER_QUEUE_EXCHANGE, RabbitConfig.STROKE_OVER_KEY, mqMessage);
    }
 
}

消费者手动确认

@Component
public class MQConsumer{
    @Autowired
    private StrokeHandler strokeHandler;
 
    /**
     * 行程超时监听
     *
     * @param massage
     * @param channel
     * @param tag
     */
    @RabbitListener(
            bindings =
                    {
                            @QueueBinding(value = @Queue(value = RabbitConfig.STROKE_DEAD_QUEUE, durable = "true"),
                                    exchange = @Exchange(value = RabbitConfig.STROKE_DEAD_QUEUE_EXCHANGE), key = RabbitConfig.STROKE_DEAD_KEY)
                    })
    @RabbitHandler
    public void processStroke(Message massage, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        StrokeVO strokeVO = JSON.parseObject(massage.getBody(), StrokeVO.class);
        if (null == strokeVO) {
            return;
        }
        try {
            //处理消息
            strokeHandler.timeoutHandel(strokeVO);
            //RabbitMQ的一种确认机制，手动确认消息已被成功处理
            channel.basicAck(tag, false);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
 
}

@RabbitListener注解，它标记下面的方法作为RabbitMQ消息的消费者。定义了一个队列绑定，该绑定详细描述了队列、交换机和路由键的信息。

@QueueBinding注解，用于定义队列绑定信息

@Queue定义了一个队列，名称为RabbitConfig.STROKE_DEAD_QUEUE，并且这个队列是持久的（durable="true"）。

@Exchange定义了一个交换机，名称为RabbitConfig.STROKE_DEAD_QUEUE_EXCHANGE

key = RabbitConfig.STROKE_DEAD_KEY是绑定到交换机的路由键。

@RabbitHandler表示下面的方法是处理RabbitMQ消息的方法