死信队列理解与使用

一、简介

在rabbitMQ中常用的交换机有三种，直连交换机、广播交换机、主题交换机；

直连交换机中队列与交换机需要约定好routingKey去进行绑定；

广播交换机并不需要routingKey绑定,只需队列与交换机绑定即可；

主题交换机最大的特点可以通过*和#去匹配队列；

而死信队列其实就是平常的队列的一种，通常我会使用直连交换机来作为死信队列；所以说，死信队列其实就是我们在处理业务中慢慢衍生出来的一个名词、一种方案；它和普通的队列是一样的。

二、使用场景

        我们知道在使用队列时有几种应答模式,比如自动应答（auto）、手动应答（manual）等，而在使用自动应答时，无论消息是否成功消费，达到重试次数后就会自动的把此消息给删除掉了，当然我们是不想把没有消费成功的消息给删除掉的。而开启手动应答时，配置的重试机制会失效 当有消费失败的消息时 会进入死循环。

        那么为了解决此场景，就引入了死信队列。当有不能正常消费的消息时 就把此消息给打到死信队列中，然后再根据实际情况去处理此信息。

关于自动应答和手动应答可参考这篇博客： 

rabbitMQ手动应答与自动应答_骑着蜗牛打天下的博客-CSDN博客

 

在 RabbitMQ 中充当主角的就是消息，在不同场景下，消息会有不同地表现。

死信就是消息在特定场景下的一种表现形式，这些场景包括：

1. 消息被拒绝访问，即 RabbitMQ返回 basicNack 的信号时。 或者拒绝basicReject

2. 消费者发生异常，超过重试次数 。

3. 消息的 TTL 过期时。

4. 消息队列达到最大长度。

三、代码实现

父pom文件

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.7.1</version>
<!--        <version>2.2.5.RELEASE</version>--><relativePath/> <!-- lookup parent from repository --></parent><groupId>com.chensir</groupId><artifactId>spring-boot-rabbitmq</artifactId><version>0.0.1-SNAPSHOT</version><name>spring-boot-rabbitmq</name><properties><java.version>8</java.version><hutool.version>5.8.3</hutool.version><lombok.version>1.18.24</lombok.version></properties><description>spring-boot-rabbitmq</description><packaging>pom</packaging><modules><module>direct-exchange</module><module>fanout-exchange</module><module>topic-exchange</module><module>game-exchange</module><module>dead-letter-queue</module><module>delay-queue</module><module>delay-queue2</module></modules><dependencyManagement><dependencies><dependency><groupId>cn.hutool</groupId><artifactId>hutool-all</artifactId><version>${hutool.version}</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>${lombok.version}</version></dependency></dependencies></dependencyManagement></project>

pom文件

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>com.chensir</groupId><artifactId>spring-boot-rabbitmq</artifactId><version>0.0.1-SNAPSHOT</version><relativePath>../pom.xml</relativePath></parent><artifactId>dead-letter-queue</artifactId><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>cn.hutool</groupId><artifactId>hutool-all</artifactId></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit-test</artifactId><scope>test</scope></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

配置文件

server.port=8084
#host
spring.rabbitmq.host=121.40.100.66
#默认5672
spring.rabbitmq.port=5672
#用户名
spring.rabbitmq.username=guest
#密码
spring.rabbitmq.password=guest
#连接到代理时用的虚拟主机
spring.rabbitmq.virtual-host=/
#每个消费者每次可最大处理的nack消息数量
spring.rabbitmq.listener.simple.prefetch=1
#表示消息确认方式，其有三种配置方式，分别是none、manual(手动)和auto(自动)；默认auto
spring.rabbitmq.listener.simple.acknowledge-mode=auto
#监听重试是否可用
spring.rabbitmq.listener.simple.retry.enabled=true
#最大重试次数
spring.rabbitmq.listener.simple.retry.max-attempts=5
#最大重试时间间隔
spring.rabbitmq.listener.simple.retry.max-interval=3000ms
#第一次和第二次尝试传递消息的时间间隔
spring.rabbitmq.listener.simple.retry.initial-interval=1000ms
#应用于上一重试间隔的乘数
spring.rabbitmq.listener.simple.retry.multiplier=2
#决定被拒绝的消息是否重新入队；默认是true（与参数acknowledge-mode有关系）
spring.rabbitmq.listener.simple.default-requeue-rejected=false

config

正常队列config

package com.chensir.config;import org.springframework.amqp.core.*;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitConfig {@Beanpublic MessageConverter messageConverter(){return  new Jackson2JsonMessageConverter();}@Beanpublic DirectExchange directExchange(){return  new DirectExchange("DirectExchange",true,false);}@Beanpublic Queue directQueueLong(){return   QueueBuilder.durable("DirectQueue").deadLetterExchange("DeadLetterExchange").deadLetterRoutingKey("dead")//20s还没消费就打到死信队列中.ttl(20000)//当队列中长度有500个消息,也打入死信队列.maxLength(500).build();}@Beanpublic Binding binding(){return  BindingBuilder.bind(directQueueLong()).to(directExchange()).with("direct123");}
}

死信队列config

package com.chensir.config;import org.springframework.amqp.core.*;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;/*** 死信队列 一般由运维在rebbitMQ服务创建交换机和队列 不需要代码配置*/
//@Configuration
public class DeadLetterConfig {@Beanpublic MessageConverter messageConverter(){return  new Jackson2JsonMessageConverter();}@Beanpublic DirectExchange directExchange() {DirectExchange directExchange = new DirectExchange("DeadLetterExchange");return directExchange;}@Beanpublic Queue queue() {Queue deadLetterQueue = QueueBuilder.durable("DeadLetterQueue").build();return deadLetterQueue;}@Beanpublic Binding binding() {Binding binding = BindingBuilder.bind(queue()).to(directExchange()).with("dead");return binding;}}

生产者

package com.chensir.provider;import cn.hutool.json.JSONUtil;
import com.chensir.model.OrderIngOk;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;import javax.annotation.Resource;@Component
public class DirectProvider {@Resourceprivate RabbitTemplate rabbitTemplate;public  void  send(){// 死信队列
//        rabbitTemplate.convertAndSend("DeadLetterExchange", "dead","123");for (int i=1;i<7;i++){OrderIngOk orderIngOk = new OrderIngOk();orderIngOk.setOrderNo("202308289687-"+i);orderIngOk.setId(i);orderIngOk.setUserName("倪海杉");
//            String s = JSONUtil.toJsonStr(orderIngOk);rabbitTemplate.convertAndSend("DirectExchange", "direct123",orderIngOk);}}}

消费者

正常队列消费者

package com.chensir.consumer;import cn.hutool.json.JSONUtil;
import com.chensir.model.OrderIngOk;
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;@Component
public class DirectConsumer {@RabbitHandler@RabbitListener(queues = "DirectQueue" )public void  process(OrderIngOk orderIngOk) throws IOException {try {// 处理业务开始if(orderIngOk.getId().equals(5)){int a =  0;int b= 2/a;}System.out.println("接受到消息,并正常处理结束"+ JSONUtil.toJsonStr(orderIngOk));} catch (Exception ex){System.out.println(ex.getMessage());System.out.println("接受到消息,发生异常"+ JSONUtil.toJsonStr(orderIngOk));//自动应答 当消费者成功消费消息时会自动把消息删除,而没有成功消费消息时需要给重试机制抛出个异常 重试机制才会开启重试throw ex;//手动模式//channel.basicReject(deliveryTag,true);//channel.basicNack(deliveryTag,false,true);}}
}

死信队列消费者

package com.chensir.consumer;import com.chensir.model.OrderIngOk;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;@Component
public class DeadConsumer {@RabbitHandler@RabbitListener(queues = "DeadLetterQueue")public void  process(OrderIngOk orderIngOk)  {System.out.println("这条信息在运行时发生了未知的异常，此信息被打到了死信队列,被死信队列消费者消费成功"+orderIngOk);}
}

结果

接受到消息,并正常处理结束{"id":1,"OrderNo":"202308289687-1","userName":"倪海杉"}
接受到消息,并正常处理结束{"id":2,"OrderNo":"202308289687-2","userName":"倪海杉"}
接受到消息,并正常处理结束{"id":3,"OrderNo":"202308289687-3","userName":"倪海杉"}
接受到消息,并正常处理结束{"id":4,"OrderNo":"202308289687-4","userName":"倪海杉"}
/ by zero
接受到消息,发生异常{"id":5,"OrderNo":"202308289687-5","userName":"倪海杉"}
/ by zero
接受到消息,发生异常{"id":5,"OrderNo":"202308289687-5","userName":"倪海杉"}
/ by zero
接受到消息,发生异常{"id":5,"OrderNo":"202308289687-5","userName":"倪海杉"}
/ by zero
接受到消息,发生异常{"id":5,"OrderNo":"202308289687-5","userName":"倪海杉"}
/ by zero
接受到消息,发生异常{"id":5,"OrderNo":"202308289687-5","userName":"倪海杉"}
2023-08-28 16:45:39.848  WARN 24432 --- [ntContainer#1-1] o.s.a.r.r.RejectAndDontRequeueRecoverer  : Retries exhausted for message (Body:'[B@1a6e663a(byte[58])' MessageProperties [headers={__TypeId__=com.chensir.model.OrderIngOk}, contentType=application/json, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=DirectExchange, receivedRoutingKey=direct123, deliveryTag=5, consumerTag=amq.ctag-f9Up1UES-F3rDvb-AK16xw, consumerQueue=DirectQueue])这条信息在运行时发生了未知的异常，此信息被打到了死信队列,被死信队列消费者消费成功OrderIngOk(id=5, OrderNo=202308289687-5, userName=倪海杉)
接受到消息,并正常处理结束{"id":6,"OrderNo":"202308289687-6","userName":"倪海杉"}