RabbitMQ 核心概念与消息模式详解

流程：

1. 生产者声明队列（queueDeclare）
2. 生产者发送消息，routingKey = 队列名，发送到默认 Exchange
3. 消费者从该队列消费消息

适用场景：

• 简单异步任务、通知，例如：
  • 发短信验证码
  • 简单的异步写入日志

关键点：

• 消息只能被一个消费者消费一次
• 不需要复杂路由，只是'把任务丢给队列，让后台慢慢处理'

代码：

生产者：

public class Producer { public static void main(String[] args) throws Exception { // 创建连接工厂 ConnectionFactory connectionFactory = new ConnectionFactory(); // 设置主机地址 connectionFactory.setHost("127.0.0.1"); // 设置连接端口号：默认为 5672 connectionFactory.setPort(5672); // 虚拟主机名称：默认为 / connectionFactory.setVirtualHost("/"); // 设置连接用户名；默认为 guest connectionFactory.setUsername("guest"); // 设置连接密码；默认为 guest connectionFactory.setPassword("guest"); // 创建连接 Connection connection = connectionFactory.newConnection(); // 创建频道 Channel channel = connection.createChannel(); // 声明（创建）队列 // queue 参数 1：队列名称 // durable 参数 2：是否定义持久化队列，当 MQ 重启之后还在 // exclusive 参数 3：是否独占本次连接。若独占，只能有一个消费者监听这个队列且 Connection 关闭时删除这个队列 // autoDelete 参数 4：是否在不使用的时候自动删除队列，也就是在没有 Consumer 时自动删除 // arguments 参数 5：队列其它参数 channel.queueDeclare("simple_queue", true, false, false, null); // 要发送的信息 String message = "你好；小兔子！"; // 参数 1：交换机名称，如果没有指定则使用默认 Default Exchange // 参数 2：路由 key，简单模式可以传递队列名称 // 参数 3：配置信息 // 参数 4：消息内容 channel.basicPublish("", "simple_queue", null, message.getBytes()); System.out.println("已发送消息：" + message); // 关闭资源 channel.close(); connection.close(); } }

消费者：

public class Consume { public static void main(String[] args) throws Exception { // 1.创建连接工厂 ConnectionFactory factory = new ConnectionFactory(); // 2. 设置参数 factory.setHost("127.0.0.1"); factory.setPort(5672); factory.setVirtualHost("/"); factory.setUsername("guest"); factory.setPassword("guest"); // 3. 创建连接 Connection Connection connection = factory.newConnection(); // 4. 创建 Channel Channel channel = connection.createChannel(); // 5. 创建队列 // 如果没有一个名字叫 simple_queue 的队列，则会创建该队列，如果有则不会创建 // 参数 1. queue：队列名称 // 参数 2. durable：是否持久化。如果持久化，则当 MQ 重启之后还在 // 参数 3. exclusive：是否独占。 // 参数 4. autoDelete：是否自动删除。当没有 Consumer 时，自动删除掉 // 参数 5. arguments：其它参数。 channel.queueDeclare("simple_queue",true,false,false,null); // 接收消息 DefaultConsumer consumer = new DefaultConsumer(channel){ // 回调方法，当收到消息后，会自动执行该方法 // 参数 1. consumerTag：标识 // 参数 2. envelope：获取一些信息，交换机，路由 key... // 参数 3. properties：配置信息 // 参数 4. body：数据 @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println("consumerTag:"+consumerTag); System.out.println("Exchange:"+envelope.getExchange()); System.out.println("RoutingKey:"+envelope.getRoutingKey()); System.out.println("properties:"+properties); System.out.println("body:"+new String(body)); } }; // 参数 1. queue：队列名称 // 参数 2. autoAck：是否自动确认，类似咱们发短信，发送成功会收到一个确认消息 // 参数 3. callback：回调对象 // 消费者类似一个监听程序，主要是用来监听消息 channel.basicConsume("simple_queue",true,consumer); } }

2. 工作队列模式

特点：

• 一个生产者 → 一个队列 → 多个消费者 C1…Cn
• 多个消费者竞争消费同一个队列，一条消息只会被一个消费者消费
• 默认使用轮询（Round-Robin）分发给多个消费者

流程：

1. 生产者往队列里持续扔任务
2. 多个 worker（消费者）同时从队列取任务
3. RabbitMQ 按轮询或公平分发策略，把任务分给不同 worker

适用场景：

• 耗时任务异步处理，支持负载均衡：
  • 图片/视频处理、压缩
  • 订单后续处理（发短信、积分等）

关键点：

• 消息不会重复消费：每条消息只给一个 worker
• 可以通过 QoS + 手动 ACK 实现'能者多劳'（公平分发）：
  • 设置 prefetchCount = 1，同时只给一个未确认消息
  • 消费者处理慢就少拿点，处理快就多拿点

代码：

生产者：

public class Producer { public static final String QUEUE_NAME = "work_queue"; public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(QUEUE_NAME,true,false,false,null); for (int i = 1; i <= 10; i++) { String body = i+"hello rabbitmq~~~"; channel.basicPublish("",QUEUE_NAME,null,body.getBytes()); } channel.close(); connection.close(); } }

消费者 1：

public class Consumer1 { static final String QUEUE_NAME = "work_queue"; public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(QUEUE_NAME,true,false,false,null); Consumer consumer = new DefaultConsumer(channel){ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println("Consumer1 body:"+new String(body)); } }; channel.basicConsume(QUEUE_NAME,true,consumer); } }

消费者 2：

public class Consumer2 { static final String QUEUE_NAME = "work_queue"; public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(QUEUE_NAME,true,false,false,null); Consumer consumer = new DefaultConsumer(channel){ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println("Consumer2 body:"+new String(body)); } }; channel.basicConsume(QUEUE_NAME,true,consumer); } }

3. 发布/订阅模式

特点：

• 一个生产者 → 一个 Fanout Exchange → 多个队列 → 多个消费者
• 广播：消息会被所有绑定到该 Exchange 的队列收到
• Fanout Exchange 会忽略 routing key，直接发给所有绑定队列

流程：

1. 声明 Fanout Exchange：exchangeDeclare("logs", "fanout")
2. 每个消费者声明自己的队列，并绑定到这个 Exchange：
   • queueDeclare() → 得到一个临时队列（exclusive/autoDelete）
   • queueBind(queueName, "logs", "")
3. 生产者往 logs Exchange 发消息，所有队列都会收到一份

适用场景：

• 广播型事件：
  • 日志系统：所有日志服务都收到同一份日志
  • 实时新闻、通知广播

关键点：

• 每个消费者有自己的独立队列，互不影响
• 如果没有队列绑定到 Exchange，消息会被丢弃（Fanout 不存储消息）

代码：

生产者：

public class Producer { public static void main(String[] args) throws Exception { // 1、获取连接 Connection connection = ConnectionUtil.getConnection(); // 2、创建频道 Channel channel = connection.createChannel(); // 参数 1. exchange：交换机名称 // 参数 2. type：交换机类型 // DIRECT("direct")：定向 // FANOUT("fanout")：扇形（广播），发送消息到每一个与之绑定队列。 // TOPIC("topic")：通配符的方式 // HEADERS("headers")：参数匹配 // 参数 3. durable：是否持久化 // 参数 4. autoDelete：自动删除 // 参数 5. internal：内部使用。一般 false // 参数 6. arguments：其它参数 String exchangeName = "test_fanout"; // 3、创建交换机 channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT,true,false,false,null); // 4、创建队列 String queue1Name = "test_fanout_queue1"; String queue2Name = "test_fanout_queue2"; channel.queueDeclare(queue1Name,true,false,false,null); channel.queueDeclare(queue2Name,true,false,false,null); // 5、绑定队列和交换机 // 参数 1. queue：队列名称 // 参数 2. exchange：交换机名称 // 参数 3. routingKey：路由键，绑定规则 // 如果交换机的类型为 fanout，routingKey 设置为"" channel.queueBind(queue1Name,exchangeName,""); channel.queueBind(queue2Name,exchangeName,""); String body = "日志信息：张三调用了 findAll 方法...日志级别：info..."; // 6、发送消息 channel.basicPublish(exchangeName,"",null,body.getBytes()); // 7、释放资源 channel.close(); connection.close(); } }

消费者 1：

public class Consumer1 { public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); String queue1Name = "test_fanout_queue1"; channel.queueDeclare(queue1Name,true,false,false,null); Consumer consumer = new DefaultConsumer(channel){ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println("body:"+new String(body)); System.out.println("队列 1 消费者 1 将日志信息打印到控制台....."); } }; channel.basicConsume(queue1Name,true,consumer); } }

消费者 2：

public class Consumer2 { public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); String queue2Name = "test_fanout_queue2"; channel.queueDeclare(queue2Name,true,false,false,null); Consumer consumer = new DefaultConsumer(channel){ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println("body:"+new String(body)); System.out.println("队列 2 消费者 2 将日志信息打印到控制台....."); } }; channel.basicConsume(queue2Name,true,consumer); } }

4. 路由模式

特点：

• 使用 Direct Exchange
• 消息带 routing key，队列绑定到 Exchange 时指定 binding key
• routing key == binding key 时，消息才会路由到该队列

流程：

1. 声明 Direct Exchange：exchangeDeclare("direct_logs", "direct")
2. 队列 Q1 绑定：queueBind(Q1, "direct_logs", "error")
3. 队列 Q2 绑定：queueBind(Q2, "direct_logs", "info") 和 "warning"
4. 生产者发送：
   • routingKey = "error" → 只有 Q1 收到
   • routingKey = "info" → 只有 Q2 收到

适用场景：

• 按业务类型/级别分发：
  • 日志级别：error / warning / info
  • 业务类型：订单支付 / 订单物流 / 订单退款

关键点：

• 一个队列可以绑定多个 routing key
• 同一个 routing key 可以绑定多个队列（实现'多播'）

代码：

生产者：

public class Producer { public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); String exchangeName = "test_direct"; // 创建交换机 channel.exchangeDeclare(exchangeName,BuiltinExchangeType.DIRECT,true,false,false,null); // 创建队列 String queue1Name = "test_direct_queue1"; String queue2Name = "test_direct_queue2"; // 声明（创建）队列 channel.queueDeclare(queue1Name,true,false,false,null); channel.queueDeclare(queue2Name,true,false,false,null); // 队列绑定交换机 // 队列 1 绑定 error channel.queueBind(queue1Name,exchangeName,"error"); // 队列 2 绑定 info error warning channel.queueBind(queue2Name,exchangeName,"info"); channel.queueBind(queue2Name,exchangeName,"error"); channel.queueBind(queue2Name,exchangeName,"warning"); String message = "日志信息：张三调用了 delete 方法。错误了，日志级别 warning"; // 发送消息 channel.basicPublish(exchangeName,"warning",null,message.getBytes()); System.out.println(message); // 释放资源 channel.close(); connection.close(); } } 

消费者 1：

public class Consumer1 { public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); String queue1Name = "test_direct_queue1"; channel.queueDeclare(queue1Name,true,false,false,null); Consumer consumer = new DefaultConsumer(channel){ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println("body:"+new String(body)); System.out.println("Consumer1 将日志信息打印到控制台....."); } }; channel.basicConsume(queue1Name,true,consumer); } } 

消费者 2：

public class Consumer2 { public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); String queue2Name = "test_direct_queue2"; channel.queueDeclare(queue2Name,true,false,false,null); Consumer consumer = new DefaultConsumer(channel){ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println("body:"+new String(body)); System.out.println("Consumer2 将日志信息存储到数据库....."); } }; channel.basicConsume(queue2Name,true,consumer); } }

5. 主题模式 / 通配符模式

特点：

• 使用 Topic Exchange
• routing key 是带分隔符的字符串，例如 order.payment.success
• 队列绑定的是模式，支持通配符：
  • *：匹配恰好一个单词
  • #：匹配 0 或多个单词

例子：

• 绑定模式：
  • order.* → 匹配 order.payment、order.shipping
  • order.# → 匹配所有以 order. 开头的路由键
• routing key：
  • order.payment.success → 被 order.# 匹配
  • order.payment → 被 order.* 匹配

适用场景：

• 多维度分类路由：
  • 日志：error.app、error.db、warn.*
  • 地域业务：asia.china、asia.#

关键点：

• 比 Direct 更灵活，可以实现'分类 + 层级'路由
• 注意不要滥用太复杂的模式，会影响性能和可读性

代码：

生产者：

public class Producer { public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); String exchangeName = "test_topic"; channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC,true,false,false,null); String queue1Name = "test_topic_queue1"; String queue2Name = "test_topic_queue2"; channel.queueDeclare(queue1Name,true,false,false,null); channel.queueDeclare(queue2Name,true,false,false,null); // 绑定队列和交换机 // 参数 1. queue：队列名称 // 参数 2. exchange：交换机名称 // 参数 3. routingKey：路由键，绑定规则 // 如果交换机的类型为 fanout ,routingKey 设置为"" // routing key 常用格式：系统的名称。日志的级别。 // 需求：所有 error 级别的日志存入数据库，所有 order 系统的日志存入数据库 channel.queueBind(queue1Name,exchangeName,"#.error"); channel.queueBind(queue1Name,exchangeName,"order.*"); channel.queueBind(queue2Name,exchangeName,"*.*"); // 分别发送消息到队列：order.info、goods.info、goods.error String body = "[所在系统：order][日志级别：info][日志内容：订单生成，保存成功]"; channel.basicPublish(exchangeName,"order.info",null,body.getBytes()); body = "[所在系统：goods][日志级别：info][日志内容：商品发布成功]"; channel.basicPublish(exchangeName,"goods.info",null,body.getBytes()); body = "[所在系统：goods][日志级别：error][日志内容：商品发布失败]"; channel.basicPublish(exchangeName,"goods.error",null,body.getBytes()); channel.close(); connection.close(); } }

消费者 1：

public class Consumer1 { public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); String QUEUE_NAME = "test_topic_queue1"; channel.queueDeclare(QUEUE_NAME,true,false,false,null); Consumer consumer = new DefaultConsumer(channel){ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println("body:"+new String(body)); } }; channel.basicConsume(QUEUE_NAME,true,consumer); } }

消费者 2：

public class Consumer2 { public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); Channel channel = connection.createChannel(); String QUEUE_NAME = "test_topic_queue2"; channel.queueDeclare(QUEUE_NAME,true,false,false,null); Consumer consumer = new DefaultConsumer(channel){ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println("body:"+new String(body)); } }; channel.basicConsume(QUEUE_NAME,true,consumer); } }

6. RPC 模式

特点：

• 利用 RabbitMQ 实现'请求 - 响应'式的远程调用
• 并不是'标准的 RPC 框架'（如 Dubbo、gRPC），而是一种基于消息的 RPC 模式

典型流程：

1. 客户端（调用方）：
   • 声明一个临时回调队列（exclusive、autoDelete）
   • 发送请求消息，设置属性：
     • replyTo：回调队列名
     • correlationId：唯一请求 ID
   • 监听回调队列
2. 服务端：
   • 从请求队列取出消息
   • 处理请求
   • 将结果发送到 replyTo 指定的回调队列，并带上 correlationId
3. 客户端：
   • 从回调队列收到响应，根据 correlationId 找到对应的请求

适用场景：

• 分布式计算、查询服务：
  • 计算密集型任务
  • 需要跨服务的同步查询

关键点：

• 需要自己处理 correlationId 与请求的映射
• 注意超时、重试、幂等等问题

7. Publisher Confirms（发布确认）

特点：

• 不是业务上的'消息模式'，而是可靠性机制
• Channel 开启 confirms 模式后：
  • 每条消息发布后，Broker 会回传一个确认（ack/nack）
  • 生产者可以据此知道消息是否成功到达队列

适用场景：

• 对数据一致性要求高的场景：
  • 订单、支付、重要业务事件
  • 不允许消息在发送阶段丢失

关键点：

• 可以批量发布 + 批量确认，提升性能
• 一般配合 事务（transaction） 或 idempotent 发布一起设计

四、四种 Exchange 类型与消息模式的对应

• '竞争消费'？ → 用队列 + 多消费者（Work Queue）
• '广播'？ → 用 Fanout Exchange
• '按业务类型分发'？ → 用 Direct Exchange
• '多维度分类'？ → 用 Topic Exchange

五、消息可靠性

1. 消息持久化

• Exchange 持久化：durable = true
• Queue 持久化：queueDeclare(..., durable=true, ...)
• Message 持久化：deliveryMode = 2（PERSISTENT）

注意：持久化不等于绝对不丢，只是落到磁盘；仍可能因磁盘故障等丢失，需要配合集群、镜像等高可用方案。

2. ACK 与 NACK

• 消费者处理完消息后，发送 basic.ack 确认
• 如果处理失败，可以 basic.nack 或 basic.reject，配合：
  • requeue = true：重新入队
  • requeue = false：进入死信队列（DLX）

3. QoS（prefetch）

• 设置 basicQos(prefetchCount)：
  • 限制'未确认消息的最大数量'
  • 避免某个消费者被压垮，实现更公平的任务分配

六、根据业务选择消息模式

• 只是异步处理一个任务，不需要广播：
  → 简单模式 / 工作队列
• 一条消息要被多个系统同时处理：
  → 发布/订阅 / 路由 / 主题
• 需要按业务维度/日志级别/地域等分类路由：
  → Direct 或 Topic
• 需要同步返回结果：
  → RPC 模式

七、死信队列

死信队列是 RabbitMQ 实现消息可靠性兜底、异常消息隔离、失败重试的核心机制，本质上它是一个普通队列，仅用于接收「触发了死信规则的异常消息」，配套的核心组件是死信交换机。

**1、**核心概念

死信：触发了 RabbitMQ 预设规则，无法被正常消费的消息，会被标记为死信。

死信交换机：用于接收死信消息的普通交换机，类型支持 direct/topic/fanout/headers，和常规交换机用法完全一致。

死信队列：绑定了 DLX 的普通队列，专门用于存储死信消息，供兜底消费、排查问题、归档处理。

核心逻辑：正常队列必须提前绑定 DLX 和死信路由键，消息触发死信规则后，会被 RabbitMQ 自动转发到 DLX，最终路由到死信队列。

2、死信消息的 3 大核心触发条件

只有满足以下任一条件，消息才会成为死信，缺一不可：

3、死信队列核心架构

生产者 → 正常交换机 → 正常队列（绑定 DLX+ 死信路由键） ↓ 触发死信规则 死信消费者 ← 死信队列 ← 死信交换机（DLX）

4、核心使用场景

• 异常消息兜底：业务消费失败的消息，避免无限循环重试，转入死信队列做人工排查 / 兜底处理，防止消息丢失。
• 超时消息处理：订单未支付超时、会话超时等场景，通过 TTL 触发死信，实现超时逻辑。
• 队列溢出保护：设置队列最大长度，流量峰值时溢出的消息转入死信队列，避免 RabbitMQ 内存溢出崩溃。
• 消息重试机制：结合延迟队列，实现消费失败后的指数退避重试，多次重试失败后转入死信归档。

八、延迟队列

延迟队列是指：消息发送后，不希望消费者立即消费，而是等待指定的延迟时间后，再被消费者消费的队列。RabbitMQ 本身没有直接提供延迟队列功能，而是通过「TTL + 死信队列」或「官方延迟插件」两种方式实现，是订单超时、定时提醒、延迟重试等场景的核心解决方案。

1. 实现方式一：TTL + 死信队列

这是 RabbitMQ 原生支持的实现方式，无需额外安装插件，基于前文的死信队列能力实现。

核心原理

1. 创建一个无消费者的缓冲队列，给队列 / 消息设置 TTL（延迟时间）。
2. 缓冲队列绑定 DLX 和死信路由键，消息在缓冲队列中到期后，会触发 TTL 死信规则，被转发到 DLX。
3. 死信队列绑定 DLX，消费者监听死信队列，此时收到的消息就是已经过了延迟时间的消息，实现延迟消费。

生产者 → 延迟交换机 → 缓冲队列（带 TTL，无消费者，绑定 DLX） ↓ TTL 到期，触发死信 延迟消费者 ← 目标消费队列 ← 死信交换机（DLX）

实现代码

基于前文的死信配置，仅需修改缓冲队列配置，无需额外依赖：

@Configuration public class DelayQueueConfig { // 延迟缓冲交换机、队列 public static final String DELAY_BUFFER_EXCHANGE = "delay.buffer.exchange"; public static final String DELAY_BUFFER_QUEUE = "delay.buffer.queue"; public static final String DELAY_BUFFER_ROUTING_KEY = "delay.buffer.key"; // 延迟目标交换机、队列（实际消费的队列） public static final String DELAY_TARGET_EXCHANGE = "delay.target.exchange"; public static final String DELAY_TARGET_QUEUE = "delay.target.queue"; public static final String DELAY_TARGET_ROUTING_KEY = "delay.target.key"; // 1. 声明缓冲交换机 @Bean public DirectExchange delayBufferExchange() { return ExchangeBuilder.directExchange(DELAY_BUFFER_EXCHANGE).durable(true).build(); } // 2. 声明目标交换机（即死信交换机） @Bean public DirectExchange delayTargetExchange() { return ExchangeBuilder.directExchange(DELAY_TARGET_EXCHANGE).durable(true).build(); } // 3. 声明延迟缓冲队列，设置 TTL+DLX，无消费者监听 @Bean public Queue delayBufferQueue() { Map<String, Object> args = new HashMap<>(); // 绑定死信交换机（目标交换机） args.put("x-dead-letter-exchange", DELAY_TARGET_EXCHANGE); // 死信路由键（目标路由键） args.put("x-dead-letter-routing-key", DELAY_TARGET_ROUTING_KEY); // 设置延迟时间：30 分钟，单位毫秒（订单超时场景） args.put("x-message-ttl", 30 * 60 * 1000); return QueueBuilder.durable(DELAY_BUFFER_QUEUE).withArguments(args).build(); } // 4. 声明目标消费队列 @Bean public Queue delayTargetQueue() { return QueueBuilder.durable(DELAY_TARGET_QUEUE).build(); } // 5. 绑定关系 @Bean public Binding delayBufferBinding() { return BindingBuilder.bind(delayBufferQueue()).to(delayBufferExchange()).with(DELAY_BUFFER_ROUTING_KEY); } @Bean public Binding delayTargetBinding() { return BindingBuilder.bind(delayTargetQueue()).to(delayTargetExchange()).with(DELAY_TARGET_ROUTING_KEY); } }

消息发送与消费

// 消息发送：发送到缓冲队列，无需消费 rabbitTemplate.convertAndSend(DelayQueueConfig.DELAY_BUFFER_EXCHANGE, DelayQueueConfig.DELAY_BUFFER_ROUTING_KEY, "订单 123456"); // 延迟消费：监听目标队列，30 分钟后收到消息 @RabbitListener(queues = "delay.target.queue") public void consumeDelay(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException { System.out.println("收到延迟消息，执行订单超时取消：" + message); channel.basicAck(deliveryTag, false); }

优缺点

2、实现方式二：官方延迟消息插件

RabbitMQ 官方推出了rabbitmq_delayed_message_exchange插件，完美解决了 TTL 方案的惰性检查问题，支持任意时间的延迟消息，是生产环境的首选方案。

核心原理

插件新增了一种x-delayed-message类型的交换机，核心逻辑：

1. 消息发送到该交换机后，不会立即投递到队列，而是先存储到 Mnesia 分布式表中。
2. 交换机根据消息头部的x-delay参数（延迟时间），定时检查消息是否到期。
3. 消息到期后，交换机将消息投递到绑定的目标队列，消费者直接监听目标队列即可。

步骤 1：插件安装

• 下载与你的 RabbitMQ 版本完全匹配的插件
• 将下载的.ez插件文件，放入 RabbitMQ 的plugins目录
• 执行启用命令：rabbitmq-plugins enable rabbitmq_delayed_message_exchange
• 重启 RabbitMQ 服务，管理端 Exchange 页面可看到x-delayed-message类型，即为安装成功

注意：集群环境下，每个节点都必须安装并启用该插件，否则会出现路由失败。

步骤 2：代码实现

@Configuration public class PluginDelayConfig { public static final String DELAYED_EXCHANGE = "delayed.exchange"; public static final String DELAYED_QUEUE = "delayed.queue"; public static final String DELAYED_ROUTING_KEY = "delayed.key"; // 声明延迟交换机：类型为 x-delayed-message，指定 x-delayed-type 为 direct @Bean public CustomExchange delayedExchange() { Map<String, Object> args = new HashMap<>(); // 指定底层交换机类型，支持 direct/topic/fanout args.put("x-delayed-type", "direct"); return new CustomExchange(DELAYED_EXCHANGE, "x-delayed-message", true, false, args); } // 声明目标消费队列，无特殊配置 @Bean public Queue delayedQueue() { return QueueBuilder.durable(DELAYED_QUEUE).build(); } // 绑定队列与延迟交换机 @Bean public Binding delayedBinding() { return BindingBuilder.bind(delayedQueue()).to(delayedExchange()).with(DELAYED_ROUTING_KEY).noargs(); } }

消息发送与消费

// 消息发送：通过 header 设置 x-delay，单位毫秒，支持任意延迟时间 String orderId = "123456"; int delayTime = 10 * 60 * 1000; // 10 分钟延迟 rabbitTemplate.convertAndSend(PluginDelayConfig.DELAYED_EXCHANGE, PluginDelayConfig.DELAYED_ROUTING_KEY, orderId, message -> { // 设置延迟时间，核心参数 message.getMessageProperties().setHeader("x-delay", delayTime); return message; }); // 延迟消费：直接监听目标队列，到期自动收到消息 @RabbitListener(queues = "delayed.queue") public void consumeDelayed(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) throws IOException { System.out.println("收到插件延迟消息：" + message); channel.basicAck(deliveryTag, false); }

优缺点

3、两种方案选型对比

4、核心使用场景

• 订单超时自动取消（30 分钟未支付关闭订单、恢复库存）
• 预约提醒（会议前 15 分钟、出行前 1 小时提醒）
• 延迟重试（消费失败后，10s/30s/1min/5min 指数退避重试）
• 缓存预热（指定时间预热热点数据）
• 超时关闭会话（用户长时间无操作自动登出）