问题发现问题解决方法一只监听死信队列在死信队列里面处理业务逻辑方法二修改预取值

问题发现

“死信”这意味着当以下四个事件中的任何一个发生时这些消息将被重新发布到

basic.reject

之前的文章里面有讲解过TTL过期后不进入死信队列的疑惑和解决办法然后上手去实践另一个死信队列的方法结果又是一道坑等着我示例代码如下

默认自动应答模式

DirectExchange(dead.exchange);}/***

死信队列和死信交换机绑定*

BindingBuilder.bind(deadQueue()).to(deadExchange()).with(dead);}/***

return*/Beanpublic

normalQueue.addArgument(x-dead-letter-exchange,

dead.exchange);

normalQueue.addArgument(x-dead-letter-routing-key,

dead);

normalQueue.addArgument(x-max-length,

5);//设置队列最大长度

normalQueue.addArgument(x-overflow,reject-publish);//最近发布的消息将被丢弃

方法二return

QueueBuilder.durable(normal_queue).deadLetterExchange(dead.exchange).deadLetterRoutingKey(dead).maxLength(5)

return*/Beanpublic

DirectExchange(normal.exchange);}/***

return*/Beanpublic

BindingBuilder.bind(queue()).to(normalExchange()).with(normal);}

Component

LoggerFactory.getLogger(MQReceiver.class);RabbitHandlerpublic

void

LoggerFactory.getLogger(MQReceiver2.class);RabbitHandlerpublic

void

world_i;template.convertAndSend(normal.exchange,

normal,

按道理会将队列前面的5条消息进入死信队列然后剩下的五条消息正常消费才对我们检查一下队列是否设置成功如图所示

问题解决

我们先在页面上向普通队交换机发送10条消息然后查看它的状态如图所示

我们可以看到普通队列存放的是最后5条消息前面的5条消息进入死信队列。

也就是说再没有进入普通消费者之前会将队列前面删除或死信消息进入消费者之前将消息进行分配。

这种做法也是网上大多数文章的一种处理方法另外一种情况就是进入普通消费者还没被消费完的情况下消费者挂了然后队列就会重新分配将从队列前面删除或者进入死信队列如图所示

但是这些做法都是基于没有普通消费者监听的情况下进行的感觉和我理解的略有偏差应该是再有普通消费者监听和死信队列监听的情况下发送消息时会对消息进行分配处理。

发送方代码和配置的代码就不重复展示了参考之前示例监听死信队列自动确认模式和手动确认模式都一样示例代码如下

Component

LoggerFactory.getLogger(MQReceiver2.class);RabbitHandlerpublic

void

{log.info(死信队列收到消息{},msg);channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);}

}调用send()方法执行结果如图

再监听普通队列时抛异常或者手动拒绝重新进入队列这两种方式都不能达到我想要的效果我发现只要有普通消费方监听就会进入消费然后我就在想是不是因为预期值的问题导致可能我消息数量太少了然后默认预期值太高导致消息直接进入消费然后将预取值改为1自动确认模式示例代码如下

spring.rabbitmq.listener.simple.prefetch1发送方代码和配置的代码就不重复展示了参考之前示例监听普通队列示例代码如下

Component

LoggerFactory.getLogger(MQReceiver.class);RabbitHandlerpublic

void

模拟业务处理队列等待场景Thread.sleep(10000);}

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

LoggerFactory.getLogger(MQReceiver2.class);RabbitHandlerpublic

void

当然有的时候也不一定完全按照你设置的最大长度进入死信队列有的时候消费速度太快队列的第一个已经被消费了的情况得看实际情况至少可以确保再设置了大于队列最大长度时是可以正常进入死信队列的。

归根结底消息数量太少了。

另外我们再来介绍一下溢出方式一般默认情况下溢出方式为drop-head从队列前面删除或者进入死信队列除此之外还有两种

reject-publish和reject-publish-dlx最近发布的消息将被丢弃。

reject-publish

reject-publish-dlx

DirectExchange(dead.exchange);}/***

死信队列和死信交换机绑定*

BindingBuilder.bind(deadQueue()).to(deadExchange()).with(dead);}/***

return*/Beanpublic

normalQueue.addArgument(x-dead-letter-exchange,

dead.exchange);

normalQueue.addArgument(x-dead-letter-routing-key,

dead);

normalQueue.addArgument(x-max-length,

5);//设置队列最大长度

normalQueue.addArgument(x-overflow,reject-publish);//最近发布的消息将被丢弃

方法二return

QueueBuilder.durable(normal_queue).deadLetterExchange(dead.exchange).deadLetterRoutingKey(dead).maxLength(5)

设置队列最大长度为5.overflow(QueueBuilder.Overflow.dropHead).build();}/***

return*/Beanpublic

DirectExchange(normal.exchange);}/***

return*/Beanpublic

BindingBuilder.bind(queue()).to(normalExchange()).with(normal);}

然后我们再来看看将溢出模式设置为reject-publish-dlxQueueBuilder.Overflow没有该参数手动定义示例代码如下

Configuration

Queue(normal_queue);normalQueue.addArgument(x-dead-letter-exchange,

dead.exchange);

死信队列normalQueue.addArgument(x-dead-letter-routing-key,

dead);

死信队列routingKeynormalQueue.addArgument(x-max-length,

5);//设置队列最大长度normalQueue.addArgument(x-overflow,reject-publish-dlx);

normalQueue;}//忽略普通队列创建过程参考前面示例...