devbox
空白诗007
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

RabbitMQ - 死信队列，延时队列_rabbitmq 死信队列和延迟队列

作者：空白诗007 | 2024-08-05 00:47:02

赞

踩

rabbitmq 死信队列和延迟队列

Time-To-Live and Expiration — RabbitMQ

一、死信队列

Dead Letter Exchanges — RabbitMQ

死信队列：

DLX 全称（Dead-Letter-Exchange）,称之为死信交换器，当消息变成一个死信之后，如果这个消息所在的队列存在x-dead-letter-exchange参数，那么它会被发送到x-dead-letter-exchange对应值的交换器上，这个交换器就称之为死信交换器，与这个死信交换器绑定的队列就是死信队列

死信消息：

  • 消息被拒绝（Basic.Reject或Basic.Nack）并且设置 requeue 参数的值为 false
  • 消息过期（消息TTL过期。TTL：Time To Live的简称，即过期时间）
  • 队列达到最大的长度

过期消息：

在 rabbitmq 中存在2种方法可设置消息的过期时间:

  • 第一种通过对队列进行设置，这种设置后，该队列中所有的消息都存在相同的过期时间
  • 第二种通过对消息本身进行设置，那么每条消息的过期时间都不一样

如果同时使用这2种方法，那么以过期时间小的那个数值为准。当消息达到过期时间还没有被消费，那么那个消息就成为了一个 死信 消息

队列设置：在队列申明的时候使用** x-message-ttl **参数，单位为 毫秒；

  • 队列中这个属性的设置要在第一次声明队列的时候设置才有效，如果队列一开始已存在且没有这个属性，则要删掉队列再重新声明才可以。
  • 队列的 TTL 只能被设置为某个固定的值，一旦设置后则不能更改，否则会抛出异常

单个消息设置：是设置消息属性的 expiration 参数的值，单位为 毫秒。

说明：

对于第一种设置队列属性的方法，一旦消息过期，就会从队列中抹去；而在第二种方法中，即使消息过期，也不会马上从队列中抹去，因为每条消息是否过期是在即将投递到消费者之前判定的

 1. 生产者：
　　声明队列的时候用属性指定其死信队列交换机名称。

测试：

  1. package rabbitmq;
  2.  
  3. import java.io.IOException;
  4. import java.util.HashMap;
  5. import java.util.Map;
  6. import java.util.concurrent.TimeoutException;
  7.  
  8. import com.rabbitmq.client.BuiltinExchangeType;
  9. import com.rabbitmq.client.Channel;
  10. import com.rabbitmq.client.Connection;
  11. import com.rabbitmq.client.ConnectionFactory;
  12.  
  13. public class Producer {
  14.  
  15.     public static ConnectionFactory getConnectionFactory() {
  16.         // 创建连接工程，下面给出的是默认的case
  17.         ConnectionFactory factory = new ConnectionFactory();
  18.         factory.setHost("192.168.99.100");
  19.         factory.setPort(5672);
  20.         factory.setUsername("guest");
  21.         factory.setPassword("guest");
  22.         factory.setVirtualHost("/");
  23.         return factory;
  24.     }
  25.  
  26.     public static void main(String[] args) throws IOException, TimeoutException  {
  27.         ConnectionFactory connectionFactory = getConnectionFactory();
  28.         Connection newConnection = null;
  29.         Channel createChannel = null;
  30.         try {
  31.             newConnection = connectionFactory.newConnection();
  32.             createChannel = newConnection.createChannel();
  33.             
  34.             // 声明一个正常的direct类型的交换机
  35.             createChannel.exchangeDeclare("order.exchange", BuiltinExchangeType.DIRECT);
  36.             // 声明死信交换机为===order.dead.exchange
  37.             String dlxName = "order.dead.exchange";
  38.             createChannel.exchangeDeclare(dlxName, BuiltinExchangeType.DIRECT);
  39.             // 声明队列并指定死信交换机为上面死信交换机
  40.             Map<String, Object> arg = new HashMap<String, Object>();
  41.             arg.put("x-dead-letter-exchange", dlxName);
  42.             createChannel.queueDeclare("myQueue", true, false, false, arg);
  43.             
  44.             String message = "测试消息";
  45.             createChannel.basicPublish("order.exchange", "routing_key_myQueue", null, message.getBytes());
  46.             System.out.println("消息发送成功");
  47.         } catch (Exception e) {
  48.             e.printStackTrace();
  49.         } finally {
  50.             if (createChannel != null) {
  51.                 createChannel.close();
  52.             }
  53.             if (newConnection != null) {
  54.                 newConnection.close();
  55.             }
  56.         }
  57.         
  58.     }
  59. }

结果：

(1)生成两个Exchange

 (2)队列myQueue的死信队列有属性

2. 消费者： 
　　一个消费者监听正常队列，一个消费者监听死信队列。(只是绑定的交换机不同)

消费者一：监听正常队列

  1. package rabbitmq;
  2.  
  3. import java.io.IOException;
  4. import java.util.Date;
  5. import java.util.concurrent.TimeUnit;
  6. import java.util.concurrent.TimeoutException;
  7.  
  8. import com.rabbitmq.client.Channel;
  9. import com.rabbitmq.client.Connection;
  10. import com.rabbitmq.client.ConnectionFactory;
  11. import com.rabbitmq.client.DefaultConsumer;
  12. import com.rabbitmq.client.Envelope;
  13. import com.rabbitmq.client.AMQP.BasicProperties;
  14.  
  15. public class Consumer {
  16.  
  17.     public static ConnectionFactory getConnectionFactory() {
  18.         // 创建连接工程，下面给出的是默认的case
  19.         ConnectionFactory factory = new ConnectionFactory();
  20.         factory.setHost("192.168.99.100");
  21.         factory.setPort(5672);
  22.         factory.setUsername("guest");
  23.         factory.setPassword("guest");
  24.         factory.setVirtualHost("/");
  25.         return factory;
  26.     }
  27.  
  28.     public static void main(String[] args) throws IOException, TimeoutException {
  29.         ConnectionFactory connectionFactory = getConnectionFactory();
  30.         Connection newConnection = null;
  31.         Channel createChannel = null;
  32.         try {
  33.             newConnection = connectionFactory.newConnection();
  34.             createChannel = newConnection.createChannel();
  35.  
  36.             // 队列绑定交换机-channel.queueBind(队列名, 交换机名, 路由key[广播消息设置为空串])
  37.             createChannel.queueBind("myQueue", "order.exchange", "routing_key_myQueue");
  38.  
  39.             createChannel.basicConsume("myQueue", false, "", new DefaultConsumer(createChannel) {
  40.                 @Override
  41.                 public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
  42.                         byte[] body) throws IOException {
  43.  
  44.                     System.out.println("consumerTag: " + consumerTag);
  45.                     System.out.println("envelope: " + envelope);
  46.                     System.out.println("properties: " + properties);
  47.                     String string = new String(body, "UTF-8");
  48.                     System.out.println("接收到消息： -》 " + string);
  49.  
  50.                     long deliveryTag = envelope.getDeliveryTag();
  51.                     Channel channel = this.getChannel();
  52.                     System.out.println("拒绝消息， 使之进入死信队列");
  53.                     System.out.println("时间： " + new Date());
  54.                     try {
  55.                         TimeUnit.SECONDS.sleep(3);
  56.                     } catch (InterruptedException e) {
  57.                     }
  58.                     
  59.                     // basicReject第二个参数为false进入死信队列或丢弃
  60.                     channel.basicReject(deliveryTag, false);
  61.                 }
  62.             });
  63.  
  64.         } catch (Exception e) {
  65.             e.printStackTrace();
  66.         } finally {
  67.         }
  68.  
  69.     }
  70. }

消费者二：监听死信队列

  1. package rabbitmq;
  2.  
  3. import java.io.IOException;
  4. import java.util.Date;
  5. import java.util.concurrent.TimeoutException;
  6.  
  7. import com.rabbitmq.client.Channel;
  8. import com.rabbitmq.client.Connection;
  9. import com.rabbitmq.client.ConnectionFactory;
  10. import com.rabbitmq.client.DefaultConsumer;
  11. import com.rabbitmq.client.Envelope;
  12. import com.rabbitmq.client.AMQP.BasicProperties;
  13.  
  14. public class Consumer2 {
  15.  
  16.     public static ConnectionFactory getConnectionFactory() {
  17.         // 创建连接工程，下面给出的是默认的case
  18.         ConnectionFactory factory = new ConnectionFactory();
  19.         factory.setHost("192.168.99.100");
  20.         factory.setPort(5672);
  21.         factory.setUsername("guest");
  22.         factory.setPassword("guest");
  23.         factory.setVirtualHost("/");
  24.         return factory;
  25.     }
  26.  
  27.     public static void main(String[] args) throws IOException, TimeoutException {
  28.         ConnectionFactory connectionFactory = getConnectionFactory();
  29.         Connection newConnection = null;
  30.         Channel createChannel = null;
  31.         try {
  32.             newConnection = connectionFactory.newConnection();
  33.             createChannel = newConnection.createChannel();
  34.  
  35.             // 队列绑定交换机-channel.queueBind(队列名, 交换机名, 路由key[广播消息设置为空串])
  36.             createChannel.queueBind("myQueue", "order.dead.exchange", "routing_key_myQueue");
  37.  
  38.             createChannel.basicConsume("myQueue", false, "", new DefaultConsumer(createChannel) {
  39.                 @Override
  40.                 public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
  41.                         byte[] body) throws IOException {
  42.  
  43.                     System.out.println("时间： " + new Date());
  44.                     
  45.                     System.out.println("consumerTag: " + consumerTag);
  46.                     System.out.println("envelope: " + envelope);
  47.                     System.out.println("properties: " + properties);
  48.                     String string = new String(body, "UTF-8");
  49.                     System.out.println("接收到消息： -》 " + string);
  50.  
  51.                     long deliveryTag = envelope.getDeliveryTag();
  52.                     Channel channel = this.getChannel();
  53.                     channel.basicAck(deliveryTag, true);
  54.                     System.out.println("死信队列中处理完消息息");
  55.                 }
  56.             });
  57.  
  58.         } catch (Exception e) {
  59.             e.printStackTrace();
  60.         } finally {
  61.         }
  62.  
  63.     }
  64. }

结果： 消费者一先正常监听到，basicReject为false拒绝后进入死信队列；消费者二监听的死信队列收到消息。

消费者一打出的日志如下：

  1. consumerTag: amq.ctag-0noHs24F0FsGe-dfwwqWNw
  2. envelope: Envelope(deliveryTag=1, redeliver=false, exchange=order.exchange, routingKey=routing_key_myQueue)
  3. properties: #contentHeader<basic>(content-type=null, content-encoding=null, headers=null, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
  4. 接收到消息： -》 测试消息
  5. 拒绝消息， 使之进入死信队列
  6. 时间： Sat Nov 07 12:18:44 CST 2020

消费者二打出的日志如下：

  1. 时间： Sat Nov 07 12:18:47 CST 2020
  2. consumerTag: amq.ctag-ajYMpMFkXHDiYWkD3XFJ7Q
  3. envelope: Envelope(deliveryTag=1, redeliver=false, exchange=order.dead.exchange, routingKey=routing_key_myQueue)
  4. properties: #contentHeader<basic>(content-type=null, content-encoding=null, headers={x-death=[{reason=rejected, count=1, exchange=order.exchange, time=Sat Nov 07 01:52:19 CST 2020, routing-keys=[routing_key_myQueue], queue=myQueue}]}, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
  5. 接收到消息： -》 测试消息
  6. 死信队列中处理完消息息

注意：

　　进入死信队列之后，headers 加了一些死信相关的信息，包括原队列以及进入死信的原因。

补充：在队列进入死信队列之前也可以修改其routingKey,而且只有在指定x-dead-letter-exchange的前提下才能修改下面属性，否则会报错

(1)修改生产者声明队列的方式，如下：

  1. // 声明一个正常的direct类型的交换机
  2.             createChannel.exchangeDeclare("order.exchange", BuiltinExchangeType.DIRECT);
  3.             // 声明死信交换机为===order.dead.exchange
  4.             String dlxName = "order.dead.exchange";
  5.             createChannel.exchangeDeclare(dlxName, BuiltinExchangeType.DIRECT);
  6.             // 声明队列并指定死信交换机为上面死信交换机
  7.             Map<String, Object> arg = new HashMap<String, Object>();
  8.             arg.put("x-dead-letter-exchange", dlxName);
  9.             // 修改进入死信队列的routingkey，如果不修改会使用默认的routingKey
  10.             arg.put("x-dead-letter-routing-key", "routing_key_myQueue_dead");
  11.             createChannel.queueDeclare("myQueue", true, false, false, arg);

(2)修改监听死信队列的消费者二：

  1. // 队列绑定交换机-channel.queueBind(队列名, 交换机名, 路由key[广播消息设置为空串])
  2.             createChannel.queueBind("myQueue", "order.dead.exchange", "routing_key_myQueue_dead");

结果，收到消费者二收到的信息如下：

  1. 时间： Sat Nov 07 12:27:08 CST 2020
  2. consumerTag: amq.ctag-THqpEdYH_-iNeCIccgpuaw
  3. envelope: Envelope(deliveryTag=1, redeliver=false, exchange=order.dead.exchange, routingKey=routing_key_myQueue_dead)
  4. properties: #contentHeader<basic>(content-type=null, content-encoding=null, headers={x-death=[{reason=rejected, count=1, exchange=order.exchange, time=Sat Nov 07 02:00:41 CST 2020, routing-keys=[routing_key_myQueue], queue=myQueue}]}, delivery-mode=null, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
  5. 接收到消息： -》 测试消息
  6. 死信队列中处理完消息

二、延时队列

延迟队列，即消息进入队列后不会立即被消费，只有到达指定时间后，才会被消费

RabbitMQ本身没提供延时队列，我们可以利用消息的生存时间和死信队列实现延时

典型的应用场景就是订单30分钟内未支付就关闭订单，还有一种场景，账单24小时未确认，就发送提醒消息

延时队列插件安装

2.1.1、yml配置

  1. spring:
  2.     rabbitmq:
  3.         host: 192.168.99.12
  4.         port: 5672
  5.         username: guest
  6.         password: guest
  7.         # 发送确认
  8.         publisher-confirms: true
  9.         # 路由失败回调
  10.         publisher-returns: true
  11.         template:
  12.             # 必须设置成true 消息路由失败通知监听者，false 将消息丢弃
  13.             mandatory: true
  14.         #消费端
  15.         listener:
  16.             simple:
  17.                 # 每次从RabbitMQ获取的消息数量
  18.                 prefetch: 1
  19.                 default-requeue-rejected: false
  20.                 # 每个队列启动的消费者数量
  21.                 concurrency: 1
  22.                 # 每个队列最大的消费者数量
  23.                 max-concurrency: 1
  24.                 # 签收模式为手动签收-那么需要在代码中手动ACK
  25.                 acknowledge-mode: manual
  26. #邮件队列
  27. email:
  28.     queue:
  29.         name: demo.email
  30.  
  31. #邮件交换器名称
  32. exchange:
  33.     name: demoTopicExchange
  34.  
  35. #死信队列
  36. dead:
  37.     letter:
  38.         queue:
  39.             name: demo.dead.letter
  40.         exchange:
  41.             name: demoDeadLetterTopicExchange
  42.  
  43. #延时队列
  44. delay:
  45.     queue:
  46.         name: demo.delay
  47.     exchange:
  48.         name: demoDelayTopicExchange

2.1.2、延时队列配置

  1. /**
  2.  * rabbitmq 配置
  3.  *
  4.  * @author DUCHONG
  5.  * @since 2020-08-23 14:05
  6.  **/
  7. @Configuration
  8. @Slf4j
  9. public class RabbitmqConfig {
  10.  
  11.  
  12.     @Value("${email.queue.name}")
  13.     private String emailQueue;
  14.     @Value("${exchange.name}")
  15.     private String topicExchange;
  16.     @Value("${dead.letter.queue.name}")
  17.     private String deadLetterQueue;
  18.     @Value("${dead.letter.exchange.name}")
  19.     private String deadLetterExchange;
  20.     @Value("${delay.queue.name}")
  21.     private String delayQueue;
  22.     @Value("${delay.exchange.name}")
  23.     private String delayExchange;
  24.  
  25.     @Bean
  26.     public Queue emailQueue() {
  27.  
  28.         Map<String, Object> arguments = new HashMap<>(2);
  29.         // 绑定死信交换机
  30.         arguments.put("x-dead-letter-exchange", deadLetterExchange);
  31.         // 绑定死信的路由key
  32.         arguments.put("x-dead-letter-routing-key", deadLetterQueue+".#");
  33.  
  34.         return new Queue(emailQueue,true,false,false,arguments);
  35.     }
  36.  
  37.  
  38.     @Bean
  39.     TopicExchange emailExchange() {
  40.         return new TopicExchange(topicExchange);
  41.     }
  42.  
  43.  
  44.     @Bean
  45.     Binding bindingEmailQueue() {
  46.         return BindingBuilder.bind(emailQueue()).to(emailExchange()).with(emailQueue+".#");
  47.     }
  48.  
  49.  
  50.     //私信队列和交换器
  51.     @Bean
  52.     public Queue deadLetterQueue() {
  53.         return new Queue(deadLetterQueue);
  54.     }
  55.  
  56.     @Bean
  57.     TopicExchange deadLetterExchange() {
  58.         return new TopicExchange(deadLetterExchange);
  59.     }
  60.  
  61.     @Bean
  62.     Binding bindingDeadLetterQueue() {
  63.         return BindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange()).with(deadLetterQueue+".#");
  64.     }
  65.     //延时队列
  66.     @Bean
  67.     public Queue delayQueue() {
  68.         return new Queue(delayQueue);
  69.     }
  70.  
  71.     @Bean
  72.     CustomExchange delayExchange() {
  73.         Map<String, Object> args = new HashMap<>();
  74.         args.put("x-delayed-type", "topic");
  75.         //参数二为类型：必须是x-delayed-message
  76.         return new CustomExchange(delayExchange, "x-delayed-message", true, false, args);
  77.  
  78.     }
  79.  
  80.     @Bean
  81.     Binding bindingDelayQueue() {
  82.         return BindingBuilder.bind(delayQueue()).to(delayExchange()).with(delayQueue+".#").noargs();
  83.     }
  84. }

2.2、消息发送方

30分钟时间太久了，这里延时2分钟来看效果

  1. @Configuration
  2. @EnableScheduling
  3. @Slf4j
  4. public class ScheduleController {
  5.  
  6.     @Autowired
  7.     RabbitTemplate rabbitTemplate;
  8.  
  9.     @Value("${exchange.name}")
  10.     private String topicExchange;
  11.  
  12.     @Value("${delay.exchange.name}")
  13.     private String delayTopicExchange;
  14.  
  15.     @Scheduled(cron = "0 0/1 * * * ?")
  16.     public void sendEmailMessage() {
  17.  
  18.         String msg = RandomStringUtils.randomAlphanumeric(8);
  19.         JSONObject email=new JSONObject();
  20.         email.put("content",msg);
  21.         email.put("to","duchong@qq.com");
  22.         CorrelationData correlationData=new CorrelationData(UUID.randomUUID().toString());
  23.         rabbitTemplate.convertAndSend(topicExchange,"demo.email.x",email.toJSONString(),correlationData);
  24.         log.info("---发送 email 消息---{}---messageId---{}",email,correlationData.getId());
  25.     }
  26.  
  27.  
  28.     @Scheduled(cron = "0 0/1 * * * ?")
  29.     public void sendDelayOrderMessage() throws Exception{
  30.  
  31.         //订单号 id实际是保存订单后返回的，这里用uuid代替
  32.         String orderId = UUID.randomUUID().toString();
  33.         // 模拟订单信息
  34.         JSONObject order=new JSONObject();
  35.         order.put("orderId",orderId);
  36.         order.put("goodsName","vip充值");
  37.         order.put("orderAmount","99.00");
  38.         CorrelationData correlationData=new CorrelationData(orderId);
  39.         MessageProperties messageProperties = new MessageProperties();
  40.         messageProperties.setMessageId(orderId);
  41.         //30分钟时间太长，这里延时120s消费
  42.         messageProperties.setHeader("x-delay", 120000);
  43.         Message message = new Message(order.toJSONString().getBytes(CharEncoding.UTF_8), messageProperties);
  44.  
  45.         rabbitTemplate.convertAndSend(delayTopicExchange,"demo.delay.x",message,correlationData);
  46.  
  47.         log.info("---发送 order 消息---{}---orderId---{}",order,correlationData.getId());
  48.         //睡一会，为了看延迟效果
  49.         TimeUnit.MINUTES.sleep(10);
  50.     }
  51. }

2.3、消息消费方

  1. @Component
  2. @Slf4j
  3. public class MessageHandler {
  4.  
  5.  
  6.     /**
  7.      * 邮件发送
  8.      * @param message
  9.      * @param channel
  10.      * @param headers
  11.      * @throws IOException
  12.      */
  13.     @RabbitListener(queues ="demo.email")
  14.     @RabbitHandler
  15.     public void handleEmailMessage(Message message, Channel channel, @Headers Map<String,Object> headers) throws IOException {
  16.  
  17.         try {
  18.  
  19.             String msg=new String(message.getBody(), CharEncoding.UTF_8);
  20.             JSONObject jsonObject = JSON.parseObject(msg);
  21.             jsonObject.put("messageId",headers.get("spring_returned_message_correlation"));
  22.             log.info("---接受到消息---{}",jsonObject);
  23. 			//主动异常
  24. 			int m=1/0;
  25.             //手动签收
  26.             channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
  27.         }
  28.         catch (Exception e) {
  29.             log.info("handleEmailMessage捕获到异常,拒绝重新入队---消息ID---{}", headers.get("spring_returned_message_correlation"));
  30.             //异常，ture 重新入队,或者false,进入死信队列
  31.             channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);
  32.  
  33.         }
  34.     }
  35.  
  36.     /**
  37.      * 死信消费者，自动签收开启状态下，超过重试次数，或者手动签收，reject或者Nack
  38.      * @param message
  39.      */
  40.     @RabbitListener(queues = "demo.dead.letter")
  41.     public void handleDeadLetterMessage(Message message, Channel channel,@Headers Map<String,Object> headers) throws IOException {
  42.  
  43.         //可以考虑数据库记录，每次进来查数量，达到一定的数量，进行预警，人工介入处理
  44.         log.info("接收到死信消息:---{}---消息ID---{}", new String(message.getBody()),headers.get("spring_returned_message_correlation"));
  45. 		//回复ack
  46.         channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
  47.     }
  48.  
  49.     /**
  50.      * 延时队列消费
  51.      * @param message
  52.      * @param channel
  53.      * @param headers
  54.      * @throws IOException
  55.      */
  56.     @RabbitListener(queues ="demo.delay")
  57.     @RabbitHandler
  58.     public void handleOrderDelayMessage(Message message, Channel channel, @Headers Map<String,Object> headers) throws IOException {
  59.  
  60.         try {
  61.  
  62.             String msg=new String(message.getBody(), CharEncoding.UTF_8);
  63.             JSONObject jsonObject = JSON.parseObject(msg);
  64.             log.info("---接受到订单消息---orderId---{}",message.getMessageProperties().getMessageId());
  65.             log.info("---订单信息---order---{}",jsonObject);
  66.             //业务逻辑，根据订单id获取订单信息，如果还未支付，设置关闭状态，如果已支付，不做任何处理
  67.             //手动签收
  68.             channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
  69.         }
  70.         catch (Exception e) {
  71.             log.info("handleOrderDelayMessage捕获到异常,重新入队---orderId---{}", headers.get("spring_returned_message_correlation"));
  72.             //异常，ture 重新入队,或者false,进入死信队列
  73.             channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,true);
  74.  
  75.         }
  76.     }
  77.  
  78. }

2.4、结果

运行结果显示，同一个订单号的消息，发送过后2分钟，消费者才接受到，符合预期

https://www.cnblogs.com/geekdc/p/13550620.html

消息队列RabbitMQ（五）：死信队列与延迟队列

rabbitmq的延迟队列和死信队列_死信队列和延时队列的区别_zhuwenaptx的博客-CSDN博客

RabbitMQ的死信队列和延时队列 - 简书

RabbitMQ死信队列与延迟队列_51CTO博客_rabbitmq延迟队列

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/空白诗007/article/detail/930166
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号