SpringBoot教程（十五） | SpringBoot集成RabbitMq（死信队列、延迟队列）

（一）死信队列
- 使用场景
- 具体用法
- 前提
- 示例:
（二）延迟队列
- 使用场景
- 方法一：通过死亡队列实现
- 方法二：通过延迟消息插件（rabbitmq_delayed_message_exchange）实现

（一）死信队列

死信队列是一个重要的概念，用于处理那些因各种原因无法被正常消费的消息。
它不是RabbitMQ直接提供的一个现成的方法或工具，而是通过特定的配置和机制来实现的。

使用场景

死信队列在多种场景下都非常有用，包括但不限于：

消息重试机制：当消息处理失败时，可以将其发送到死信队列进行重试。
异常消息处理：对于无法被正常处理的异常消息，可以将其存储在死信队列中，以便后续分析处理。
延迟消息处理：通过结合消息的TTL（Time-To-Live，生存时间）和死信队列，可以实现消息的延迟处理。
确保消息不丢失：在消息处理过程中，如果发生消费者崩溃或网络故障等情况，消息可能会丢失。通过死信队列，可以确保这些消息得到保留，并在系统恢复后重新处理。

具体用法

要在RabbitMQ中设置和使用死信队列，通常需要按照以下步骤进行：

定义死信交换机（DLX）：首先，需要定义一个交换机作为死信交换机，它可以是任何类型的交换机（如direct、fanout、topic等）。
配置原队列：在声明原队列时，需要指定两个参数：x-dead-letter-exchange和x-dead-letter-routing-key。前者指定了当消息变成死信时应该发送到的交换机（即死信交换机），后者指定了发送到该交换机的路由键。
声明死信队列：接着，需要声明一个或多个死信队列，并将它们绑定到死信交换机上。这样，当死信消息被发送到死信交换机时，就可以根据路由键将其路由到相应的死信队列中。
处理死信消息：最后，需要编写消费者代码来监听死信队列中的消息，并对这些消息进行相应的处理。

前提

要想进入死信队列，得出现异常，出现异常后，会根据你的配置帮你放到死信队列中所以异常不要被捕获。
如果实在要捕获的话，就得你在消费者这边去做“发送消息的”操作，自己把发送过来消息塞到死信队列中

示例:

消费者 mq的yml配置（重试机制）

spring:rabbitmq:host: 127.0.0.1port: 5672username: guestpassword: guestlistener:simple:# 重试机制retry:enabled: true #是否开启消费者重试

配置类：

package com.example.reactboot.config;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.HashMap;
import java.util.Map;@Configuration
public class DirectExchangeConfig {//===========================普通===========================//定义队列的名称常量public static final String DIRECT_QUEUE = "directQueue";public static final String DIRECT_QUEUE2 = "directQueue2";//定义直接交换机的名称常量public static final String DIRECT_EXCHANGE = "directExchange";//定义路由键常量，用于交换机和队列之间的绑定public static final String DIRECT_ROUTING_KEY = "direct";//定义路由键常量，用于交换机和队列之间的绑定public static final String DIRECT_ROUTING_KEY_2= "direct2";//定义队列，名称为DIRECT_QUEUE//为普通队列 绑设置 死信参数@Beanpublic Queue directQueue() {//return new Queue(DIRECT_QUEUE, true);Map<String, Object> args = new HashMap<>();// 设置死信交换机args.put("x-dead-letter-exchange", DLX_EXCHANGE);// 设置发送到死信交换机的路由键args.put("x-dead-letter-routing-key", DLX_ROUTING_KEY);// 创建队列，设置为持久化、非排他、非自动删除，并附带死信参数return new Queue(DIRECT_QUEUE, true, false, false, args);}//定义直接交换机@Beanpublic DirectExchange directExchange() {return new DirectExchange(DIRECT_EXCHANGE, true, false);}//定义队列，名称为DIRECT_QUEUE2@Beanpublic Queue directQueue2() {return new Queue(DIRECT_QUEUE2, true);}//定义一个绑定，将directQueue队列绑定到directExchange交换机上，//使用direct作为路由键@Beanpublic Binding bindingDirectExchange(Queue directQueue, DirectExchange directExchange) {return BindingBuilder.bind(directQueue).to(directExchange).with(DIRECT_ROUTING_KEY);}// 定义一个绑定Bean，将directQueue2队列也绑定到directExchange交换机上，@Beanpublic Binding bindingDirectExchange2(Queue directQueue2, DirectExchange directExchange) {return BindingBuilder.bind(directQueue2).to(directExchange).with(DIRECT_ROUTING_KEY_2);}//===========================死信===========================// 定义死信交换机的名称public static final String DLX_EXCHANGE = "dlx_exchange";// 定义发送到死信交换机的路由键public static final String DLX_ROUTING_KEY = "dlx.routing.key";// 定义死信队列的名称public static final String DLX_QUEUE = "dlx_queue";/*** 声明死信交换机，这里使用Direct类型。* @return 返回一个配置好的DirectExchange对象。*/@BeanDirectExchange dlxExchange() {// 创建并返回Direct类型的交换机return new DirectExchange(DLX_EXCHANGE,true, false);}/*** 声明死信队列。* @return 返回一个配置好的Queue对象，用作死信队列。*/@BeanQueue dlxQueue() {// 创建并返回死信队列，设置为持久化return new Queue(DLX_QUEUE, true);}/*** 绑定死信队列到死信交换机，使用指定的路由键。*/@BeanBinding binding(Queue dlxQueue,DirectExchange dlxExchange) {return BindingBuilder.bind(dlxQueue).to(dlxExchange).with(DLX_ROUTING_KEY);}}

生产者发送消息：

package com.example.reactboot.controller;import com.example.reactboot.config.DirectExchangeConfig;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import java.nio.charset.StandardCharsets;@RestController
public class RabbitMqTest {@AutowiredRabbitTemplate rabbitTemplate;@RequestMapping("/sendMQ")public String sendMessage() {rabbitTemplate.convertAndSend(DirectExchangeConfig.DIRECT_EXCHANGE,DirectExchangeConfig.DIRECT_ROUTING_KEY, "发送一条测试消息：direct");return "direct消息发送成功！！";}}

消费者消费消息：

package com.example.reactboot.queueListener;import com.example.reactboot.config.DirectExchangeConfig;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;/*** @className: DirectQueueListener* @description: 直连交换机的监听器* @author: sh.Liu* @date: 2021-08-23 16:03*/
@Slf4j
@Component
public class DirectQueueListener {//监听普通队列@RabbitHandler@RabbitListener(queues = DirectExchangeConfig.DIRECT_QUEUE)public void process(String xx){SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");System.out.println("DirectReceiver消费者收到消息1  : " + xx + " 接收时间：" + sdf.format(new Date()) + "\n");//先执行业务代码int i = 1 / 0;}//监听死信队列@RabbitHandler@RabbitListener(queues = DirectExchangeConfig.DLX_QUEUE)public void process3(String testMessage) {System.out.println("死信得列里面的  : " + testMessage + "\n");}}

（二）延迟队列

延迟队列是一种特殊的消息队列，其内部消息是有序的，并且具有延时属性。
在RabbitMQ中，虽然AMQP协议本身没有直接支持延迟队列，但可以通过一些变通的方法（如使用死信队列配合消息的TTL属性，或者使用RabbitMQ的延迟消息插件）来实现延迟队列的功能。

使用场景

延迟队列在多种业务场景中都有广泛的应用，包括但不限于：

订单超时未支付自动取消：用户下单后，如果在规定时间内未完成支付，系统可以自动取消订单。
退款超时通知：用户申请退款后，如果长时间未得到处理，系统可以自动通知相关运营人员介入。
新用户注册后的引导邮件：用户注册账号后，系统可以在一段时间后发送欢迎邮件或引导邮件。
会议提醒：在预定的会议开始前一段时间，系统自动发送提醒给参会人员。
任务调度：在指定时间后执行某项任务，如定时清理日志、执行批处理任务等。

方法一：通过死亡队列实现

以下是使用死信队列配合TTL属性实现延迟队列的基本步骤：

定义死信交换机（DLX, Dead-Letter Exchange）和死信队列（DLQ, Dead-Letter Queue）
设置普通队列的TTL和死信交换机：在创建普通队列时，可以为其设置TTL属性，指定消息在该队列中的最大存活时间。同时，需要将该队列的死信交换机设置为前面定义的DLX，以便消息在过期后能够被发送到DLQ。
生产者发送消息：生产者将消息发送到普通队列，并指定消息的TTL。消息在队列中等待，直到TTL过期。
消息过期并发送到死信队列：当消息的TTL过期后，RabbitMQ会自动将该消息发送到其配置的死信交换机，再由死信交换机根据路由键将其发送到DLQ。
消费者从死信队列消费消息：消费者监听DLQ，当有新消息到达时，进行消费处理。

就是：把普通队列的消息设置存活时间，目前有两者方式：
1.在队列上面设置消息的过期时间
2.直接在消息上面设置过期时间。

方式一（队列上面设置消息过期时间）：

上面的关于死信示例完全可以复用进行测试

在以下的方法里面多加一行 args.put(“x-message-ttl”, 10000);

    //定义队列，名称为DIRECT_QUEUE//为普通队列 绑设置 死信参数@Beanpublic Queue directQueue() {//return new Queue(DIRECT_QUEUE, true);Map<String, Object> args = new HashMap<>();// 设置消息TTL为10秒args.put("x-message-ttl", 10000);// 设置死信交换机args.put("x-dead-letter-exchange", DLX_EXCHANGE);// 设置发送到死信交换机的路由键args.put("x-dead-letter-routing-key", DLX_ROUTING_KEY);// 创建队列，设置为持久化、非排他、非自动删除，并附带死信参数return new Queue(DIRECT_QUEUE, true, false, false, args);}

你可以把 DirectQueueListener 里面的 process 方法注释掉（以免被消费掉）。
再执行生产者的 sendMessage 方法。
这个时候你就可以看到下面关于监听死信队列的方法，等10秒后就会打印你发的消息了

方式二（消息上面设置过期时间）：

上面的关于死信示例完全可以复用进行测试

改一下这个生产者发送消息：

package com.example.reactboot.controller;import com.example.reactboot.config.DirectExchangeConfig;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import java.nio.charset.StandardCharsets;@RestController
public class RabbitMqTest {@AutowiredRabbitTemplate rabbitTemplate;@RequestMapping("/sendMQ")public String sendMessage() {SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");rabbitTemplate.convertAndSend(DirectExchangeConfig.DIRECT_EXCHANGE, DirectExchangeConfig.DIRECT_ROUTING_KEY, "发送一条测试消息：direct! "+sdf.format(new Date()), new MessagePostProcessor() {@Overridepublic Message postProcessMessage(Message message) throws AmqpException {//设置过期时间，超过5秒消息就会消失message.getMessageProperties().setExpiration("5000");//设置编码格式message.getMessageProperties().setContentEncoding("UTF-8");return message;}});     return "direct消息发送成功！！";}}