死信队列理解与使用

桃扇骨 2023-10-15 12:01 35阅读 0赞

## 一、简介 ##

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

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

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

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

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

## 二、使用场景 ##

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

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

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

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

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

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

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

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

3. 消息的 TTL 过期时。

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

## 三、代码实现 ##

![3e8b477521b64cfc812009c095a86121.png][]

### 父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>
    
            <relativePath/> 
        </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 {
    
        @Bean
        public MessageConverter messageConverter(){
            return  new Jackson2JsonMessageConverter();
        }
    
        @Bean
        public DirectExchange directExchange(){
            return  new DirectExchange("DirectExchange",true,false);
        }
    
        @Bean
        public Queue directQueueLong(){
            return   QueueBuilder.durable("DirectQueue")
                    .deadLetterExchange("DeadLetterExchange")
                    .deadLetterRoutingKey("dead")
                    //20s还没消费就打到死信队列中
                    .ttl(20000)
                    //当队列中长度有500个消息,也打入死信队列
                    .maxLength(500)
                    .build();
        }
    
        @Bean
        public 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 {
        @Bean
        public MessageConverter messageConverter(){
    
            return  new Jackson2JsonMessageConverter();
        }
    
        @Bean
        public DirectExchange directExchange() {
            DirectExchange directExchange = new DirectExchange("DeadLetterExchange");
            return directExchange;
        }
    
        @Bean
        public Queue queue() {
            Queue deadLetterQueue = QueueBuilder.durable("DeadLetterQueue").build();
            return deadLetterQueue;
        }
    
        @Bean
        public 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 {
    
        @Resource
        private 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":"倪海杉"}

[rabbitMQ_-CSDN]: https://blog.csdn.net/weixin_45326523/article/details/132531487
[3e8b477521b64cfc812009c095a86121.png]: https://img-blog.csdnimg.cn/3e8b477521b64cfc812009c095a86121.png