RabbitMQ实现死信队列

简介

死信队列（Dead-Letter-Queue，DLQ）用于处理无法被正常消费的消息。达到最大重试次数后，若消费依然失败，则表明消费者在正常情况下无法正确地消费该消息，此时 CMQ 不会立刻将消息丢弃，而是将其发送到该消费者对应的特殊队列中，这个特殊队列就是 DLQ。对新建的队列，或存量的队列，您都可以启用死信队列。

死信队列

死信，在官网中对应的单词为 “Dead Letter”，可以看出翻译确实非常的简单粗暴。那么死信是个什么东西呢？

“死信” 是 RabbitMQ 中的一种消息机制，当你在消费消息时，如果队列里的消息出现以下情况：

1. 消息被否定确认，使用 channel.basicNack 或 channel.basicReject ，并且此时 requeue 属性被设置为 false。
2. 消息在队列的存活时间超过设置的 TTL 时间。
3. 消息队列的消息数量已经超过最大队列长度。

那么该消息将成为 “死信”。

“死信” 消息会被 RabbitMQ 进行特殊处理，如果配置了死信队列信息，那么该消息将会被丢进死信队列中，如果没有配置，则该消息将会被丢弃。

配置实现

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

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

/**
 * @author 曳戈泰尔
 * @description 死信队列 配置
 * @date 日期：2019/12/19 11:07
 */
@Configuration
public class DeadLetterConfig {

    /** 延迟队列 TTL 名称 */
    private static final String ORDER_DELAY_QUEUE = "order.delay.queue";
    /** DLX，dead letter 发送到的 exchange 延时消息就是发送到该交换机的 */
    public static final String ORDER_DELAY_EXCHANGE = "order.delay.exchange";
    /** routing key 名称 路由键 具体消息发送在该 routingKey 的 */
    public static final String ORDER_DELAY_ROUTING_KEY = "order.delay.key";

    /** 死信队列 TTL 名称 */
    public static final String DEAD_LETTER_QUEUE = "dead.letter.queue";
    /** 死信队列交换机 */
    public static final String DEAD_LETTER_EXCHANGE = "dead.letter.exchange";
    /** 死信队列 路由键 */
    public static final String DEAD_LETTER_ROUTING_KEY = "dead.letter.key";

    /** 死信接收队列 */
    @Bean
    public Queue deadLetterQueue() {
        return new Queue(DEAD_LETTER_QUEUE, true);
    }

    /**
     * 死信交换机 将路由键和某模式进行匹配。此时队列需要绑定要一个模式上。
     * 符号“#”匹配一个或多个词，符号“*”匹配不多不少一个词。因此“audit.#”能够匹配到“audit.irs.corporate”，但是“audit.*”
     * 只会匹配到“audit.irs”。
     */
    @Bean
    public TopicExchange deadLetterTopicExchange() {
        return new TopicExchange(DEAD_LETTER_EXCHANGE);
    }

    /** 死信接收队列绑定交换机 */
    @Bean
    public Binding deadLetterBinding() {
        // TODO 如果要让延迟队列之间有关联,这里的 routingKey 和 绑定的交换机很关键
        return BindingBuilder.bind(deadLetterQueue())
                .to(deadLetterTopicExchange())
                .with(DEAD_LETTER_ROUTING_KEY);
    }

    /**
     * 延迟队列配置(死信队列)
     *
     * <p>1、params.put("x-message-ttl", 5 * 1000); 第一种方式是直接设置 Queue 延迟时间
     * 但如果直接给队列设置过期时间,这种做法不是很灵活,（当然二者是兼容的,默认是时间小的优先） 2、rabbitTemplate.convertAndSend(book, message
     * -> { message.getMessageProperties().setExpiration(2 * 1000 + ""); return message; });
     * 第二种就是每次发送消息动态设置延迟时间,这样我们可以灵活控制
     */
    @Bean
    public Queue delayOrderQueue() {

        Map<String, Object> params = new HashMap<>();
        // x-dead-letter-exchange 声明了队列里的死信转发到的DLX名称，
        params.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);
        // x-dead-letter-routing-key 声明了这些死信在转发时携带的 routing-key 名称。
        params.put("x-dead-letter-routing-key", DEAD_LETTER_ROUTING_KEY);

        return new Queue(ORDER_DELAY_QUEUE, true, false, false, params);
    }

    /**
     * 需要将一个队列绑定到交换机上，要求该消息与一个特定的路由键完全匹配。 这是一个完整的匹配。如果一个队列绑定到该交换机上要求路由键 “dog”，则只有被标记为“dog”的消息才被转发，
     * 不会转发dog.puppy，也不会转发dog.guard，只会转发dog。
     *
     * @return DirectExchange
     */
    @Bean
    public DirectExchange delayOrderExchange() {
        return new DirectExchange(ORDER_DELAY_EXCHANGE);
    }

    /** 延迟队列绑定交换机 */
    @Bean
    public Binding dlxBinding() {
        return BindingBuilder.bind(delayOrderQueue())
                .to(delayOrderExchange())
                .with(ORDER_DELAY_ROUTING_KEY);
    }
}