【Java教程】springcloud中RabbitMQ死信队列与延迟交换机如何实现详解

零

5229
文章

44
评论

2022 年 5 月 31 日15:25:10Java教程评论52字数 5891阅读19分38秒阅读模式2022 年 5 月 31 日

所需工具：

Java

聪明的大脑文章源自灵鲨社区-https://www.0s52.com/bcjc/javajc/12250.html

勤劳的双手文章源自灵鲨社区-https://www.0s52.com/bcjc/javajc/12250.html

文章源自灵鲨社区-https://www.0s52.com/bcjc/javajc/12250.html

注意：本站只提供教程，不提供任何成品+工具+软件链接，仅限用于学习和研究，禁止商业用途，未经允许禁止转载/分享等文章源自灵鲨社区-https://www.0s52.com/bcjc/javajc/12250.html

文章源自灵鲨社区-https://www.0s52.com/bcjc/javajc/12250.html

介绍

0.引言
1. 死信队列
1.2 什么是死信？
1.3 什么是死信队列？
1.4 创建死信交换机、死信队列
1.5 实现死信消息
1.5.1 基于消费者进行reject或nack实现死信消息
1.5.2 基于生存时间实现
1.5.3 基于队列max_length实现
1.6 基于死信队列实现消息延迟发送
基于死信队列实现消息延迟发送的问题
2. 延迟交换机
3. 应用场景
4. 练习题文章源自灵鲨社区-https://www.0s52.com/bcjc/javajc/12250.html

文章源自灵鲨社区-https://www.0s52.com/bcjc/javajc/12250.html

教程如下

0.引言

死信队列是消息队列中非常重要的概念，同时我们需要业务场景中都需要延迟发送的概念，比如12306中的30分钟后未支付订单取消。那么本期，我们就来讲解死信队列，以及如何通过延迟交换机来实现延迟发送的需求。文章源自灵鲨社区-https://www.0s52.com/bcjc/javajc/12250.html

1. 死信队列

1.2 什么是死信？

理解死信队列前，我们先讲解什么是死信，所谓死信就是没有被成功消费的消息，但并不是所有未成功消费的消息都是死信消息，死信消息的产生来源于以下三种途径：（1）消息被消费者拒绝，参数requeue设置为false的消息（2）过期的消息，过期消息分为两种： a. 发送消息时，设置了某一条消息的生存时间（message TTL），如果生存时间到了，消息还没有被消费，就会被标注为死信消息 b. 设置了队列的消息生存时间，针对队列中所有的消息，如果生存时间到了，消息还没有被消费，就会被标注为死信消息（3）当队列达到了最大长度后，再发送过来的消息就会直接变成死信消息文章源自灵鲨社区-https://www.0s52.com/bcjc/javajc/12250.html

1.3 什么是死信队列？

直接来讲，用来盛装死信的队列就是死信队列，好像是一句废话，所以其重点在于理解死信的概念。文章源自灵鲨社区-https://www.0s52.com/bcjc/javajc/12250.html

死信队列的作用：（1）队列在已满的情况下，会将消息发送到死信队列中，这样消息就不会丢失了，回头再从死信队列里将消息取出来进行消费即可（2）可以基于死信队列实现延迟消费的效果。具体的实现我们后续讲解

1.4 创建死信交换机、死信队列

死信交换机、死信队列其实都是普通的交换机、队列，只是专门声明出来用于存储死信消息的。我们只需要通过deadLetterExchange方法来声明死信交换机，然后用deadLetterRoutingKey方法来声明死信队列

如下代码所示，我们创建了test.queue、test.exchange及dead.queue、dead.exchange，并且在test.queue中将死信交换机和死信路由指定到了测试队列中

注意：涉及到修改队列、交换机属性的，如果该队列、交换机已经存在需要将其删除后才能生效，否则可能还会报错。

[php]

@Configuration
public class RabbitMqConfig {

private static final String TEST_EXCHANGE = "test.exchange";
private static final String TEST_QUEUE = "test.queue";
private static final String TEST_ROUTING_KEY = "test.routing.key";
private static final String DEAD_EXCHANGE = "dead.exchange";
private static final String DEAD_QUEUE = "dead.queue";
private static final String DEAD_ROUTING_KEY = "dead.routing.key";

@Bean
public Queue deadQueue(){
return new Queue(DEAD_QUEUE);
}
public DirectExchange deadExchange(){
// 设置演示，使用了直接交换机Direct，大家可以根据自己的业务情况声明为其他类型的交换机
return new DirectExchange(DEAD_EXCHANGE);
public Binding deadBinding(Queue deadQueue,Exchange deadExchange){
return BindingBuilder.bind(deadQueue).to(deadExchange).with(DEAD_ROUTING_KEY).noargs();
public Queue testQueue(){
return QueueBuilder.durable(TEST_QUEUE).deadLetterExchange(DEAD_EXCHANGE).deadLetterRoutingKey(DEAD_ROUTING_KEY).build();
public DirectExchange testExchange(){
return new DirectExchange(TEST_EXCHANGE);
public Binding testQueueBing(Queue testQueue, DirectExchange testExchange){
return BindingBuilder.bind(testQueue).to(testExchange).with(TEST_ROUTING_KEY);
}

[/php]

1.5 实现死信消息

1.5.1 基于消费者进行reject或nack实现死信消息

[php]

@Component
public class QueueListener {
@RabbitListener(queues = RabbitMqConfig.TEST_QUEUE)
public void handler(MyMessage messageInfo, Message message, Channel channel) {
try{
System.out.println("接收的消息："+messageInfo.toString());
// requeue参数设置为false 设置死信消息
channel.basicReject(message.getMessageProperties().getDeliveryTag(),false);
// multiple和requeue设置为false 设置死信消息
channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,false);
// 返回ack 确认接收到消息
// channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
}catch (IOException e){
try {
channel.basicRecover();
} catch (IOException ex) {
ex.printStackTrace();
log.error("消息处理失败：{}",e.getMessage());
}
}
}
}

[/php]

1.5.2 基于生存时间实现

（1）发送消息时设置生存时间

[php]

@GetMapping("sendTestQueueWithExpiration")
public String sendTestQueueWithExpiration(){
MyMessage message = new MyMessage(1L,"物流提醒","到达装货区域，注意上传凭证",new Date());
rabbitTemplate.convertAndSend(RabbitMqConfig.TEST_EXCHANGE,RabbitMqConfig.TEST_ROUTING_KEY, message,msg -> {
msg.getMessageProperties().setExpiration("5000");
return msg;
});
return "发送成功";
}

[/php]

（2）队列设置生存时间

[php]

@Bean
public Queue testQueue(){
return QueueBuilder.durable(TEST_QUEUE)
.deadLetterExchange(DEAD_EXCHANGE)
.deadLetterRoutingKey(DEAD_ROUTING_KEY)
// 10s 过期
.ttl(10000)
.build();
}

[/php]

1.5.3 基于队列max_length实现

[php]

@Bean
public Queue testQueue(){
return QueueBuilder.durable(TEST_QUEUE)
.deadLetterExchange(DEAD_EXCHANGE)
.deadLetterRoutingKey(DEAD_ROUTING_KEY)
// 容量最大100条
.maxLength(100)
.build();
}

[/php]

1.6 基于死信队列实现消息延迟发送

上述我们说过死信队列还可以消息延迟发送，其思路就是：（1）消息发送时设置消息的生存时间，其生存时间就是我们想要延迟的时间（2）消息者监控死信队列进行消费

正常队列的消息因为没有消费者消费，同时又指定了生存时间，到达时间后消息转发到死信队列中，消费者监听了死信队列从而将其消费掉。

基于死信队列实现消息延迟发送的问题

如果有两个消息，一个是5s生存时间，一个是10s生存时间，当我们先发送了10s生存时间的消息到queue中时，因为rabbitmq只会监控队列最外侧的消息的生存时间，也就是监控10s生存时间的消息，而5s生存时间的消息只会在最外侧的10s消息到期后才会监控，也就导致我实际需要5s生存的消息，实际需要10s才监听到了。

所以呢，基于死信队列实现的延迟消息，只使用于延迟时间一致的消息。

为了适配更多的延迟场景，已经更加简单的实现延迟消息，我们引入了延迟交换机

2. 延迟交换机

延迟交换机并不是rabbitmq自带的功能，而是要通过安装延迟交换机插件delayed_message_exchange来实现

其插件的安装我们之间已经讲解过，不再累叙，可以参考如下博文 springcloud：安装rabbitmq并配置延迟队列插件

通过延迟交换机实现的延迟消息，其重点主要在交换机上，队列就是普通队列，消息发送到交换机上后，会记录消息的延迟时间，到达时间后才会发送到队列中，这样消费者通过监控队列，就能在指定时间获取到消息

因此延迟交换机与普通交换机的实现，只在创建交换机时，其他的操作与普通交换机无异，因此使用起来也很方便

创建延迟交换机，通过x-delayed-type属性声明交换机类型，可以是direct也可以是topic，具体支持4中交换机类型，如果不清楚的可以参考之前的博文

[php]

@Configuration
public class RabbitMqDelayConfig {
public static final String DELAY_EXCHANGE = "delay.exchange";
public static final String DELAY_QUEUE = "delay.queue";
public static final String DELAY_ROUTING_KEY = "delay.routing.key";
@Bean
public Exchange delayExchange(){
Map<String, Object> arguments = new HashMap<>(1);
arguments.put("x-delayed-type","direct");
return new CustomExchange(DELAY_EXCHANGE,"x-delayed-message",true,false,arguments);
}
@Bean
public Queue delayQueue(){
return new Queue(DELAY_QUEUE);
}
@Bean
public Binding delayBinding(Queue delayQueue, Exchange delayExchange){
return BindingBuilder.bind(delayQueue).to(delayExchange).with(DELAY_ROUTING_KEY).noargs();
}

}

[/php]

发送消息时指定延迟时间，单位毫秒

[php]

rabbitTemplate.convertAndSend(DelayedConfig.DELAYED_EXCHANGE, "delayed.abc", "xxxx", new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
message.getMessageProperties().setDelay(30000);
return message;
}
});

[/php]

我们还可以将该方法封装为工具类方法，方便之后调用

[php]

/**
* 发送延迟队列
* @param exchange 交换机
* @param routeKey 路由
* @param message 消息
* @param delaySecond 延迟秒数
*/
public void send(String exchange, String routeKey, Object message, int delaySecond){
rabbitTemplate.convertAndSend(exchange,routeKey,message,msg -> {
// 消息持久化
msg.getMessageProperties().setDelay(delaySecond * 1000);
return msg;
});
}

[/php]

3. 应用场景

延迟消息的应用场景丰富，除了我们开篇所说的30分钟未支付自动取消订单，还比如到货后72小时未签收自动签收

基本上所有需要延迟触发的业务场景都可以用rabbitmq延迟队列来实现。

4. 练习题

对于刚接触rabbitmq的同学，这里我提供一个练习题给大家，也让大家在实操中加强对于rabbitmq的理解:

需求：订单到货后72小时未签收，自动签收讲解：我们这里要实现订单到货后的自动签收功能，订单到货后会触发发送自动签收消息的方法，订单已签收的状态status为2，到货状态为1，如果72小时前已经签收了即status被更新为2了，那么需要取消自动签收（不执行自动签收，即忽略自动签收消息）

历史上的今天

5 月

【Java教程】springcloud中RabbitMQ死信队列与延迟交换机如何实现详解

所需工具：

介绍

教程如下

0.引言

1. 死信队列

1.2 什么是死信？

1.3 什么是死信队列？

1.4 创建死信交换机、死信队列

1.5 实现死信消息

1.5.1 基于消费者进行reject或nack实现死信消息

1.5.2 基于生存时间实现

1.5.3 基于队列max_length实现

1.6 基于死信队列实现消息延迟发送

基于死信队列实现消息延迟发送的问题

2. 延迟交换机

3. 应用场景

4. 练习题

历史上的今天

Excel Export 踩坑注意点+导出方案设计

Spring Boot集成canal快速入门demo

Java 8 中 20 个高频面试题及答案

通过 GraalVM 将 Java 程序编译成本地机器码后，效率能和rust比吗？

常见的分布式ID解决方案

reduce规约：深入理解java8中的规约reduce

深入剖析Java继承机制：父类与子类的加载与初始化顺序

传输对象模式（Data Transfer Object Pattern）：电商平台商品信息传输实战案例分析

一文理解RSA算法

使用ThreadLocal存储用户登录信息

加载中...

发表评论

热门搜索

所需工具：

介绍

教程如下

0.引言

1. 死信队列

1.2 什么是死信？

1.3 什么是死信队列？

1.4 创建死信交换机、死信队列

1.5 实现死信消息

1.5.1 基于消费者进行reject或nack实现死信消息

1.5.2 基于生存时间实现

1.5.3 基于队列max_length实现

1.6 基于死信队列实现消息延迟发送

基于死信队列实现消息延迟发送的问题

2. 延迟交换机

3. 应用场景

4. 练习题

历史上的今天

发表评论