使用RabbitMQ死信交换机实现延迟消息

在开发过程中，我们常常会遇到需要延迟处理某些消息的场景，例如订单的支付超时处理、短信的定时发送等。本文将介绍如何使用RabbitMQ的死信交换机（Dead Letter Exchange，DLX）来实现延迟消息的处理。

什么是死信交换机？

死信交换机是一种特殊的交换机，用于处理不能被正常消费的消息。当消息在队列中出现以下几种情况时，会被转发到死信交换机：

1. 消息被拒绝（Basic.Reject或Basic.Nack）并且requeue参数设置为false。
2. 消息在队列中的存活时间超过了TTL（Time To Live）。
3. 队列的最大长度已满，导致消息被丢弃。

通过配置死信交换机，我们可以将这些“死信”转发到一个特殊的队列，从而进行后续处理。

死信交换机实现延迟消息的思路

利用消息的TTL属性和死信交换机，我们可以实现延迟消息的处理。具体步骤如下：

1. 创建一个普通交换机和队列，队列绑定一个死信交换机。
2. 发送消息到普通队列，并设置消息的TTL。
3. 消息过期后，会转发到死信交换机，死信交换机再将消息路由到实际处理的队列中。

实现过程

配置类

首先，我们需要配置普通交换机、队列和绑定关系：

package com.itheima.consumer.config;

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

@Configuration
public class NormalConfiguration {

    @Bean
    public DirectExchange normalExchange() {
        return ExchangeBuilder.directExchange("normal.direct").build();
    }

    @Bean
    public Queue normalQueue() {
        return QueueBuilder.durable("normal.queue")
                .deadLetterExchange("dlx.direct")
                .build();
    }

    @Bean
    public Binding normalExchangeBinding(Queue normalQueue, DirectExchange normalExchange) {
        return BindingBuilder.bind(normalQueue).to(normalExchange).with("hi");
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

在上述配置中，我们创建了一个普通交换机normal.direct，以及一个普通队列normal.queue，并将队列绑定到了死信交换机dlx.direct。

消费者监听死信队列

接下来，我们需要定义一个消费者来监听死信队列，并处理延迟后的消息：

import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.core.ExchangeTypes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

@Component
public class DlxMessageListener {

    private static final Logger log = LoggerFactory.getLogger(DlxMessageListener.class);

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "dlx.queue", durable = "true"),
            exchange = @Exchange(name = "dlx.direct", type = ExchangeTypes.DIRECT),
            key = {"hi"}
    ))
    public void listenDlxQueue1(String message) {
        log.info("消费者监听到dlx.queue消息：{}", message);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

发送延迟消息

最后，我们编写一个测试方法来发送延迟消息：

import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;

@SpringBootTest
public class DelayMessageTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    void testSendDelayMessage() {
        rabbitTemplate.convertAndSend("normal.direct", "hi", "hello", message -> {
            message.getMessageProperties().setExpiration("10000");
            return message;
        });
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

在上述测试方法中，我们向普通队列发送了一条消息，并设置其TTL为10000毫秒（即10秒）。当消息在普通队列中存活超过10秒后，会被转发到死信交换机，然后由消费者监听并处理。

注意事项

需要注意的是，RabbitMQ的消息过期是基于追溯方式来实现的，也就是说当一个消息的TTL到期以后，不一定会立即被移除或投递到死信交换机，而是在消息恰好处于队首时才会被处理。当队列中消息堆积很多的时候，过期消息可能不会被按时处理，因此你设置的TTL时间不一定准确。

总结

通过以上配置和代码，我们实现了使用RabbitMQ死信交换机来处理延迟消息。利用死信交换机的机制，我们可以方便地实现各种复杂的消息处理场景，提高系统的灵活性和可靠性。