办理RabbitMQ设置x-max-length队列最大长度后不进入死信队列

题目发现

近来再学习RabbitMQ过程中，看到关于死信队列内容：
   来自队列的消息可以是 “死信”，这意味着当以下四个事故中的任何一个发生时，这些消息将被重新发布到 Exchange。
  

• 利用 basic.reject 或 basic.nack 且 requeue 参数设置为 false 的利用者否定该消息
  
• 消息由于每条消息的 TTL 而逾期
  
• 队列超出了长度限定
  

  之前的文章内里有讲授过TTL逾期后不进入死信队列的疑惑息争决办法，然后上手去实践另一个死信队列的方法，结果又是一道坑等着我，示例代码如下：
默认自动应答模式

1. @Configuration
2. public class MQConfig {
3.     /**
4.      * 死信队列
5.      * @return
6.      */
7.     @Bean
8.     public Queue deadQueue(){
9.         return new Queue("dead_queue");
10.     }
11.     /**
12.      * 死信队列交换机
13.      * @return
14.      */
15.     @Bean
16.     public DirectExchange deadExchange(){
17.         return new DirectExchange("dead.exchange");
18.     }
20.     /**
21.      * 死信队列和死信交换机绑定
22.      * @return
23.      */
24.     @Bean
25.     public Binding deadBinding(){
26.         return BindingBuilder.bind(deadQueue()).to(deadExchange()).with("dead");
27.     }
29.     /**
30.      * 普通队列
31.      * @return
32.      */
33.     @Bean
34.     public Queue queue(){
35. //          方法一
36. //        Queue normalQueue = new Queue("normal_queue");
37. //        normalQueue.addArgument("x-dead-letter-exchange", "dead.exchange"); // 死信队列
38. //        normalQueue.addArgument("x-dead-letter-routing-key", "dead"); // 死信队列routingKey
39. //        normalQueue.addArgument("x-max-length", 5);//设置队列最大长度
40. //        normalQueue.addArgument("x-overflow","reject-publish");//最近发布的消息将被丢弃
41. //        方法二
42.         return  QueueBuilder.durable("normal_queue")
43.                 .deadLetterExchange("dead.exchange")
44.                 .deadLetterRoutingKey("dead")
45.                 .maxLength(5) // 设置队列最大长度为5
46.                 .build();
47.     }
48.     /**
49.      * 普通交换机
50.      * @return
51.      */
52.     @Bean
53.     public DirectExchange normalExchange(){
54.         return new DirectExchange("normal.exchange");
55.     }
57.     /**
58.      * 普通队列和普通交换机绑定
59.      * @return
60.      */
61.     @Bean
62.     public Binding binding(){
63.         return BindingBuilder.bind(queue()).to(normalExchange()).with("normal");
64.     }
65. }

复制代码

监听平凡队列消耗方，示例代码如下：

1. @Component
2. @RabbitListener(queues = "normal_queue")
3. public class MQReceiver {
4.     private static final Logger log = LoggerFactory.getLogger(MQReceiver.class);
6.     @RabbitHandler
7.     public void receive(String msg) {
8.         log.info("收到消息："+msg);
9.     }
10. }

复制代码

监听死信队列消耗方，示例代码如下：

1. @Component
2. @RabbitListener(queues = "dead_queue")
3. public class MQReceiver2 {
4.     private static final Logger log = LoggerFactory.getLogger(MQReceiver2.class);
6.     @RabbitHandler
7.     public void receive(String msg) {
8.         log.info("死信队列收到消息：{}",msg);
9.     }
10. }

复制代码

发送方发送10条消息，示例代码如下：

1. @Component
2. public class MQSender {
3.     @Autowired
4.     private RabbitTemplate template;
6.     public void send() {
7.         for (int i = 0; i < 10; i++) {
8.             String msg = "hello world_"+i;
9.             template.convertAndSend("normal.exchange", "normal", msg);
10.         }
11.     }
12. }

复制代码

然后调用send()方法，实行结果如图：

按原剖析将队列前面的5条消息进入死信队列，然后剩下的五条消息正常消耗才对，我们查抄一下队列是否设置乐成，如图所示

设置没有题目，那就很懵逼了。。。
题目办理

我们先在页面上向平凡队互换机发送10条消息，然后查察它的状态，如图所示：

凌驾5条消息就会放入死信队列中，如图所示：

然后再看一下，默认运动是从队列前面删除或死信消息，如图所示：

我们可以看到平凡队列存放的是末了5条消息，前面的5条消息进入死信队列。也就是说，再没有进入平凡消耗者之前会将队列前面删除或死信消息（进入消耗者之前将消息举行分配）。
方法一：只监听死信队列，在死信队列内里处理处罚业务逻辑

这种做法也是网上大多数文章的一种处理处罚方法（别的一种情况就是进入平凡消耗者，还没被消耗完的情况下，消耗者挂了，然后队列就会重新分配，将从队列前面删除大概进入死信队列），如图所示：

但是这些做法都是基于没有平凡消耗者监听的情况下举行的，感觉和我明确的略有毛病（应该是再有平凡消耗者监听和死信队列监听的情况下，发送消息时会对消息举行分配处理处罚）。
发送方代码和设置的代码就不重复展示了（参考之前示例），监听死信队列（自动确认模式和手动确认模式都一样），示例代码如下：

1. @Component
2. @RabbitListener(queues = "dead_queue")
3. public class MQReceiver2 {
4.     private static final Logger log = LoggerFactory.getLogger(MQReceiver2.class);
6.     @RabbitHandler
7.     public void receive(String msg,Message message,Channel channel) throws IOException {
8.         log.info("死信队列收到消息：{}",msg);
9.         channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
11.     }
12. }

复制代码

调用send()方法，实行结果如图：

然后再死信队列内里处理处罚业务逻辑即可。
方法二：修改预取值

再监听平凡队列时抛非常大概手动拒绝重新进入队列，这两种方式都不能到达我想要的结果，我发现只要有平凡消耗方监听就会进入消耗，然后我就在想是不是由于预期值的题目导致，大概我消息数目太少了，然后默认预期值太高导致消息直接进入消耗，然后将预取值改为1，自动确认模式，示例代码如下：

1. spring.rabbitmq.listener.simple.prefetch=1

复制代码

发送方代码和设置的代码就不重复展示了（参考之前示例），监听平凡队列，示例代码如下：

1. @Component
2. @RabbitListener(queues = "normal_queue")
3. public class MQReceiver {
4.     private static final Logger log = LoggerFactory.getLogger(MQReceiver.class);
6.     @RabbitHandler
7.     public void receive(String msg) throws InterruptedException {
8.         log.info("收到消息："+msg);
9.         // 模拟业务处理队列等待场景
10.         Thread.sleep(10000);
11.     }
12. }

复制代码

监听死信队列，示例代码如下：

1. @Component
2. @RabbitListener(queues = "dead_queue")
3. public class MQReceiver2 {
4.     private static final Logger log = LoggerFactory.getLogger(MQReceiver2.class);
6.     @RabbitHandler
7.     public void receive(String msg) {
8.         log.info("死信队列收到消息：{}",msg);
9.     }
10. }

复制代码

调用send()方法，实行结果如图：

固然有的时间也不愿定完全按照你设置的最大长度进入死信队列，有的时间消耗速率太快（队列的第一个已经被消耗了的情况），得看实际情况，至少可以确保再设置了大于队列最大长度时是可以正常进入死信队列的。归根结底：消息数目太少了。
别的我们再来先容一下溢出方式，一样平常默认情况下溢出方式为：drop-head（从队列前面删除大概进入死信队列），除此之外尚有两种：

• reject-publish和reject-publish-dlx：近来发布的消息将被抛弃。reject-publish 和 reject-publish-dlx 之间的区别在于 reject-publish-dlx 也是死信拒绝消息。
  

将设置的溢出模式改为reject-publish，示例代码如下：

1. @Configuration
2. public class MQConfig {
3.     /**
4.      * 死信队列
5.      * @return
6.      */
7.     @Bean
8.     public Queue deadQueue(){
9.         return new Queue("dead_queue");
10.     }
11.     /**
12.      * 死信队列交换机
13.      * @return
14.      */
15.     @Bean
16.     public DirectExchange deadExchange(){
17.         return new DirectExchange("dead.exchange");
18.     }
20.     /**
21.      * 死信队列和死信交换机绑定
22.      * @return
23.      */
24.     @Bean
25.     public Binding deadBinding(){
26.         return BindingBuilder.bind(deadQueue()).to(deadExchange()).with("dead");
27.     }
29.     /**
30.      * 普通队列
31.      * @return
32.      */
33.     @Bean
34.     public Queue queue() {
35. //          方法一
36. //        Queue normalQueue = new Queue("normal_queue");
37. //        normalQueue.addArgument("x-dead-letter-exchange", "dead.exchange"); // 死信队列
38. //        normalQueue.addArgument("x-dead-letter-routing-key", "dead"); // 死信队列routingKey
39. //        normalQueue.addArgument("x-max-length", 5);//设置队列最大长度
40. //        normalQueue.addArgument("x-overflow","reject-publish");//最近发布的消息将被丢弃
41. //        方法二
42.         return  QueueBuilder.durable("normal_queue")
43.                 .deadLetterExchange("dead.exchange")
44.                 .deadLetterRoutingKey("dead")
45.                 .maxLength(5) // 设置队列最大长度为5
46.                 .overflow(QueueBuilder.Overflow.dropHead)
47.                 .build();
48.     }
49.     /**
50.      * 普通交换机
51.      * @return
52.      */
53.     @Bean
54.     public DirectExchange normalExchange(){
55.         return new DirectExchange("normal.exchange");
56.     }
58.     /**
59.      * 普通队列和普通交换机绑定
60.      * @return
61.      */
62.     @Bean
63.     public Binding binding(){
64.         return BindingBuilder.bind(queue()).to(normalExchange()).with("normal");
65.     }
66. }

复制代码

重新创建队列，调用send()方法，如图所示：

由图可以，死信队列并不会受到消息。
然后我们再来看看将溢出模式设置为reject-publish-dlx（QueueBuilder.Overflow没有该参数，手动界说），示例代码如下：

1. @Configuration
2. public class MQConfig {
3.     //忽略死信队列创建和绑定过程，参考前面示例...
5.     /**
6.      * 普通队列
7.      * @return
8.      */
9.     @Bean
10.     public Queue queue() {
11.         Queue normalQueue = new Queue("normal_queue");
12.         normalQueue.addArgument("x-dead-letter-exchange", "dead.exchange"); // 死信队列
13.         normalQueue.addArgument("x-dead-letter-routing-key", "dead"); // 死信队列routingKey
14.         normalQueue.addArgument("x-max-length", 5);//设置队列最大长度
15.         normalQueue.addArgument("x-overflow","reject-publish-dlx"); //最近发布的消息进入死信队列
16.         return normalQueue;
17.     }
18.     //忽略普通队列创建过程，参考前面示例...
19. }

复制代码

重新创建队列，调用send()方法，如图所示：

如果你有更好的方法大概差别的明确，欢迎批评区交换。