RabbitMQ 利用死信队列来实现延迟消息

RabbitMQ 利用死信队列来实现延迟消息

基于 TTL（Time-To-Live）+ 死信队列（DLX）的方式来实现延迟消息

首先消息会被推送到普通队列中，该消息设置了TTL，当TTL到期未被消费掉，则会自动进入死信队列（DLQ）中，由死信队列消费者消费，来达到延迟消息的效果

首先让我们来安装 Rabbit MQ 服务端

由于服务器基本都是使用 Linux 系统
以下介绍常见的 Ubuntu/Debian 和 CentOS 系统安装 RabbitMQ 的方法

Ubuntu / Debian

# 添加 Erlang 仓库
curl -s https://packagecloud.io/install/repositories/rabbitmq/erlang/script.deb.sh | sudo bash

# 安装兼容版本的 Erlang（例如 25.3.2）
sudo apt-get install -y erlang=1:25.3.2+dfsg-1

# 添加 RabbitMQ 官方仓库
curl -s https://packagecloud.io/install/repositories/rabbitmq/rabbitmq-server/script.deb.sh | sudo bash

# 安装 RabbitMQ
sudo apt-get update
# 安装最新稳定版
sudo apt-get install -y rabbitmq-server

# 启动服务
sudo systemctl start rabbitmq-server
sudo systemctl enable rabbitmq-server  # 设置开机自启

# 然后是开放 Linux 的防火墙端口
# Ubuntu/Debian
sudo ufw allow 5672/tcp # 5672 是 RabbitMQ 服务端口
sudo ufw allow 15672/tcp # 15672 是 RabbitMQ 的后台管理端口

CentOS 7

# 添加Erlang仓库
curl -s https://packagecloud.io/install/repositories/rabbitmq/erlang/script.rpm.sh | sudo bash

# 安装兼容版本的Erlang（例如25.3.2）
sudo yum install -y erlang-25.3.2.8-1.el7

curl -s https://packagecloud.io/install/repositories/rabbitmq/rabbitmq-server/script.rpm.sh | sudo bash
sudo yum install -y rabbitmq-server-3.12.7-1.el7

# 启动RabbitMQ服务
sudo systemctl start rabbitmq-server
# 设置开机自启
sudo systemctl enable rabbitmq-server
# 检查服务状态
sudo systemctl status rabbitmq-server

# 开放AMQP协议端口（默认5672）
sudo firewall-cmd --permanent --add-port=5672/tcp
# 开放管理界面端口（默认15672）
sudo firewall-cmd --permanent --add-port=15672/tcp
# 重新加载防火墙规则
sudo firewall-cmd --reload

到这里你已经安装完成了， 然后该开始初始化 Rabbit MQ 设置一些账号相关的配置了

# 安装完后用 rabbitmqctl 命令来查看服务状态
sudo rabbitmqctl status

# 然后是安装后台管理页面，虽然都是用代码来操作MQ，
# 但是业务上难免是需要看MQ运行状态的，所以强烈建议是装上可视化界面
sudo rabbitmq-plugins enable rabbitmq_management

# 接下来是创建账号，创建虚拟主机，给用户授权，这些账号是要配置在 application.properties 中的

# 创建新用户
sudo rabbitmqctl add_user yourUserName yourPassword # 替换成你的账号密码
# 创建虚拟主机
sudo rabbitmqctl add_vhost yourvhost # 替换成你的虚拟主机
# 为用户授予虚拟主机的权限
sudo rabbitmqctl set_permissions -p yourvhost yourUserName ".*" ".*" ".*"
# 查看用户权限
sudo rabbitmqctl list_user_permissions yourUserName
# 设置 yourUserName 为管理员角色
# 这一步很重要，你要设置成管理员，后面才能登录后台管理页面
sudo rabbitmqctl set_user_tags yourUserName administrator
# 设置完后你就可以通过浏览器 访问 http://localhost:15672/ 登录后台（自己替换成 Linux 的IP)

顺带提一嘴，权限级别有多级, administrator 为最高权限

• none：无特殊权限（默认角色）
• management：可以访问管理 API 和 Web 界面
• policymaker：包含management权限，还能管理策略和参数
• monitoring：包含management权限，还能查看节点和集群信息
• administrator：最高权限，可管理所有资源和用户

接下来接入到 Spring 工程里

application.properties

spring.application.name=RabbitMQDemo

spring.rabbitmq.host=localhost # 替换为你的RabbitMQ服务IP
spring.rabbitmq.port=5672
spring.rabbitmq.username=yourUserName
spring.rabbitmq.password=yourPassword

Config 配置类

package vip.erichong.rabbitmqdemo.mq.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

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

/**
 * @author eric
 */
@Configuration
public class RabbitMQConfig {

    // 死信队列相关配置
    public static final String DEAD_LETTER_EXCHANGE = "dlx.exchange";
    public static final String DEAD_LETTER_QUEUE = "dlx.queue";
    public static final String DEAD_LETTER_ROUTING_KEY = "dlx.key";

    // 业务队列相关配置
    public static final String WORK_EXCHANGE = "work.exchange";
    public static final String WORK_QUEUE = "work.queue";
    public static final String WORK_ROUTING_KEY = "work.key";

    // 配置死信交换器
    @Bean
    DirectExchange deadLetterExchange() {
        return new DirectExchange(DEAD_LETTER_EXCHANGE);
    }

    // 配置死信队列
    @Bean
    Queue deadLetterQueue() {
        return new Queue(DEAD_LETTER_QUEUE, true);
    }

    // 绑定死信队列到死信交换器
    @Bean
    Binding deadLetterBinding() {
        return BindingBuilder.bind(deadLetterQueue())
                .to(deadLetterExchange())
                .with(DEAD_LETTER_ROUTING_KEY);
    }

    // 配置业务交换器
    @Bean
    DirectExchange workExchange() {
        return new DirectExchange(WORK_EXCHANGE);
    }

    // 配置业务队列，并设置死信交换器
    @Bean
    Queue workQueue() {
        Map<String, Object> args = new HashMap<>();
        // 这个是死信队列的关键，指定了 work.queue 消息过期时要移入死信队列
        args.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE);
        args.put("x-dead-letter-routing-key", DEAD_LETTER_ROUTING_KEY);
        return new Queue(WORK_QUEUE, true, false, false, args);
    }

    // 绑定业务队列到业务交换器
    @Bean
    Binding workBinding() {
        return BindingBuilder.bind(workQueue())
                .to(workExchange())
                .with(WORK_ROUTING_KEY);
    }

    // 配置消息转换器
    @Bean
    public Jackson2JsonMessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    // 配置RabbitTemplate
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        template.setMessageConverter(jsonMessageConverter());
        return template;
    }
}

生产者类

package vip.erichong.rabbitmqdemo.mq.producer;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;
import vip.erichong.rabbitmqdemo.mq.config.RabbitMQConfig;

import java.time.LocalDateTime;

/**
 * @author eric
 */
@Service
public class RabbitMQProducer {
    private final RabbitTemplate rabbitTemplate;

    public RabbitMQProducer(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    // 发送延迟消息
    public void sendDelayedMessage(String message, int delaySeconds) {
        System.out.println("[" + LocalDateTime.now() + "] 发送延迟消息: " + message + ", 延迟: " + delaySeconds + "秒");
        rabbitTemplate.convertAndSend(
                RabbitMQConfig.WORK_EXCHANGE,
                RabbitMQConfig.WORK_ROUTING_KEY,
                message,
                msg -> {
                    msg.getMessageProperties().setExpiration(String.valueOf(delaySeconds * 1000));
                    return msg;
                }
        );
    }
}

消费者类

package vip.erichong.rabbitmqdemo.mq.consumer;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import vip.erichong.rabbitmqdemo.mq.config.RabbitMQConfig;

import java.time.LocalDateTime;

/**
 * @author eric
 */
@Component
public class MessageConsumer {
    @RabbitListener(queues = RabbitMQConfig.DEAD_LETTER_QUEUE)
    public void handleDeadLetterMessage(Message message) {
        System.out.println("[" + LocalDateTime.now() + "] 收到延迟消息: " + new String(message.getBody()));
        // 处理延迟后的业务逻辑
    }
}

测试 Controller

package vip.erichong.rabbitmqdemo.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import vip.erichong.rabbitmqdemo.mq.producer.RabbitMQProducer;

import java.util.Optional;

/**
 * @author eric
 */
@RestController
@RequestMapping("/message")
public class TestController {
    @Autowired
    private RabbitMQProducer rabbitMQProducer;

    @GetMapping("/pushDelayMessage")
    public void pushDelayMessage(@RequestParam(value = "message", required = false) String message) {
        String msg = Optional.ofNullable(message).orElse("delay message");

        // 延迟 20 秒后，由 MessageConsumer.handleDeadLetterMessage 消费消息
        rabbitMQProducer.sendDelayedMessage(msg, 20);
    }
}

把项目启动起来，模拟HTTP请求发一条延迟消息吧，20秒钟之后就能看到消费者接收到消息了。

期间你可以在后台管理页面看到消息会先进到普通队列 work.queue 中，等待了20秒之后消息过期，消息被移入到与之绑定的死信队列 dlx.queue 中，由 MessageConsumer.handleDeadLetterMessage 消费该消息。

恭喜你，已经实现了一个简单的延迟消息功能。

完整的代码 Demo 点我获取