【RabbitMQ】 RabbitMQ高级特性（二）

文章目录

一、重试机制
- 1.1、重试配置
- 1.2、配置交换机&队列
- 1.3、发送消息
- 1.4、消费消息
- 1.5、运行程序
- 1.6、手动确认
二、TTL
- 2.1、设置消息的TTL
- 2.2、设置队列的TTL
- 2.3、两者区别
三、死信队列
- 6.1 死信的概念
- 3.2 代码示例
- - 3.2.1、声明队列和交换机
  - 3.2.2、正常队列绑定死信交换机
  - 3.2.3 制造死信产生的条件
  - 3.2.4、发送消息
  - 3.2.5、测试死信
- 3.3、常见面试题
四、延迟队列
- 4.1、概念
- 4.2、应用场景
- 4.3、TTL+死信队列实现
- 4.4、常见面试题
五、事务
- 5.1、配置事务管理器
- 5.2、声明队列
- 5.3、生产者
结语

本文延续上文RabbitMQ高级特性（一）为大家继续讲解RabbitMQ其他高级特性

在这里插入图片描述

一、重试机制

在消息传递过程中, 可能会遇到各种问题, 如网络故障, 服务不可用, 资源不足等, 这些问题可能导致消息处理失败. 为了解决这些问题, RabbitMQ 提供了重试机制, 允许消息在处理失败后重新发送. 但如果是程序逻辑引起的错误, 那么多次重试也是没有⽤的, 可以设置重试次数

1.1、重试配置

spring:
 rabbitmq:
	 addresses: amqp://study:study@110.41.51.65:15673/jiaohuan
	 listener:
		 simple:
			 acknowledge-mode: auto #消息接收确认
			 retry:
				 enabled: true # 开启消费者失败重试
				 initial-interval: 5000ms # 初始失败等待时⻓为5秒
				 max-attempts: 5 # 最⼤重试次数(包括⾃⾝消费的⼀次)

1.2、配置交换机&队列

   //重试机制
    public static final String RETRY_QUEUE = "retry_queue";
    public static final String RETRY_EXCHANGE_NAME = "retry_exchange";

    //重试机制 发布订阅模式
	//1. 交换机
    @Bean("retryExchange")
    public Exchange retryExchange() {
        return ExchangeBuilder.fanoutExchange(Constant.RETRY_EXCHANGE_NAME).durable(true).build();
    }

    //2. 队列
    @Bean("retryQueue")
    public Queue retryQueue() {
        return QueueBuilder.durable(Constant.RETRY_QUEUE).build();
    }

    //3. 队列和交换机绑定 Binding
    @Bean("retryBinding")
    public Binding retryBinding(@Qualifier("retryExchange") FanoutExchange
                                        exchange, @Qualifier("retryQueue") Queue queue) {
        return BindingBuilder.bind(queue).to(exchange);
    }

1.3、发送消息

@RequestMapping("/retry")
public String retry(){
 	rabbitTemplate.convertAndSend(Constant.RETRY_EXCHANGE_NAME, "", "retry test...");
 	return "发送成功!"; 
 }

1.4、消费消息

@Component
public class RetryQueueListener {
    //指定监听队列的名称
    @RabbitListener(queues = Constant.RETRY_QUEUE)
    public void ListenerQueue(Message message) throws Exception {
        System.out.printf("接收到消息: %s, deliveryTag: %d%n", new String(message.getBody(), "UTF-8"),
                message.getMessageProperties().getDeliveryTag());
        //模拟处理失败
        int num = 3 / 0;
        System.out.println("处理完成");
    }
}

1.5、运行程序

我们可以观察到结果

接收到消息: retry test..., deliveryTag: 1
接收到消息: retry test..., deliveryTag: 1
接收到消息: retry test..., deliveryTag: 1
接收到消息: retry test..., deliveryTag: 1
接收到消息: retry test..., deliveryTag: 1
o.s.a.r.r.RejectAndDontRequeueRecoverer : Retries exhausted for message (Body:'consumer ack test...' MessageProperties [headers={}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=ack_exchange, receivedRoutingKey=ack, deliveryTag=1, consumerTag=amq.ctag-vYckQBt9_0-5v2oG9oBnFw, consumerQueue=ack_queue])
org.springframework.amqp.rabbit.support.ListenerExecutionFailedException: 
Listener method 'public void com.jiaohuan.rabbitmq.listener.AckQueueListener.ListenerQueue(org.springframework.amqp.core.Message,com.rabbitmq.client.Channel) throws java.lang.Exception'threw exception

如果对异常进行捕获, 那么就不会进行重试代码修改如下：

System.out.printf("接收到消息: %s, deliveryTag: %d%n", new String(message.getBody(),"UTF-8"), 
message.getMessageProperties().getDeliveryTag());
//模拟处理失败
try {
	 int num = 3/0;
	 System.out.println("处理完成");
}catch (Exception e){
	 System.out.println("处理失败");
}

重新运行程序, 结果如下:

接收到消息: consumer ack test..., deliveryTag: 1
处理失败

1.6、手动确认

改为手动确认

  @RabbitListener(queues = Constant.RETRY_QUEUE)
    public void ListenerQueue(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            System.out.printf("接收到消息: %s, deliveryTag: %d%n", new String(message.getBody(), "UTF-8"),
                    message.getMessageProperties().getDeliveryTag());
            //模拟处理失败
            int num = 3 / 0;
            System.out.println("处理完成");
            //3. ⼿动签收
            channel.basicAck(deliveryTag, true);
        } catch (Exception e) {
            //4. 异常了就拒绝签收
            Thread.sleep(1000);
            //第三个参数requeue, 是否重新发送, 如果为true, 则会重新发送,,若为false, 则直接丢弃
            channel.basicNack(deliveryTag, true, true);
        }
    }

运⾏结果：

接收到消息: retry test..., deliveryTag: 1
接收到消息: retry test..., deliveryTag: 2
接收到消息: retry test..., deliveryTag: 3
接收到消息: retry test..., deliveryTag: 4
接收到消息: retry test..., deliveryTag: 5
接收到消息: retry test..., deliveryTag: 6
接收到消息: retry test..., deliveryTag: 7
接收到消息: retry test..., deliveryTag: 8
接收到消息: retry test..., deliveryTag: 9
接收到消息: retry test..., deliveryTag: 10
接收到消息: retry test..., deliveryTag: 11

可以看到, 手动确认模式时, 重试次数的限制不会像在自动确认模式下那样直接生效, 因为是否重试以及何时重试更多地取决于应⽤程序的逻辑和消费者的实现. ⾃动确认模式下, RabbitMQ 会在消息被投递给消费者后自动确认消息. 如果消费者处理消息时抛出异常, RabbitMQ 根据配置的重试参数自动将消息重新⼊队, 从而实现重试. 重试次数和重试间隔等参数可以直接在RabbitMQ的配置中设定，并且RabbitMQ会负责执行这些重试策略.
⼿动确认模式下, 消费者需要显式地对消息进行确认. 如果消费者在处理消息时遇到异常, 可以选择不确认消息使消息可以重新⼊队. 重试的控制权在于应用程序本身, 而不是RabbitMQ的内部机制. 应用程序可以通过自己的逻辑和利用RabbitMQ的⾼级特性来实现有效的重试策略。

在这里插入图片描述

使⽤重试机制时需要注意:
1 . ⾃动确认模式下: 程序逻辑异常, 多次重试还是失败, 消息就会被自动确认, 那么消息就丢失了
2 . ⼿动确认模式下: 程序逻辑异常, 多次重试消息依然处理失败, 无法被确认, 就⼀直是unacked的状态, 导致消息积压

二、TTL

TTL(Time to Live, 过期时间), 即过期时间. RabbitMQ可以对消息和队列设置TTL.当消息到达存活时间之后, 还没有被消费, 就会被⾃动清除。

咱们在⽹上购物, 经常会遇到⼀个场景, 当下单超过24⼩时还未付款, 订单会被⾃动取消还有类似的, 申请退款之后, 超过7天未被处理, 则⾃动退款

2.1、设置消息的TTL

目前有两种方法可以设置消息的TTL.
⼀是设置队列的TTL, 队列中所有消息都有相同的过期时间. ⼆是对消息本身进行单独设置, 每条消息的TTL可以不同. 如果两种方法⼀起使用, 则消息的TTL以两者之间较小的那个数值为准. 先看针对每条消息设置TTL。针对每条消息设置TTL的方法是在发送消息的方法中加入expiration的属性参数，单位为毫秒.

配置交换机&队列：


    //TTL
    public static final String TTL_QUEUE = "ttl_queue";
    public static final String TTL_EXCHANGE_NAME = "ttl_exchange";

    //ttl
	//1. 交换机
    @Bean("ttlExchange")
    public Exchange ttlExchange() {
        return ExchangeBuilder.fanoutExchange(Constant.TTL_EXCHANGE_NAME).durable(true).build();
    }

    //2. 队列
    @Bean("ttlQueue")
    public Queue ttlQueue() {
        return QueueBuilder.durable(Constant.TTL_QUEUE).build();
    }

    //3. 队列和交换机绑定 Binding
    @Bean("ttlBinding")
    public Binding ttlBinding(@Qualifier("ttlExchange") FanoutExchange exchange,
                              @Qualifier("ttlQueue") Queue queue) {
        return BindingBuilder.bind(queue).to(exchange);
    }

发送消息：

    @RequestMapping("/ttl")
    public String ttl() {
        String ttlTime = "10000";//10s
        rabbitTemplate.convertAndSend(Constant.TTL_EXCHANGE_NAME, "", "ttl 
                test...", messagePostProcessor -> {
        messagePostProcessor.getMessageProperties().setExpiration(ttlTime);
        return messagePostProcessor;
    });
        return"发送成功!";
    }

观看结果：

发送消息后, 可以看到, Ready消息为1：
在这里插入图片描述
10秒钟之后, 刷新页面, 发现消息已被删除：

如果不设置TTL，则表⽰此消息不会过期;如果将TTL设置为0，则表示除非此时可以直接将消息投递到消费者，否则该消息会被立即丢弃

2.2、设置队列的TTL

设置队列TTL的方法是在创建队列时, 加⼊ x-message-ttl 参数实现的, 单位是毫秒。

配置队列和绑定关系：

	public static final String TTL_QUEUE2 = "ttl_queue2";

    //设置ttl
    @Bean("ttlQueue2")
    public Queue ttlQueue2() {
        //设置20秒过期
        return QueueBuilder.durable(Constant.TTL_QUEUE2).ttl(20 * 1000).build();
    }

    //3. 队列和交换机绑定 Binding
    @Bean("ttlBinding2")
    public Binding ttlBinding2(@Qualifier("ttlExchange") FanoutExchange exchange,
                               @Qualifier("ttlQueue2") Queue queue) {
        return BindingBuilder.bind(queue).to(exchange);
    }

设置过期时间, 也可以采⽤以下方式:

 @Bean("ttlQueue2")
    public Queue ttlQueue2() {
        Map<String, Object> arguments = new HashMap<>();
        arguments.put("x-message-ttl", 20000);//20秒过期
        return QueueBuilder.durable(Constant.TTL_QUEUE2).withArguments(arguments).build();
    }

发送消息：

  @RequestMapping("/ttl")
    public String ttl() {
        // String ttlTime = "30000";//10s
        // //发送带ttl的消息
        // rabbitTemplate.convertAndSend(Constant.TTL_EXCHANGE_NAME, "", "ttl 
        test...", messagePostProcessor -> {
        // messagePostProcessor.getMessageProperties().setExpiration(ttlTime);
        // return messagePostProcessor;
        //});
        //发送不带ttl的消息
        rabbitTemplate.convertAndSend(Constant.TTL_EXCHANGE_NAME, "", "ttl 
                test...");
        return "发送成功!";
    }

运行程序：
运行之后发现,新增了⼀个队列, 队列Features有⼀个TTL标识：
在这里插入图片描述
调用接口, 发送消息:

发送消息后, 可以看到, Ready消息为1：
在这里插入图片描述

采⽤发布订阅模式, 所有与该交换机绑定的队列(ttl_queue和ttl_queue2)都会收到消息

20秒钟之后, 刷新页面, 发现消息已被删除
在这里插入图片描述
由于ttl_queue队列, 未设置过期时间, 所以ttl_queue的消息未删除。

2.3、两者区别

设置队列TTL属性的方法, ⼀旦消息过期, 就会从队列中删除设置消息TTL的方法, 即使消息过期, 也不会马上从队列中删除, 而是在即将投递到消费者之前进行判定的.
为什么这两种方法处理的方式不⼀样?
因为设置队列过期时间, 队列中已过期的消息肯定在队列头部, RabbitMQ只要定期从队头开始扫描是否有过期的消息即可. ⽽设置消息TTL的方式, 每条消息的过期时间不同, 如果要删除所有过期消息需要扫描整个队列, 所以不如等到此消息即将被消费时再判定是否过期, 如果过期再进⾏删除即可.

三、死信队列

6.1 死信的概念

死信(dead message) 简单理解就是因为种种原因, ⽆法被消费的信息, 就是死信. 有死信, ⾃然就有死信队列. 当消息在⼀个队列中变成死信之后，它能被重新被发送到另⼀个交换器中，这个交换器就是DLX( Dead Letter Exchange ), 绑定DLX的队列, 就称为死信队列(Dead Letter Queue，简称DLQ)
在这里插入图片描述
消息变成死信⼀般是由于以下几种情况:

消息被拒绝（ Basic.Reject/Basic.Nack )，并且设置 requeue 参数为 false.
消息过期.
队列达到最大长度

3.2 代码示例

3.2.1、声明队列和交换机

包含两部分:
• 声明正常的队列和正常的交换机
• 声明死信队列和死信交换机

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

    //死信队列
public static final String DLX_EXCHANGE_NAME = "dlx_exchange";
public static final String DLX_QUEUE = "dlx_queue";
public static final String NORMAL_EXCHANGE_NAME = "normal_exchange";
public static final String NORMAL_QUEUE = "normal_queue";
    
import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import rabbitmq.Constant;

/**
 * 死信队列相关配置
 */
@Configuration
public class DLXConfig {
    //死信交换机
    @Bean("dlxExchange")
    public Exchange dlxExchange() {
        return
                ExchangeBuilder.topicExchange(Constant.DLX_EXCHANGE_NAME).durable(true).build();
    }

    //2. 死信队列
    @Bean("dlxQueue")
    public Queue dlxQueue() {
        return QueueBuilder.durable(Constant.DLX_QUEUE).build();
    }

    //3. 死信队列和交换机绑定 Binding
    @Bean("dlxBinding")
    public Binding dlxBinding(@Qualifier("dlxExchange") Exchange exchange,
                              @Qualifier("dlxQueue") Queue queue) {
        return BindingBuilder.bind(queue).to(exchange).with("dlx").noargs();
    }

    //正常交换机
    @Bean("normalExchange")
    public Exchange normalExchange() {
        return ExchangeBuilder.topicExchange(Constant.NORMAL_EXCHANGE_NAME).durable(true).build();
    }

    //正常队列
    @Bean("normalQueue")
    public Queue normalQueue() {
        return QueueBuilder.durable(Constant.NORMAL_QUEUE).build();
    }

    //正常队列和交换机绑定 Binding
    @Bean("normalBinding")
    public Binding normalBinding(@Qualifier("normalExchange") Exchange
                                         exchange, @Qualifier("normalQueue") Queue queue) {
        return BindingBuilder.bind(queue).to(exchange).with("normal").noargs();
    }
}

3.2.2、正常队列绑定死信交换机

当这个队列中存在死信时, RabbitMQ会自动的把这个消息重新发布到设置的DLX上, 进而被路由到另一个队列, 即死信队列.可以监听这个死信队列中的消息以进⾏相应的处理

  @Bean("normalQueue")
    public Queue normalQueue() {
        Map<String, Object> arguments = new HashMap<>();
        arguments.put("x-dead-letter-exchange", Constant.DLX_EXCHANGE_NAME);//绑定死 信队列
        arguments.put("x-dead-letter-routing-key", "dlx");//设置发送给死信队列的
        RoutingKey
        return QueueBuilder.durable(Constant.NORMAL_QUEUE).withArguments(arguments).build();
    }

3.2.3 制造死信产生的条件

  @Bean("normalQueue")
    public Queue normalQueue() {
        Map<String, Object> arguments = new HashMap<>();
        arguments.put("x-dead-letter-exchange", Constant.DLX_EXCHANGE_NAME);//绑定死 信队列
        arguments.put("x-dead-letter-routing-key", "dlx");//设置发送给死信队列的
        RoutingKey
        //制造死信产⽣的条件
        arguments.put("x-message-ttl", 10000);//10秒过期
        arguments.put("x-max-length", 10);//队列⻓度
        return QueueBuilder.durable(Constant.NORMAL_QUEUE).withArguments(arguments).build();
    }

3.2.4、发送消息

@RequestMapping("/dlx")
 public void dlx() {
 //测试过期时间, 当时间达到TTL, 消息⾃动进⼊到死信队列
 rabbitTemplate.convertAndSend(Constant.NORMAL_EXCHANGE_NAME, "normal", "dlx test...");
	 //测试队列⻓度
	// for (int i = 0; i < 20; i++) {
	//  	rabbitTemplate.convertAndSend(Constant.NORMAL_EXCHANGE_NAME, "normal", "dlx test...");
	// }
	 //测试消息拒收
	// rabbitTemplate.convertAndSend(Constant.NORMAL_EXCHANGE_NAME, "normal", "dlx test...");
 }

3.2.5、测试死信

程序启动之后, 观察队列：
在这里插入图片描述

队列Features说明:
D: durable的缩写, 设置持久化
TTL: Time to Live, 队列设置了TTL
Lim: 队列设置了长度(x-max-length)
DLX: 队列设置了死信交换机(x-dead-letter-exchange)
DLK: 队列设置了死信RoutingKey(x-dead-letter-routing-key)

测试过期时间, 到达过期时间之后, 进⼊死信队列

发送之后：
在这里插入图片描述
10秒后, 消息进入到死信队列：

生产者首先发送⼀条消息,然后经过交换器(normal_exchange)顺利地存储到队列(normal_queue)中. 由于队列normal_queue设置了过期时间为10s, 在这10s内没有消费者消费这条消息, 那么判定这条消息过期. 由于设置了DLX, 过期之时, 消息会被丢给交换器(dlx_exchange)中, 这时根据RoutingKey匹配, 找到匹配的队列(dlx_queue), 最后消息被存储在queue.dlx这个死信队列中.

测试达到队列长度, 消息进入死信队列
队列⻓度设置为10, 我们发送20条数据, 会有10条数据直接进⼊到死信队列发送前, 死信队列只有⼀条数据

发送20条消息：

//测试队列⻓度
for (int i = 0; i < 20; i++) {
	 rabbitTemplate.convertAndSend(Constant.NORMAL_EXCHANGE_NAME, "normal", "dlx test...");
}

运⾏后, 可以看到死信队列变成了11条
在这里插入图片描述

过期之后, 正常队列的10条也会进入到死信队列

在这里插入图片描述

3.3、常见面试题

死信队列作为RabbitMQ的高级特性,也是面试的一大重点。

死信队列的概念死信（Dead Letter）是消息队列中的⼀种特殊消息, 它指的是那些无法被正常消费或处理的消息. 在消息队列系统中, 如RabbitMQ, 死信队列用于存储这些死信消息
死信的来源
1)消息过期: 消息在队列中存活的时间超过了设定的TTL
2)消息被拒绝: 消费者在处理消息时, 可能因为消息内容错误, 处理逻辑异常等原因拒绝处理该消息. 如果拒绝时指定不重新入队(requeue=false), 消息也会成为死信.
3)队列满了: 当队列达到最大长度, 无法再容纳新的消息时, 新来的消息会被处理为死信.
死信队列的应用场景对于RabbitMQ来说, 死信队列是⼀个非常有用的特性. 它可以处理异常情况下,消息不能够被消费者正确消费而被置⼊死信队列中的情况, 应用程序可以通过消费这个死信队列中的内容来分析当时所遇到的异常情况, 进而可以改善和优化系统. 比如: 用户支付订单之后, 支付系统会给订单系统返回当前订单的⽀付状态。为了保证支付信息不丢失, 需要使用到死信队列机制. 当消息消费异常时, 将消息投入到死信队列中, 由订单系统的其他消费者来监听这个队列, 并对数据进行处理(比如发送工单等,进行人工确认).
场景的应用场景还有:
• 消息重试：将死信消息重新发送到原队列或另⼀个队列进行重试处理.
• 消息丢弃：直接丢弃这些无法处理的消息，以避免它们占⽤系统资源.
• ⽇志收集：将死信消息作为日志收集起来，用于后续分析和问题定位.

四、延迟队列

4.1、概念

延迟队列(Delayed Queue)，即消息被发送以后, 并不想让消费者⽴刻拿到消息, 而是等待特定时间后, 消费者才能拿到这个消息进行消费.

4.2、应用场景

延迟队列的使用场景有很多, 比如:

智能家居: 用户希望通过手机远程遥控家⾥的智能设备在指定的时间进行⼯作. 这时候就可以将用户指令发送到延迟队列, 当指令设定的时间到了再将指令推送到智能设备.
⽇常管理: 预定会议后，需要在会议开始前十五分钟提醒参会⼈参加会议
⽤⼾注册成功后, 7天后发送短信, 提高用户活跃度等
…

RabbitMQ本身没有直接支持延迟队列的的功能, 但是可以通过前面所介绍的TTL+死信队列的⽅式组合模拟出延迟队列的功能. 假设⼀个应用中需要将每条消息都设置为10秒的延迟, 生产者通过 normal_exchange 这个交换器将发送的消息存储在 normal_queue 这个队列中. 消费者订阅的并非是 normal_queue 这个队列, 而是 dlx_queue 这个队列. 当消息从normal_queue 这个队列中过期之后被存入 dlx_queue 这个队列中，消费者就恰巧消费到了延迟10秒的这条消息.

所以死信队列展⽰的也是延迟队列的使用.

4.3、TTL+死信队列实现

代码实现：
先看TTL+死信队列实现延迟队列
继续沿用死信队列的代码即可
声明队列

  //正常队列
    @Bean("normalQueue")
    public Queue normalQueue() {
        Map<String, Object> arguments = new HashMap<>();
        arguments.put("x-dead-letter-exchange", Constant.DLX_EXCHANGE_NAME);//绑定死 信队列
        arguments.put("x-dead-letter-routing-key", "dlx");//设置发送给死信队列的
        RoutingKey
        return QueueBuilder.durable(Constant.NORMAL_QUEUE).withArguments(arguments).build();
    }

⽣产者:
发送两条消息, ⼀条消息10s后过期, 第二条20s后过期

    @RequestMapping("/delay")
    public String delay() {
        //发送带ttl的消息
        rabbitTemplate.convertAndSend(Constant.NORMAL_EXCHANGE_NAME, "normal",
                "ttl test 10s..." + new Date(), messagePostProcessor -> {
                    messagePostProcessor.getMessageProperties().setExpiration("10000");
					//10s过期
                    return messagePostProcessor;
                });
        rabbitTemplate.convertAndSend(Constant.NORMAL_EXCHANGE_NAME, "normal",
                "ttl test 20s..." + new Date(), messagePostProcessor -> {
                    messagePostProcessor.getMessageProperties().setExpiration("20000");
					//20s过期
                    return messagePostProcessor;
                });
        return "发送成功!";
    }

消费者:

//指定监听队列的名称
@RabbitListener(queues = Constant.DLX_QUEUE)
public void ListenerDLXQueue(Message message, Channel channel) throws Exception {
 	long deliveryTag = message.getMessageProperties().getDeliveryTag();
	 System.out.printf("%tc 死信队列接收到消息: %s, deliveryTag: %d%n", newDate(),newString(message.getBody(),"UTF-8"), 
	message.getMessageProperties().getDeliveryTag());
}

运行程序:
通过控制台观察死信队列消费情况:

死信队列接收到消息: ttl test 10s…Wed May 22 11:58:50 CST , deliveryTag: 1
死信队列接收到消息: ttl test 20s…Wed May 22 11:58:50 CST , deliveryTag: 2

可以看到, 两条消息按照过期时间依次进入了死信队列. 延迟队列, 就是希望等待特定的时间之后, 消费者才能拿到这个消息. TTL刚好可以让消息延迟⼀段时间成为死信, 成为死信的消息会被投递到死信队列⾥, 这样消费者⼀直消费死信队列里的消息就可以了.

存在问题

接下来把⽣产消息的顺序修改⼀下先发送20s过期数据, 再发送10s过期数据

 @RequestMapping("/delay")
    public String delay() {
        //发送带ttl的消息
        rabbitTemplate.convertAndSend(Constant.NORMAL_EXCHANGE_NAME, "normal",
                "ttl test 20s..." + new Date(), messagePostProcessor -> {
                    messagePostProcessor.getMessageProperties().setExpiration("20000");
					//20s过期
                    return messagePostProcessor;
                });
        rabbitTemplate.convertAndSend(Constant.NORMAL_EXCHANGE_NAME, "normal",
                "ttl test 10s..." + new Date(), messagePostProcessor -> {
                    messagePostProcessor.getMessageProperties().setExpiration("10000");
					//10s过期
                    return messagePostProcessor;
                });
        return "发送成功!";
   }

这时会发现: 10s过期的消息,，也是在20s后才进入到死信队列.
消息过期之后, 不⼀定会被马上丢弃. 因为RabbitMQ只会检查队首消息是否过期, 如果过期则丢到死信队列. 此时就会造成⼀个问题, 如果第⼀个消息的延时时间很长, 第二个消息的延时时间很短, 那第二个消息并不会优先得到执行.
所以在考虑使用TTL+死信队列实现延迟任务队列的时候, 需要确认业务上每个任务的延迟时间是⼀致的, 如果遇到不同的任务类型需要不同的延迟的话, 需要为每⼀种不同延迟时间的消息建⽴单独的消息队列。

另外注意，同样可以使用插件使得消息按照延迟时间到达消费者

4.4、常见面试题

延迟队列作为RabbitMQ的高级特性,也是面试的一大重点. 介绍下RabbitMQ的延迟队列。延迟队列是⼀个特殊的队列, 消息发送之后, 并不立即给消费者, 而是等待特定的时间, 才发送给消费者. 延迟队列的应用场景有很多, 比如:

订单在十分钟内未支付自动取消
用户注册成功后, 3天后发调查问卷
用户发起退款, 24小时后商家未处理, 则默认同意, 自动退款
…
但RabbitMQ本身并没直接实现延迟队列, 通常有两种方法:
1 . TTL+死信队列组合的方式
2 . 使用官方提供的延迟插件实现延迟功能

⼆者对比:

基于死信实现的延迟队列
a. 优点: 1) 灵活不需要额外的插件支持
b. 缺点: 1) 存在消息顺序问题 2) 需要额外的逻辑来处理死信队列的消息, 增加了系统的复杂性
基于插件实现的延迟队列
a. 优点: 1) 通过插件可以直接创建延迟队列, 简化延迟消息的实现. 2) 避免了DLX的时序问题
b. 缺点: 1) 需要依赖特定的插件, 有运维工作 2) 只适用特定版本

五、事务

RabbitMQ是基于AMQP协议实现的, 该协议实现了事务机制, 因此RabbitMQ也支持事务机制. SpringAMQP也提供了对事务相关的操作. RabbitMQ事务允许开发者确保消息的发送和接收是原子性的, 要么全部成功, 要么全部失败.

5.1、配置事务管理器

import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.transaction.RabbitTransactionManager;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class TransactionConfig {
    @Bean
    public RabbitTransactionManager
    transactionManager(CachingConnectionFactory connectionFactory) {
        return new RabbitTransactionManager(connectionFactory);
    }
    @Bean
    public RabbitTemplate rabbitTemplate(CachingConnectionFactory
                                                 connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setChannelTransacted(true);
        return rabbitTemplate;
    }
}

5.2、声明队列

@Bean("transQueue")
 public Queue transQueue() {
	 return QueueBuilder.durable("trans_queue").build();
 }

5.3、生产者

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RequestMapping("/trans")
@RestController
public class TransactionProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Transactional
    @RequestMapping("/send")
    public String send() {
        rabbitTemplate.convertAndSend("", "trans_queue", "trans test 1...");
        int a = 5 / 0;
        rabbitTemplate.convertAndSend("", "trans_queue", "trans test 2...");
        return "发送成功";
    }
}