RabbitMQ中死信交换机的应用，工作原理，实现案例

目录

一、介绍

1. 概述

2. 应用场景

3. 工作原理

二、应用

1. 讲述

2. 运用

三、案例

1. 实践 

2. 代码整合

每篇一获

一、介绍

1. 概述

死信交换机是用来处理消息队列中无法被消费者正确处理的消息的交换机。当消息在队列中变成死信时，它会被重新发送到死信交换机，然后被路由到死信队列中进行处理。

死信交换机的作用是将死信消息重新路由到指定的死信队列中，以便进行后续处理。这样可以帮助系统更好地处理无法被消费者正确处理的消息，保证消息队列的稳定运行。

在RabbitMQ中，可以通过设置队列的属性来指定死信交换机和死信队列。当消息变成死信时，会根据队列的属性将消息发送到对应的死信交换机，然后再路由到指定的死信队列中。

• 当消息无法路由到队列时，确认消息路由失败。消息成功路由时，当需要发送的队列都发送成功后，进行确认消息，对于持久化队列意味着写入磁盘，对于镜像队列意味着所有镜像接收成功

• 通过实现 ConfirmCallback 接口，消息发送到 Broker 后触发回调，确认消息是否到达 Broker 服务器，也就是只确认是否正确到达 Exchange 中

通过使用死信交换机，可以更好地处理消息队列中的异常情况，保证系统的稳定性和可靠性。因此，死信交换机在消息队列中起着非常重要的作用。

2. 应用场景

死信交换机在项目中有多种应用场景，其中一些包括：        

        1. 错误处理：当消息在队列中无法被正确处理时，可以将其发送到死信交换机，然后再路由到死信队列中进行错误处理和调查。

        2. 重试机制：当消息处理失败时，可以将其发送到死信交换机，并设置重试次数。如果消息在一定次数内仍然无法被处理，可以将其发送到死信队列中，以便进一步处理或记录错误信息。

        3. 延迟消息处理：可以通过设置死信交换机和死信队列来实现延迟消息处理。当消息需要延迟处理时，可以将其发送到死信交换机，并设置一定的延迟时间，然后再路由到指定的死信队列中进行处理。

        4. 日志记录：可以将无法被正确处理的消息发送到死信交换机，并将其路由到死信队列中进行日志记录，以便后续分析和排查问题。

        总之，死信交换机在项目中可以用于处理消息队列中的异常情况，提高系统的稳定性和可靠性，同时也可以用于实现一些特定的消息处理需求，如延迟处理、重试机制等。因此，在消息队列的设计和应用中，死信交换机是一个非常有用的工具。

3. 工作原理

死信交换机（Dead Letter Exchange，DLX）的工作原理如下：

        1. 定义死信队列：首先需要在消息队列中定义一个死信队列，用于存放无法被正确处理的消息。

        2. 设置队列属性：在定义普通队列时，可以设置一些属性，如消息的过期时间、消息的最大重试次数等。当消息满足这些属性时，会变成死信消息。

        3. 绑定死信交换机：在定义普通队列时，可以指定该队列的死信交换机和死信路由键。当消息变成死信时，会被发送到死信交换机，并根据指定的路由键路由到死信队列中。

        4. 处理死信消息：一旦消息被发送到死信队列中，就可以进行后续处理，如记录日志、重试处理、延迟处理等。

具体工作流程如下：

• 1.  当消息在普通队列中满足某些条件（如消息过期、消息被拒绝、队列长度超过限制等）时，会变成死信消息。
• 2.  死信消息会被发送到指定的死信交换机中。
• 3.  死信交换机会根据指定的路由键将死信消息路由到死信队列中。
• 4.  死信队列中的消息可以被消费者重新处理，或者进行其他后续处理。

通过死信交换机，可以更好地处理消息队列中的异常情况，保证系统的稳定性和可靠性。同时，也可以实现一些特定的消息处理需求，如延迟处理、重试机制等。因此，死信交换机在消息队列中具有非常重要的作用。

二、应用

1. 讲述

死信，在官网中对应的单词为“Dead Letter”,它是 RabbitMQ 的一种消息机制。

般来说，生产者将消息投递到 broker 或者直接到 queue 里了，consumer 从 queue 取出消息进行消费，如果它一直无法消费某条数据，那么可以把这条消息放入死信队列里面。等待

条件满足了再从死信队列中取出来再次消费，从而避免消息丢失。

生产者代码： 

在生产者的RabbitConfig中增加以下代码：

java">  @Bean
    public Queue queueA() {
        return new Queue("queueA");
    }

    @Bean
    public DirectExchange directExchangeA() {
        return new DirectExchange("directExchangeA");
    }

    @Bean
    public Binding bindingA() {
        return BindingBuilder
                .bind(queueA())
                .to(directExchangeA())
                .with("dA");
    }


    @Bean
    public Queue queueB() {
        return new Queue("queueB");
    }

    @Bean
    public DirectExchange directExchangeB() {
        return new DirectExchange("directExchangeB");
    }

    @Bean
    public Binding bindingB() {
        return BindingBuilder
                .bind(queueB())
                .to(directExchangeB())
                .with("dB");
    }

解析：

以上代码定义了一个名为"RabbitMQ死信队列处理"的配置类，该类包含三个队列：queueA、queueB和directExchangeA、directExchangeB。

其中，queueA和directExchangeA进行绑定，并指定路由键为dA；queueB和directExchangeB进行绑定，并指定路由键为dB。

在生产者的TestController中增加以下代码：

  @RequestMapping("test07")
    public String test07() {
        template.convertAndSend("directExchangeA","dA","Hello,directExchangeA:QueueA!");
        return "🐲🐾";
    }

消费者代码：

在消费者中创建一个ReceiverQB（消息接收者类）它监听名为"queueB"的队列。当收到消息时，它会打印出接收到的消息。

ReceiverQB：

package ***.cloudjun.consumer;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.***ponent;

@***ponent
@SuppressWarnings("all")
@Slf4j
@RabbitListener(queues = "queueB")
public class ReceiverQB {

        @RabbitHandler
        public void queueB(String msg) {
            log.warn("queueB，接收到信息:" + msg);
        }

}

在开启项目前记得在虚拟机中开启docker服务

systemctl start docker

并且运行容器：

docker start  my-rabbitmq

如果没有就以下命令创建：

docker run -d \
--name my-rabbitmq \
-p 5672:5672 -p 15672:15672 \
--hostname my-rabbitmq-host \
-e RABBITMQ_DEFAULT_VHOST=my_vhost \
-e RABBITMQ_DEFAULT_USER=admin \
-e RABBITMQ_DEFAULT_PASS=admin \
--restart=always \
rabbitmq:management 

运行项目，在浏览器中服务对应的路径：

 根据我们设定的时间，十秒钟后就可以在消费者中你到死信队列中的信息：

三、案例

• 消息通过 ACK 确认是否被正确接收，每个 Message 都要被确认（acknowledged），可以手动去 ACK 或自动 ACK

• 自动确认会在消息发送给消费者后立即确认，但存在丢失消息的可能，如果消费端消费逻辑抛出异常，也就是消费端没有处理成功这条消息，那么就相当于丢失了消息

• 如果消息已经被处理，但后续代码抛出异常，使用 Spring 进行管理的话消费端业务逻辑会进行回滚，这也同样造成了实际意义的消息丢失

• 如果手动确认则当消费者调用 ack、nack、reject 几种方法进行确认，手动确认可以在业务失败后进行一些操作，如果消息未被 ACK 则会发送到下一个消费者

• 如果某个服务忘记 ACK 了，则 RabbitMQ 不会再发送数据给它，因为 RabbitMQ 认为该服务的处理能力有限

• ACK 机制还可以起到限流作用，比如在接收到某条消息时休眠几秒钟

• 消息确认模式有：

  • AcknowledgeMode.NONE：自动确认

  • AcknowledgeMode.AUTO：根据情况确认

  • AcknowledgeMode.MANUAL：手动确认

默认情况下消息消费者是自动 ack （确认）消息的，如果要手动 ack（确认）则需要修改确认模式为 manual

1. 实践 

在我们的生产者中，配置yml文件，配置未手动的

如：

listener:
      simple:
        acknowledge-mode: manual

代码：

server:
    port: 8848
spring:
    rabbitmq:
        host: 192.168.211.129
        username: Jun
        password: 123456
        port: 5672
        virtual-host: my_vhost
        listener:
            simple:
                acknowledge-mode: manual

在消费者中创建ReceiverQA：

package ***.cloudjun.consumer;

import ***.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.***ponent;

import java.io.IOException;

@***ponent
@SuppressWarnings("all")
@Slf4j
@RabbitListener(queues = "queueA")
public class ReceiverQA {

        @RabbitHandler
        public void queueB(String msg){
            log.warn("queueA，接收到信息:" + msg);
        }

}

消息接收者类，它监听名为"queueA"的队列。当收到消息时，它会打印出接收到的消息。注意，这里使用了 @Header 注解来获取消息头部的信息。

在生产者中RabbitConfig的queueA（）方法进行修改：

    @Bean
    public Queue queueA() {
        Map<String, Object> config = new HashMap<>();
        //message在该队列queue的存活时间最大为10秒
        config.put("x-message-ttl", 10000);
        //x-dead-letter-exchange参数是设置该队列的死信交换器（DLX）
        config.put("x-dead-letter-exchange", "directExchangeB");
        //x-dead-letter-routing-key参数是给这个DLX指定路由键
        config.put("x-dead-letter-routing-key", "dB");
        return new Queue("queueA",true,false, false, config);
    }

queueA设置了死信交换器（DLX）为directExchangeB，并设置了死信路由键为dB。当消息的生命周期结束时，该消息会被发送到directExchangeB，并指定路由键为dB。

重启项目，访问浏览器：

在消费者的ReceiverQA类进行修改：

package ***.cloudjun.consumer;

import ***.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.***ponent;

import java.io.IOException;

@***ponent
@SuppressWarnings("all")
@Slf4j
@RabbitListener(queues = "queueA")
public class ReceiverQA {

        @RabbitHandler
        public void queueB(String msg, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception {
            //如果重新入队，那么queueA会再次收到消息
            //如果直接拒绝，那么queueA不会再次收到消息
            //channel.basicAck(tag,true);
            //log.warn("queueA，接收到信息:" + msg);
            //拒绝消息，true表示会重新入队，false表示不重新入队就会立马进死信队列queueB
            log.error("queueA，接收到信息:" + msg);
            channel.basicReject(tag,true);
            Thread.sleep(1000);//模拟处理时间，每1秒
        }

}

最后再访问浏览器：

2. 代码整合

消费者的RabbitConfig：

package ***.cloudjun.publisher;

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

@Configuration
@SuppressWarnings("all")
public class RabbitConfig {

    // 创建队列
    @Bean
    public Queue messageQueue() {
        return new Queue("messageQueue");
    }

    @Bean
    public Queue messageUser() {
        return new Queue("messageUser");
    }

    /**
     * 直连交换机
     * /
     * 创建两个Binding Bean，分别与Queue01和Queue02队列进行绑定
     * 并都指向directExchange01（直连交换机），键分别为Key01和Key02
     */
    //  创建队列
    @Bean
    public Queue Queue01() {
        return new Queue("Queue01");
    }
    @Bean
    public Queue Queue02() {
        return new Queue("Queue02");
    }

    // 创建直连（direct）交换机
    @Bean
    public DirectExchange directExchange01() {
        return new DirectExchange("directExchange01");
    }

    // 创建Binding Bean，与Queue01和directExchange01绑定，键为Key01
    @Bean
    public Binding binding01() {
        return BindingBuilder
                .bind(Queue01())
                .to(directExchange01())
                .with("Key01");
    }

    // 创建Binding Bean，与Queue02和directExchange01绑定，键为Key02
    @Bean
    public Binding binding02() {
        return BindingBuilder
                .bind(Queue02())
                .to(directExchange01())
                .with("Key02");
    }

    /**
     * 主题交换机
     * /
     * binding03：将Queue01绑定到topicExchange，并使用*.*.Q1作为路由键。
     * binding04：将Queue02绑定到topicExchange，并使用*.*.Q2作为路由键。
     * binding05：将Queue01绑定到topicExchange，并使用un.#作为路由键。
     * binding06：将Queue02绑定到topicExchange，并使用un.#作为路由键。
     *  '*'代表一个单词，
     *  '#'代表任意数量的字符，也代表0个或多个
     */
    // 创建主题交换机
    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange("topicExchange");
    }
    @Bean
    public Binding binding03() {
        return BindingBuilder
                .bind(Queue01())
                .to(topicExchange())
                .with("*.*.Q1");
    }
    @Bean
    public Binding binding04() {
        return BindingBuilder
                .bind(Queue02())
                .to(topicExchange())
                .with("*.*.Q2");
    }
    @Bean
    public Binding binding05() {
        return BindingBuilder
                .bind(Queue01())
                .to(topicExchange())
                .with("un.#");
    }
    @Bean
    public Binding binding06() {
        return BindingBuilder
                .bind(Queue02())
                .to(topicExchange())
                .with("un.#");
    }

    /**
     * 扇形交换机
     *
     * 定义了一个FanoutExchange，加上Bean注解
     * 定义了两个Binding，加上Bean注解
     * 将两个队列绑定到FanoutExchange上，从而实现广播消息的功能
     * 扇形交换机会将接收到的消息路由到所有绑定到它上的队列。
     */
    // 创建扇形交换机
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("fanoutExchange");
    }

    @Bean
    public Binding binding07() {
        return BindingBuilder
                .bind(Queue01())
                .to(fanoutExchange());
    }
    @Bean
    public Binding binding08() {
        return BindingBuilder
                .bind(Queue02())
                .to(fanoutExchange());
    }





    @Bean
    public Queue queueA() {
        Map<String, Object> config = new HashMap<>();
        //message在该队列queue的存活时间最大为10秒
        config.put("x-message-ttl", 10000);
        //x-dead-letter-exchange参数是设置该队列的死信交换器（DLX）
        config.put("x-dead-letter-exchange", "directExchangeB");
        //x-dead-letter-routing-key参数是给这个DLX指定路由键
        config.put("x-dead-letter-routing-key", "dB");
        return new Queue("queueA",true,false, false, config);
    }

    @Bean
    public DirectExchange directExchangeA() {
        return new DirectExchange("directExchangeA");
    }

    @Bean
    public Binding bindingA() {
        return BindingBuilder
                .bind(queueA())
                .to(directExchangeA())
                .with("dA");
    }


    @Bean
    public Queue queueB() {
        return new Queue("queueB");
    }

    @Bean
    public DirectExchange directExchangeB() {
        return new DirectExchange("directExchangeB");
    }

    @Bean
    public Binding bindingB() {
        return BindingBuilder
                .bind(queueB())
                .to(directExchangeB())
                .with("dB");
    }







}

消费者的TestController：

package ***.cloudjun.publisher;

import ***.fasterxml.jackson.core.JsonProcessingException;
import ***.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author CloudJun
 */
@RestController
public class TestController {

    @Autowired
    private AmqpTemplate template;
    @Autowired
    private ObjectMapper objectMapper;

    @RequestMapping("test01")
    public String test01(){
        // 发送消息到名为messageQueue的队列
        // 这里的messageQueue是RabbitMQ中定义的队列名称
        // 这里的"Hello World!"是发送的消息内容
        template.convertAndSend("messageQueue", "HelloWorld!");
        return "💖";
    }

    @RequestMapping("test02")
    public String test02() throws Exception {
        // 发送消息到名为messageQueue的队列
        // 这里的messageQueue是RabbitMQ中定义的队列名称
        User user = new User("Jun", "123456");
        // 序列化对象转换为JSON字符串
        String json = objectMapper.writeValueAsString(user);
        template.convertAndSend("messageUser", json);
        return "💖";
    }


    @RequestMapping("test03")
    public String test03() {
        // 发送消息到名为directExchange01的交换机，路由键为key01，信息内容为：Hello, direct exchange!
        // 这里的directExchange01是RabbitMQ中定义的交换机名称
        // 这里的key01是RabbitMQ中定义的路由键名称
        template.convertAndSend("directExchange01","Key01", "Hello, direct exchange!");
        return "🙊👌";
    }

    @RequestMapping("test04")
    public String test04() {
        // 发送消息到名为directExchange01的交换机，路由键为key02，信息内容为：Hello, direct exchange!
        // 这里的directExchange01是RabbitMQ中定义的交换机名称
        // 这里的key02是RabbitMQ中定义的路由键名称
        template.convertAndSend("directExchange01","Key02", "Hello, direct exchange!");
        return "🙊👌";
    }


    @RequestMapping("test05")
    public String test05(String rex) {
        template.convertAndSend("topicExchange",rex,"Hello,topicExchange:Queue!");
        return "🙊👌";
    }

    @RequestMapping("test06")
    public String test06() {
        template.convertAndSend("fanoutExchange","","Hello,fanoutExchange:Queue!");
        return "🙊👌";
    }


    @RequestMapping("test07")
    public String test07() {
        template.convertAndSend("directExchangeA","dA","Hello,directExchangeA:QueueA!");
        return "🐲🐾";
    }



}

每篇一获

学习了死信交换机的技术点后，对项目开发可能会有以下收获：

        1. 异常处理和重试机制：通过死信交换机，可以更好地处理消息队列中的异常情况，实现重试机制，提高系统的可靠性和稳定性。

        2. 延迟消息处理：可以利用死信交换机实现延迟消息处理，满足一些特定的业务需求，如定时任务、延迟通知等。

        3. 日志记录和监控：可以将无法被正确处理的消息发送到死信队列中，以便进行日志记录和监控，帮助排查问题和分析系统运行情况。

        4. 架构设计优化：在项目架构设计中考虑死信交换机，可以更好地处理消息队列中的异常情况，提高系统的健壮性和可维护性。

        5. 业务流程优化：对于一些需要延迟处理或重试处理的业务流程，可以利用死信交换机来优化实现，提高系统的灵活性和可扩展性。

总之，学习了死信交换机的技术点后，可以在项目开发中更好地处理消息队列中的异常情况，提高系统的稳定性和可靠性，同时也可以实现一些特定的消息处理需求，如延迟处理、重试机制等。因此，死信交换机的应用可以为项目开发带来诸多收益。