Springboot死信队列 DLX 配置和使用思路分析

Springboot死信队列 DLX 配置和使用思路分析

目录前言什么是死信配置和测试死信思路分析配置类编写编写消息发送服务测试消息什么时候会成为死信消息？总结参考资料代码下载

前言

上一篇博客 Springboot——整合RabbitMq测试TTL中，针对设置单个消息期限或者整个队列消息期限，进行了一些配置和说明。同时也都列举了一些区别关系。

但考虑过一个问题了没有？

不管是设置哪种方式，如果消息期限到了，队列都会将该消息进行丢弃处理。这么做合适么？

假设是某个设备的重要信息，或者某个重要的订单信息，因为规定时间内未被及时消费就将其舍弃，是否会造成很严重的后果？

有人会说，设置消息永不过期！等着消费者能够成功监听到该队列，将消息消费不就可以了嘛！

但这里需要考虑另外一个问题：每个服务器的容量是有上限的！如果消息一直存在队列，如果一直不会被消费，岂不是很占用服务器资源？

如何解决这个问题，就是今天这篇文章需要说到的死信队列。

什么是死信

说道死信，可能大部分观众大姥爷会有懵逼的想法，什么是死信？

死信队列，俗称DLX，翻译过来的名称为Dead Letter Exchange 死信交换机。

当消息限定时间内未被消费，成为 Dead Message后，可以被重新发送到另一个交换机中，发挥其应有的价值！

配置和测试死信

思路分析

需要测试死信队列，则需要先梳理整体的思路，如可以采取如下方式进行配置：

从上面的逻辑图中，可以发现大致的思路：

1、消息队列分为正常交换机、正常消息队列；以及死信交换机和死信队列。

2、正常队列针对死信信息，需要将数据 重新 发送至死信交换机中。

配置类编写

结合上面的思路，编写具体的配置类。如下所示：

package cn.linkpower.config;

import org.springframework.amqp.core.*;

import org.springframework.beans.factory.annotation.Qualifier;

import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

/**

* 死信队列配置

*/

@Configuration

public class DeadMsgMqConfig {

// 定义正常交换机和正常队列信息(交换机名、队列名、路由key)

public static final String queue_name = "xj_natural_queue";

public static final String exchange_name = "xj_natural_exchange";

public static final String routing_key = "xj_natural_routingKey";

// 定义死信交换机名、死信队列名、路由key

public static final String queue_name_dead = "xj_dead_queue";

public static final String exchange_name_dead = "xj_dead_exchange";

public static final String routing_key_dead = "xj_dead_routingKey";

/**

* 设置正常的消息队列；

* 正常的消息队列具备以下几种功能：

* 1、消息正常消费，需要绑定对应的消费者(这里为了测试死信，不创建消费者)

* 2、当消息失效后，需要将指定的消息发送至 死信交换机 中

* @return

*/

@Bean(value = "getNaturalQueue")

public Queue getNaturalQueue(){

return QueueBuilder.durable(queue_name)

// 正常的队列，在消息失效后，需要将消息丢入 死信 交换机中

// 这里只需要针对名称进行绑定

.withArgument("x-dead-letter-exchange",exchange_name_dead)

// 丢入 死信交换机，需要设定指定的 routingkey

.withArgument("x-dead-letter-routing-key",routing_key_dead)

// 设置正常队列中消息的存活时间为 10s，当然也可以针对单个消息进行设定不同的过期时间

.withArgument("x-message-ttl",10000)

// 设定当前队列中，允许存放的最大消息数目

.withArgument("x-max-length",10)

.build();

}

* 设定正常的消息交换机

@Bean(value = "getNaturalExchange")

public Exchange getNaturalExchange(){

// 这里为了测试，采取 direct exchange

return ExchangeBuilder.directExchange(exchange_name)

.durable(true) // 设定持久化

* 将正常的消息交换机和正常的消息队列进行绑定

* @param queue

* @param directExchange

@Bean

public Binding bindNaturalExchangeAndQueue(

@Qualifier(value = "getNaturalQueue") Queue queue,

@Qualifier(value = "getNaturalExchange") Exchange directExchange

){

return BindingBuilder

// 绑定消息队列

.bind(queue)

// 至指定的消息交换机

.to(directExchange)

// 匹配 routingkey

.with(routing_key)

// 无参数，不加会报错提示

.noargs();

* 定义死信队列

@Bean(value = "getDealQueue")

public Queue getDealQueue(){

return QueueBuilder.durable(queue_name_dead).build();

* 定义死信交换机

@Bean(value = "getDeadExchange")

public Exchange getDeadExchange(){

return ExchangeBuilder.directExchange(exchange_name_dead).durable(true).build();

* 将死信交换机和死信队列进行绑定

* @param deadQueue

* @param directDeadExchange

public Binding bindDeadExchangeAndQueue(

@Qualifier(value = "getDealQueue") Queue deadQueue,

@Qualifier(value = "getDeadExchange") Exchange directDeadExchange

return BindingBuilder.bind(deadQueue).to(directDeadExchange).with(routing_key_dead).noargs();

}

编写消息发送服务

默认采取rabbitTemplate.convertAndSend方法，进行消息的发送处理。但为了保证消息生产者能够成功将数据发送至正常交换机，同时为了保证正常交换机能够将数据信息，推送至正常消息队列。需要对其增加监听。

package cn.linkpower.service;

import lombok.extern.slf4j.Slf4j;

import org.springframework.amqp.core.Message;

import org.springframework.amqp.rabbit.connection.CorrelationData;

import org.springframework.amqp.rabbit.core.RabbitTemplate;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.stereotype.Component;

@Slf4j

@Component

public class RabbitmqService implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {

@Autowired

private RabbitTemplate rabbitTemplate;

/**

* 直接发送消息

* @param exchange

* @param routingKey

* @param msg

*/

public void sendMessage(String exchange,String routingKey,Object msg) {

// 设置交换机处理失败消息的模式 true 表示消息由交换机 到达不了队列时，会将消息重新返回给生产者

// 如果不设置这个指令，则交换机向队列推送消息失败后，不会触发 setReturnCallback

rabbitTemplate.setMandatory(true);

//消息消费者确认收到消息后，手动ack回执

rabbitTemplate.setConfirmCallback(this);

// return 配置

rabbitTemplate.setReturnCallback(this);

//发送消息

rabbitTemplate.convertAndSend(exchange,routingKey,msg);

}

* 交换机并未将数据丢入指定的队列中时，触发

* channel.basicPublish(exchange_name,next.getKey(), true, properties,next.getValue().getBytes());

* 参数三：true 表示如果消息无法正常投递，则return给生产者 ；false 表示直接丢弃

* @param message 消息对象

* @paramhttp:// replyCode 错误码

* @param replyText 错误信息

* @param exchange 交换机

* @param routingKey 路由键

@Override

public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {

log.info("---- returnedMessage ----replyCode="+replyCode+" replyText="+replyText+" ");

* 消息生产者发送消息至交换机时触发，用RvXgRVP于判断交换机是否成功收到消息

* @param correlationData 相关配置信息

* @param ack exchange 交换机，判断交换机是否成功收到消息 true 表示交换机收到

* @param cause 失败原因

public void confirm(CorrelationData correlationData, boolean ack, String cause) {

log.info("---- confirm ----ack="+ack+" cause="+String.valueOf(cause));

log.info("correlationData -->"+correlationData.toString());

if(ack){

// 交换机接收到

log.info("---- confirm ----ack==true cause="+cause);

}else{

// 没有接收到

log.info("---- confirm ----ack==false cause="+cause);

}

}

测试

既然说到测试，那么需要编写一个测试类，能够将产生的消息，推送至指定的正常消息交换机中去。

package cn.linkpower.controller;

import cn.linkpower.config.DeadMsgMqConfig;

import cn.linkpower.service.RabbitmqService;

import lombok.extern.slf4j.Slf4j;

import org.apache.tomcat.jni.Time;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.web.bind.annotation.RequestMapping;

import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.TimeUnit;

@Slf4j

@RestController

public class DeadMsgController {

@Autowired

private RabbitmqService rabbitmqService;

@RequestMapping("/deadMsgTest")

public String deadMsgTest() throws InterruptedException {

// 向正常的消息队列中丢数据，测试限定时间未消费后，死信队列的情况

// 配置文件中，针对于正常队列而言，设置有10条上限大小

for (int i = 0; i < 20; i++) {

String msg = "dead msg test "+i;

log.info("发送消息，消息信息为：{}",msg);

// 向正常的消息交换机中传递数据

rabbitmqService.sendMessage(DeadMsgMqConfig.exchange_name,DeadMsgMqConfig.routing_key,msg);

TimeUnit.SECONDS.sleep(2);

}

return "ok";

}

}

启动项目，访问指定的链接，进行数据产生和将消息发送交换机操作：

http://localhost/deadMsgTest

控制台部分日志展示：

消息什么时候会成为死信消息？

1、队列消息长度到达限制；

2、消费者拒接消费消息，basicNack/basicReject,并且不把消息重新放入原目标队列,requeue=false；

channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);

3、原队列存在消息过期设置，消息到达超时时间未被消费；

总结

此处只是为了进行配置和测试需要，暂未定义任何正常消息队列消费者和死信消息队列消费者信息。

1、死信交换机和死信队列和普通的没有区别

2、当消息成为死信后，如果该队列绑定了死信交换机，则消息会被死信交换机重新路由到死信队列

参考资料

RabbitMQ死信队列在SpringBoot中的使用

代码下载

gitee 代码下载

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。