SpringBoot实践RabbitMQ小结：RabbitTemplate消息可靠性投递、死信队列使用

桃扇骨 2022-10-10 01:35 132阅读 0赞

### 文章目录 ###

*  1. RabbitMQ小结
 *  2. 引入依赖
 *  3. RabbitMQ配置
 *  4. 消息生产者与消费者
 *   *  4.1 生产者：发送字符串、实体类
     *  4.2 生产者：发送Message对象封装后的消息
     *   *  4.2.1 发送org.springframework.messaging.Message消息
         *  4.2.2 发送org.springframework.amqp.core.Message消息
     *  4.3 消费者手动ack确认
     *  4.4 死信队列处理拒绝签收的消息

# 1. RabbitMQ小结 #

下述代码主要实现：

1.  创建了消息转换器使用Jackson2, 这样消息在RabbitMQ的管理页面中就可以看到json结构的数据；
2.  配置了RabbitAdmin，主要用来创建交换机、队列、路由键；
3.  创建三种类型交换机：Direct、Fanout、Topic，同时创建多个队列与其绑定；
4.  配置了rabbitTemplate生产者的confirm与returns消息确认回调(注意配置yml文件)
5.  生产者发送的消息体：  
    （1）直接发送字符串String、对象Object  
    （2）消息体使用org.springframework.messaging.Message 封装发送;  
    （3）消息体使用org.springframework.amqp.core.Message 封装发送
6.  配置了rabbitTemplate消费者的手动ack确认
7.  死信队列处理消息未签收的消息

# 2. 引入依赖 #

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>

# 3. RabbitMQ配置 #

SpringBoot启动自动创建Queue、Exchange、Binding

server:
      port: 8080
    spring:
      #给项目来个名字
      application:
        name: rabbitmq-provider
      #配置rabbitMq 服务器
      rabbitmq:
        host: ip
        port: port
        username: name
        password: password
        publisher-returns: true
        #如果没有抵达队列,以异步发送优先回调returns确认
        template:
          mandatory: true
        publisher-confirm-type: simple

import com.qs.parentchild2common.entity.constant.RabbitConfigConstant;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.amqp.core.*;
    import org.springframework.amqp.rabbit.connection.ConnectionFactory;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitAdmin;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
    import org.springframework.amqp.support.converter.MessageConverter;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    import javax.annotation.PostConstruct;
    
    
    @Slf4j
    @Configuration
    public class RabbitMQConfig { 
    
        @Autowired
        private RabbitAdmin rabbitAdmin;
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        /** * 消息转换器修改 * 有了这项配置后, 消息在rabbitMq的管理页面就可以显示看到消息的json数据，否则是序列化的数据 * * @return */
        @Bean
        public MessageConverter jackJsonMessageConverter() { 
            return new Jackson2JsonMessageConverter();
        }
    
        /** * 配置rabbitAdmin * * @param connectionFactory * @return */
        @Bean
        public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) { 
            RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
            //只有设置为 true, spring才会加载RabbitAdmin这个类
            rabbitAdmin.setAutoStartup(true);
            return rabbitAdmin;
        }
    
        /** * 队列、交换机的创建 */
        @Bean
        public void createExchangeQueue() { 
            //direct类型交换机连接三个队列
            createDirectExchangeQueue();
            //fanout类型交换机
            createFanoutExchangeQueue();
            //topic类型交换机
            createTopicExchangeQueue();
        }
    
        /** * 配置rabbitTemplate消息确认回调 */
        @PostConstruct
        private void initRabbitTemplate() { 
            /** * 消息发送到交换机Exchange失败时, 回调 */
            rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() { 
                @Override
                public void confirm(CorrelationData correlationData, boolean ack, String cause) { 
                    if (ack) { 
                        log.info("确认ack:{} 消息correlationDataID:{}消息发送到交换机成功", ack, correlationData.getId());
                        return;
                    }
                    log.error("确认ack:{} 消息correlationDataID:{}发送到交换机失败 case:{}", ack, correlationData.getId(), cause);
                }
            });
            /** * 消息从交换机Exchange-->队列Queue失败时, 优先回调 */
            rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() { 
                @Override
                public void returnedMessage(ReturnedMessage returned) { 
                    Integer replyCode = returned.getReplyCode();
                    String replyText = returned.getReplyText();
                    String exchange = returned.getExchange();
                    String routingKey = returned.getRoutingKey();
                    Message message = returned.getMessage();
                    log.error("correlationDataID:{}发送到队列失败 replyCode:{} replyText:{} exchange:{} routingKey:{}",
                            message.getMessageProperties().getHeaders().get("spring_returned_message_correlation"),
                            replyCode, replyText, exchange, routingKey);
                }
            });
        }
    
        void createDirectExchangeQueue() { 
            DirectExchange directExchange = directExchange();
            Queue queue1 = queue1();
            Queue queue2 = queue2();
            Queue queue3 = queue3();
            Binding binding1 = BindingBuilder.bind(queue1).to(directExchange).with("queue1Key");
            Binding binding2 = BindingBuilder.bind(queue2).to(directExchange).with("queue1Key");
            Binding binding3 = BindingBuilder.bind(queue3).to(directExchange).with("queue3Key");
    
            rabbitAdmin.declareExchange(directExchange);
            rabbitAdmin.declareQueue(queue1);
            rabbitAdmin.declareQueue(queue2);
            rabbitAdmin.declareQueue(queue3);
    
            rabbitAdmin.declareBinding(binding1);
            rabbitAdmin.declareBinding(binding2);
            rabbitAdmin.declareBinding(binding3);
        }
    
        void createFanoutExchangeQueue() { 
            FanoutExchange fanoutExchange = fanoutExchange();
            Queue queue4 = queue4();
            Queue queue5 = queue5();
            Queue queue6 = queue6();
            Binding binding4 = BindingBuilder.bind(queue4).to(fanoutExchange);
            Binding binding5 = BindingBuilder.bind(queue5).to(fanoutExchange);
            Binding binding6 = BindingBuilder.bind(queue6).to(fanoutExchange);
    
            rabbitAdmin.declareExchange(fanoutExchange);
            rabbitAdmin.declareQueue(queue4);
            rabbitAdmin.declareQueue(queue5);
            rabbitAdmin.declareQueue(queue6);
    
            rabbitAdmin.declareBinding(binding4);
            rabbitAdmin.declareBinding(binding5);
            rabbitAdmin.declareBinding(binding6);
        }
    
        void createTopicExchangeQueue() { 
            TopicExchange topicExchange = topicExchange();
            Queue queue4 = queue4();
            Queue queue5 = queue5();
            Queue queue6 = queue6();
            Binding binding4 = BindingBuilder.bind(queue4).to(topicExchange).with(RabbitConfigConstant.QUS_ROUTING_TOPIC_KEY_1);
            Binding binding5 = BindingBuilder.bind(queue5).to(topicExchange).with(RabbitConfigConstant.QUS_ROUTING_TOPIC_KEY_2);
            Binding binding6 = BindingBuilder.bind(queue6).to(topicExchange).with(RabbitConfigConstant.QUS_ROUTING_TOPIC_KEY_3);
    
            rabbitAdmin.declareExchange(topicExchange);
            rabbitAdmin.declareQueue(queue4);
            rabbitAdmin.declareQueue(queue5);
            rabbitAdmin.declareQueue(queue6);
    
            rabbitAdmin.declareBinding(binding4);
            rabbitAdmin.declareBinding(binding5);
            rabbitAdmin.declareBinding(binding6);
        }
    
        /** * 队列 * durable:是否持久化,默认是false,持久化队列：会被存储在磁盘上，当消息代理重启时仍然存在，暂存队列：当前连接有效 * exclusive:默认也是false，只能被当前创建的连接使用，而且当连接关闭后队列即被删除。此参考优先级高于durable * autoDelete:是否自动删除，当没有生产者或者消费者使用此队列，该队列会自动删除。 * * @return */
        private Queue queue1() { 
            return new Queue(RabbitConfigConstant.QUS_QUEUE_1, true);
        }
    
        private Queue queue2() { 
            return new Queue(RabbitConfigConstant.QUS_QUEUE_2, true);
        }
    
        private Queue queue3() { 
            return new Queue(RabbitConfigConstant.QUS_QUEUE_3, true);
        }
    
        private Queue queue4() { 
            return new Queue(RabbitConfigConstant.QUS_QUEUE_4, true);
        }
    
        private Queue queue5() { 
            return new Queue(RabbitConfigConstant.QUS_QUEUE_5, true);
        }
    
        private Queue queue6() { 
            return new Queue(RabbitConfigConstant.QUS_QUEUE_6, true);
        }
    
        //交换机
        private DirectExchange directExchange() { 
            return new DirectExchange(RabbitConfigConstant.DIRECT_EXCHANGE, true, false);
        }
    
        private FanoutExchange fanoutExchange() { 
            return new FanoutExchange(RabbitConfigConstant.FANOUT_EXCHANGE, true, false);
        }
    
        private TopicExchange topicExchange() { 
            return new TopicExchange(RabbitConfigConstant.TOPIC_EXCHANGE, true, false);
        }
    }

# 4. 消息生产者与消费者 #

## 4.1 生产者：发送字符串、实体类 ##

1.  通用的消息推送封装

import com.qs.parentchild2common.entity.constant.RabbitConfigConstant;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    
    /** * @author QuS * @date 2021/6/25 13:46 */
    @Service
    public class RabbitMQProducer { 
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        /** * @param object 消息体 */
        public void sendMessageToMQ(Object object, String msgId) { 
            CorrelationData correlationData = new CorrelationData(msgId);
            rabbitTemplate.convertAndSend(RabbitConfigConstant.TOPIC_EXCHANGE, "123.goods", object, correlationData);
        }
    }

1.  消息发送测试方法

@RequestMapping("/sendTopicMessage")
        public String sendTopicMessage(Integer num) { 
            RabbitMQBO rabbitMQBO;
            UserBO userBO;
            for (int i = 0; i < num; i++) { 
                String id = "ID" + i;
                if (i % 2 == 0) { 
                    rabbitMQBO = new RabbitMQBO();
                    rabbitMQBO.setId(i);
                    rabbitMQBO.setUuid(UUID.randomUUID().toString());
                    rabbitMQBO.setMsg("这是生产的消息" + i);
                    rabbitMQProducer.sendMessageToMQ(rabbitMQBO, id);
                }
    
                if (i % 2 == 1) { 
                    userBO = new UserBO();
                    userBO.setAge(20 + i);
                    userBO.setPassword("password" + i);
                    userBO.setUsername("小明" + i);
                    rabbitMQProducer.sendMessageToMQ(userBO, id);
                }
    
                rabbitMQProducer.sendMessageToMQ("测试字符串" + i, id);
    
            }
            return "ok";
        }

1.  消费者

注意消费者也需要配置和生产者相同的消息转换器：Jackson2JsonMessageConverter()

import com.alibaba.fastjson.JSONObject;
    import com.qs.parentchild2common.entity.bo.RabbitMQBO;
    import com.qs.parentchild2common.entity.bo.UserBO;
    import com.qs.parentchild2common.entity.constant.RabbitConfigConstant;
    import com.rabbitmq.client.Channel;
    import org.springframework.amqp.core.ExchangeTypes;
    import org.springframework.amqp.rabbit.annotation.*;
    import org.springframework.messaging.handler.annotation.Headers;
    import org.springframework.messaging.handler.annotation.Payload;
    import org.springframework.stereotype.Component;
    
    import java.io.IOException;
    import java.util.Map;
    
    
    /** * @author QuS * @date 2021/6/22 14:20 */
    @Component
    @RabbitListener(bindings = @QueueBinding(
            exchange = @Exchange(value = RabbitConfigConstant.TOPIC_EXCHANGE, type = ExchangeTypes.TOPIC),
            value = @Queue(value = RabbitConfigConstant.QUS_QUEUE_4, durable = "true"),
            key = "#.goods"
    ))
    public class RabbitMQConsumer1 { 
        /** * 如果发送端使用的是Message，接收端使用@RabbitHandler注解接受Message会报错 * * @param * @param channel * @throws IOException */
        @RabbitHandler
        public void process1(@Payload RabbitMQBO rabbitMQBO, @Headers Map<String, Object> headers, Channel channel) throws IOException { 
            System.out.println("消费者_1: RabbitMQBO: " + JSONObject.toJSONString(rabbitMQBO) + " Header: " + JSONObject.toJSONString(headers));
        }
    
        @RabbitHandler
        public void process2(@Payload String str, @Headers Map<String, Object> headers) { 
            System.out.println("消费者_2: String: " + JSONObject.toJSONString(str) + " Header: " + JSONObject.toJSONString(headers));
        }
    
        @RabbitHandler
        public void process3(@Payload UserBO userBO, @Headers Map<String, Object> headers) { 
            System.out.println("消费者_3: UserBO: " + JSONObject.toJSONString(userBO) + " Header:" + JSONObject.toJSONString(headers));
        }
    }

1.  输出结果

生产者回调输出

2021-06-25 14:31:25.924  INFO 21992 --- [nectionFactory2] c.q.p.config.RabbitMQConfig              : 确认ack:true 消息correlationDataID:ID0消息发送到交换机成功
    2021-06-25 14:31:25.924  INFO 21992 --- [nectionFactory1] c.q.p.config.RabbitMQConfig              : 确认ack:true 消息correlationDataID:ID0消息发送到交换机成功
    2021-06-25 14:31:25.926  INFO 21992 --- [nectionFactory1] c.q.p.config.RabbitMQConfig              : 确认ack:true 消息
    ......

消费者输出

消费者_1: RabbitMQBO: {"id":0,"msg":"这是生产的消息0","uuid":"8b8c385c-d0d6-4176-9ffc-7421d8215b6c"} Header: {"amqp_receivedDeliveryMode":"PERSISTENT","amqp_receivedExchange":"qus_topic_exchange","amqp_deliveryTag":1,"amqp_consumerQueue":"qus-queue-4","amqp_redelivered":false,"amqp_receivedRoutingKey":"123.goods","amqp_contentEncoding":"UTF-8","spring_listener_return_correlation":"449d172c-89be-455e-9d4a-107967a067e3","spring_returned_message_correlation":"ID0","id":"f7701b20-565c-2487-5223-eb66bfdb3d0c","amqp_consumerTag":"amq.ctag-hLAmFEcm6YIyp-hXlicXqA","amqp_lastInBatch":false,"contentType":"application/json","__TypeId__":"com.qs.parentchild2common.entity.bo.RabbitMQBO","timestamp":1624602705686}
    消费者_2: String: "测试字符串0" Header: {"amqp_receivedDeliveryMode":"PERSISTENT","amqp_receivedExchange":"qus_topic_exchange","amqp_deliveryTag":2,"amqp_consumerQueue":"qus-queue-4","amqp_redelivered":false,"amqp_receivedRoutingKey":"123.goods","amqp_contentEncoding":"UTF-8","spring_listener_return_correlation":"449d172c-89be-455e-9d4a-107967a067e3","spring_returned_message_correlation":"ID0","id":"caccf01c-6cfa-adcf-e11a-6030caa42251","amqp_consumerTag":"amq.ctag-hLAmFEcm6YIyp-hXlicXqA","amqp_lastInBatch":false,"contentType":"application/json","__TypeId__":"java.lang.String","timestamp":1624602705738}
    消费者_3: UserBO: {"age":21,"password":"password1","username":"小明1"} Header:{"amqp_receivedDeliveryMode":"PERSISTENT","amqp_receivedExchange":"qus_topic_exchange","amqp_deliveryTag":3,"amqp_consumerQueue":"qus-queue-4","amqp_redelivered":false,"amqp_receivedRoutingKey":"123.goods","amqp_contentEncoding":"UTF-8","spring_listener_return_correlation":"449d172c-89be-455e-9d4a-107967a067e3","spring_returned_message_correlation":"ID1","id":"fa5b8cdb-d092-0e58-8c28-712cae04078f","amqp_consumerTag":"amq.ctag-hLAmFEcm6YIyp-hXlicXqA","amqp_lastInBatch":false,"contentType":"application/json","__TypeId__":"com.qs.parentchild2common.entity.bo.UserBO","timestamp":1624602705741}
    ......

1.  小结  
    （1）@RabbitListener可以提前声明queue、exchange、routingKey的绑定关系，如果RabbitMQ中不存在，在启动消费者时会自动创建  
    （2）@RabbitListener配合@RabbitHandler使用时，可以通过指定不同类型的消息类型，从而编写不同的方法去获取消息体  
    （3）消息如果没有抵达队列，会以异步发送优先回调returns确认  
    （4）上述已经实现了消息–>RabbitMQ的可靠性投递，接下来会说消费者的ack手动确认

## 4.2 生产者：发送Message对象封装后的消息 ##

### 4.2.1 发送org.springframework.messaging.Message消息 ###

注意：封装该类型消息发送时，RabbitTemplate的消息转换器使用Jackson2JsonMessageConverter会报错，当我注释掉该消息转换器，使用默认的消息转换器时，正常运行

1.  通用的消息封装

import com.qs.parentchild2common.entity.constant.RabbitConfigConstant;
    import org.springframework.amqp.rabbit.connection.CorrelationData;
    import org.springframework.amqp.rabbit.core.RabbitTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.messaging.Message;
    import org.springframework.stereotype.Service;
    
    /** * @author QuS * @date 2021/6/25 13:46 */
    @Service
    public class RabbitMQProducer { 
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        /** * @param message 消息体 */
        public void sendMessageToMQ(Message message, String msgId) { 
            CorrelationData correlationData = new CorrelationData(msgId);
            rabbitTemplate.convertAndSend(RabbitConfigConstant.TOPIC_EXCHANGE, "order.123", message, correlationData);
        }
    }

1.  生产者代码

@RequestMapping("/sendMessage1")
        public String sendMessage(Integer num) { 
            RabbitMQBO rabbitMQBO;
            UserBO userBO;
            for (int i = 0; i < num; i++) { 
                String id = "ID" + i;
                Message message;
                if (i % 2 == 0) { 
                    rabbitMQBO = new RabbitMQBO();
                    rabbitMQBO.setId(i);
                    rabbitMQBO.setUuid(UUID.randomUUID().toString());
                    rabbitMQBO.setMsg("这是生产的消息" + i);
                    Map<String, Object> headers = new HashMap<>();
                    headers.put("testId", "testId" + i);
                    message = new GenericMessage(rabbitMQBO, headers);
    
                    rabbitMQProducer.sendMessageToMQ(message, id);
                }
            }
            return "ok";
        }

1.  消费者代码

import com.alibaba.fastjson.JSONObject;
    import com.qs.parentchild2common.entity.constant.RabbitConfigConstant;
    import com.rabbitmq.client.Channel;
    import org.springframework.amqp.rabbit.annotation.RabbitListener;
    import org.springframework.messaging.Message;
    import org.springframework.messaging.MessageHeaders;
    import org.springframework.stereotype.Component;
    import java.io.IOException;
    
    /** * @author QuS * @date 2021/6/28 14:49 */
    @Component
    public class RabbitMQConsumer2 { 
    
        @RabbitListener(queues = { RabbitConfigConstant.QUS_QUEUE_5})
        private void consumeMessage(Message message, Channel channel) throws IOException { 
            Object payload = message.getPayload();
            System.out.println("payload:>> " + JSONObject.toJSONString(payload));
            MessageHeaders headers = message.getHeaders();
            long deliveryTag = (long) headers.get("amqp_deliveryTag");
            System.out.println("deliveryTag:>> " + deliveryTag + " testId:>> " + headers.get("testId"));
            System.out.println("headers:>> " + JSONObject.toJSONString(headers));
            channel.basicAck(deliveryTag, false);
        }
    }

1.  小结  
    （1）org.springframework.messaging.Message作为消息体发消息时，消息转换器使用Jackson2JsonMessageConverter会报错，使用默认的消息转换器不会报错  
    （2）使用该消息体，可以自定义拓展封装Header

### 4.2.2 发送org.springframework.amqp.core.Message消息 ###

该消息体的封装需要转换为字节数组，消费端也需要由字节数组转换为对应的类型，个人不太喜欢，感兴趣的伙伴可以研究使用

## 4.3 消费者手动ack确认 ##

默认情况下，消费端是自动ack确认，如果消费端处理出现异常，外加自动ack确认，会导致处理异常的消息丢失，所以我们可改成手动ack，在程序异常时，对消息进行处理，防止消息丢失。

1.  配置文件修改，开启手动ack

spring:
      application:
        name: rabbitmq-consumer1
      rabbitmq:
        host: host
        port: port
        username: username
        password: password
        listener:
          simple:
            acknowledge-mode: manual

1.  代码手动ack

channel.basicAck(deliveryTag, false);

@RabbitListener(queues = { RabbitConfigConstant.QUS_QUEUE_5})
        private void consumeMessage(Message message, Channel channel) { 
            try { 
                Object payload = message.getPayload();
                System.out.println("payload:>> " + JSONObject.toJSONString(payload));
                MessageHeaders headers = message.getHeaders();
                long deliveryTag = (long) headers.get("amqp_deliveryTag");
                System.out.println("deliveryTag:>> " + deliveryTag + " testId:>> " + headers.get("testId"));
                System.out.println("headers:>> " + JSONObject.toJSONString(headers));
                channel.basicAck(deliveryTag, false);
            } catch (Exception e) { 
                //异常处理
                log.error("出现异常:{}", e.toString());
                channel.basicNack(deliveryTag, false, false);
            }
        }

## 4.4 死信队列处理拒绝签收的消息 ##

消费者消费过程出现异常，nack后进入死信队列

1.  进入死信队列的过程

生产者 --> 发送消息 --> 业务交换机 --> 队列 --> 成为死信消息 --> 死信交换机 --> 死信队列 --> 监听死信队列的消费者

1. 正常业务消息被投递到正常业务的Exchange，该Exchange根据路由键将消息路由到绑定的正常队列。
    2. 正常业务队列中的消息变成了死信消息之后，会被自动投递到该队列绑定的死信交换机上（并带上配置的路由键，如果没有指定死信消息的路由键，则默认继承该消息在正常业务时设定的路由键）。
    3. 死信交换机收到消息后，将消息根据路由规则路由到指定的死信队列。
    4. 消息到达死信队列后，可监听该死信队列，处理死信消息。

1.  消息成为死信消息的场景  
    （1）消息被(reject 或 nack ) 并且 requeue = false，即消息被消费者拒绝签收，并且重新入队为false；  
    （2）消息过期，过了ttl存活时间；  
    （3）队列设置了x-max-length最大消息数量且当前队列中的消息已经达到了这个数量，再次投递，消息将被挤掉，被挤掉的是最靠近被消费那一端的消息。
2.  主要给业务队列添加"x-dead-letter-exchange" 和 "x-dead-letter-routing-key"参数去指定其死信交换机和死信路由键
3.  业务队列绑定死信队列代码示例

@Slf4j
    @Configuration
    public class RabbitMQConfig { 
    
        @Autowired
        private RabbitAdmin rabbitAdmin;
    
        /** * 队列、交换机的创建 */
        @Bean
        public void createExchangeQueue() { 
            //死信交换机、死信队列
            createDeadLetterExchangeQueue();
        }
    
        void createDeadLetterExchangeQueue() { 
            DirectExchange deadLetterExchange = deadLetterExchange();
            Queue deadLetterQueue1 = deadLetterQueue1();
            Binding binding = BindingBuilder.bind(deadLetterQueue1).to(deadLetterExchange).with(RabbitConfigConstant.DEAD_LETTER_QUEUE1_ROUTING_KEY);
    
            rabbitAdmin.declareExchange(deadLetterExchange);
            rabbitAdmin.declareQueue(deadLetterQueue1);
            rabbitAdmin.declareBinding(binding);
    
        }
        
    	//业务队列添加死信交换机参数
        private Queue queue1() { 
            Queue queue1 = QueueBuilder.durable(RabbitConfigConstant.QUS_QUEUE_1)
                    .withArgument("x-dead-letter-exchange", RabbitConfigConstant.DEAD_LETTER_DIRECT_EXCHANGE)
                    .withArgument("x-dead-letter-routing-key", RabbitConfigConstant.DEAD_LETTER_QUEUE1_ROUTING_KEY)
                    .build();
            return queue1;
        }
    
        //死信队列
        private Queue deadLetterQueue1() { 
            return new Queue(RabbitConfigConstant.DEAD_LETTER_QUEUE_1, true);
        }
    
        //死信交换机
        private DirectExchange deadLetterExchange() { 
            return new DirectExchange(RabbitConfigConstant.DEAD_LETTER_DIRECT_EXCHANGE, true, false);
        }
    }

1.  下图说明qus-queue-1队列有对应的死信交换机和死信路由键  
    ![在这里插入图片描述][20210629154430477.png]

[20210629154430477.png]: /images/20221005/1d2fd20a9a5848baac0de7ea531f8054.png