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.el7curl -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=RabbitMQDemospring.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";// 配置死信交换器@BeanDirectExchange deadLetterExchange() {return new DirectExchange(DEAD_LETTER_EXCHANGE);}// 配置死信队列@BeanQueue deadLetterQueue() {return new Queue(DEAD_LETTER_QUEUE, true);}// 绑定死信队列到死信交换器@BeanBinding deadLetterBinding() {return BindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange()).with(DEAD_LETTER_ROUTING_KEY);}// 配置业务交换器@BeanDirectExchange workExchange() {return new DirectExchange(WORK_EXCHANGE);}// 配置业务队列，并设置死信交换器@BeanQueue 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);}// 绑定业务队列到业务交换器@BeanBinding workBinding() {return BindingBuilder.bind(workQueue()).to(workExchange()).with(WORK_ROUTING_KEY);}// 配置消息转换器@Beanpublic Jackson2JsonMessageConverter jsonMessageConverter() {return new Jackson2JsonMessageConverter();}// 配置RabbitTemplate@Beanpublic 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 {@Autowiredprivate 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);}
}