RabbitMQ 死信队列和延迟队列_mysql 延迟 队列的 message queue

死信队列

• 死信队列，英文缩写：DLX （Dead Letter Exchange（死信交换机）），当消息成为 Dead message 后，可以被重新发送到另一个交换机，这个交换机就是DLX

  • 在RabbitMQ中死信队列一般指的是死信交换机，其余MQ产品则指的是实际意义的死信队列
  • 比如一条消息过期后，这条消息就成为了 Dead message，如果该消息所在的队列绑定了死信交换机
  • 可以将这个 Dead message 发送到死信交换机，被传递到与死信交换机绑定的死信队列中进行处理

  

• 消息成为死信的三种情况（重点）

  • 队列消息长度达到限制，超出限制的消息成为死信消息
  • 消费者拒接消息，调用 basicNack，并且不把消息重新放入原目标队列 ，即 requeue=false；
  • 队列存在消息过期设置，消息达到超时时间未被消费

• 队列绑定死信交换机（此队列指的是产生死信消息的队列）

  • 声明队列时给队列设置参数： x-dead-letter-exchange 和 x-dead-letter-routing-key
  • 第一个参数设置死信交换机的名称
  • 第二个参数设置死信消息发往死信交换机时的 RoutingKey（将死信交换机当作普通交换机）
    • 理解：生产者向交换机发送消息时，也可以指定 RoutingKey

代码实现

在 spring-rabbitmq-producer.xml 中编写

1. 声明死信队列和死信交换机

   <!--声明死信队列(queue_dlx)和死信交换机(exchange_dlx)-->
   <rabbit:queue name="queue_dlx" id="queue_dlx"/>
   <rabbit:topic-exchange name="exchange_dlx">
       <rabbit:bindings>
           <rabbit:binding pattern="dlx.#" queue="queue_dlx"/>
       </rabbit:bindings>
   </rabbit:topic-exchange>
   1
   2
   3
   4
   5
   6
   7

2. 声明正常队列和正常交换机，并将正常队列绑定死信交换机，设置两个参数

   <!--声明正常的队列(test_queue_dlx)和交换机(test_exchange_dlx)-->
   
   <!--声明正常队列-->
   <rabbit:queue name="test_queue_dlx" id="test_queue_dlx">
   
       <!-- 正常队列绑定死信交换机，设置两个参数-->
       <rabbit:queue-arguments>
           
           <!-- x-dead-letter-exchange：死信交换机名称 -->
           <entry key="x-dead-letter-exchange" value="exchange_dlx" />
   
           <!-- x-dead-letter-routing-key：发送给死信交换机的routingkey -->
           <entry key="x-dead-letter-routing-key" value="dlx.hehe" />
           
           <!-- 设置队列的长度限制为10（方便演示成为死信消息），使用参数x-max-length -->
           <entry key="x-max-length" value="10" value-type="java.lang.Integer" />
           
       </rabbit:queue-arguments>
   </rabbit:queue>
   
   <!--声明正常交换机-->
   <rabbit:topic-exchange name="test_exchange_dlx">
       <rabbit:bindings>
           <rabbit:binding pattern="test.dlx.#" queue="test_queue_dlx"/>
       </rabbit:bindings>
   </rabbit:topic-exchange>
   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

3. 成为死信消息的三种情况可自行测试，消息发送给了正常队列，但最终死信消息都保存在了死信队列

延迟队列

• 延迟队列指的是消息进入队列之后，不会立即被消费，而是等待指定的时间后才会被消费

• RabbitMQ并没有提供延迟队列的功能，需要使用 TTL + 死信队列 的组合来实现延迟队列

  

代码实现

1. 在 spring-rabbitmq-producer.xml 中编写

   <!--  1. 定义死信交换机(order_exchange_dlx)和队列(order_queue_dlx)-->
   <rabbit:queue id="order_queue_dlx" name="order_queue_dlx"/>
   <rabbit:topic-exchange name="order_exchange_dlx">
       <rabbit:bindings>
           <rabbit:binding pattern="dlx.order.#" queue="order_queue_dlx"/>
       </rabbit:bindings>
   </rabbit:topic-exchange>
   
   <!-- 2. 定义正常交换机(order_exchange)和队列(order_queue)-->
   <rabbit:queue id="order_queue" name="order_queue">
       <!-- 3. 绑定，设置正常队列过期时间为10秒钟-->
       <rabbit:queue-arguments>
           <entry key="x-dead-letter-exchange" value="order_exchange_dlx" />
           <entry key="x-dead-letter-routing-key" value="dlx.order.cancel" />
           <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer" />
       </rabbit:queue-arguments>
   </rabbit:queue>
   
   <!-- 定义正常交换机 -->
   <rabbit:topic-exchange name="order_exchange">
       <rabbit:bindings>
           <rabbit:binding pattern="order.#" queue="order_queue"/>
       </rabbit:bindings>
   </rabbit:topic-exchange>
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14
   15
   16
   17
   18
   19
   20
   21
   22
   23
   24

2. 编写生产者代码

   @Autowired
   private RabbitTemplate rabbitTemplate;
   
   @Test
   public void testDelay() throws InterruptedException {
       //将消息发送给正常交换机
       rabbitTemplate.convertAndSend("order_exchange", "order.msg", "消息内容~");
   }
   1
   2
   3
   4
   5
   6
   7
   8

3. 编写消费者代码

   配置文件中让消费者监听器监听死信队列 order_queue_dlx，监听器类代码与之前类似，不再赘述

4. 运行结果

   消费者过了10秒之后成功读取消息