【RabbitMQ】交换机的概念及使用

一、引言

1、什么是交换机

    RabbitMQ中，交换机是一个核心概念，主要用来将生产者生产出来的消息，传送到对应的队列中。实际上，生产者生产的消息从不会直接发送到队列，而是发送到交换机。交换机一方面接收来自生产者的消息，另一方面将这些消息推入队列。

四种类型的交换机：

1. 直连交换机（Direct exchange）：根据消息携带的routing key 将消息传递给对应的队列，用来处理消息的单播路由。
2. 扇形交换机（Fanout exchange）：将消息路由给绑定到它身上的所有队列，不理会绑定的路由键，用来交换机处理消息的广播路由。
3. 主题交换机（Topic exchange）：通过对消息的路由键和队列到交换机的绑定模式之间的匹配，将消息路由给一个或多个队列，用来实现各种分发/订阅模式及其变种。
4. 默认交换机：第一个参数是交换机的名称。无名（默认交换机）在之前的学习中，我们使用的都是默认的交换机，空字符串指名的就是默认的交换机，消息发送是通过router-key 找到指定的队列 channel.basicPublish("",QUEUE_NAME,null,message.getBytes("UTF-8"));

2、Exchange

在RabbitMQ中，生产者发送消息不会直接将消息投递到队列中，而是先将消息投递到交换机中，在由交换机转发到具体的队列，队列再将消息以推送或者拉取方式给消费者进行消费

生产者将消息发送到Exchange，由Exchange再路由到一个或多个队列中

路由键(RoutingKey)
生产者将消息发送给交换机的时候，会指定RoutingKey指定路由规则。

绑定键(BindingKey)
通过绑定键将交换机与队列关联起来，这样RabbitMQ就知道如何正确地将消息路由到队列。

关系小结
生产者将消息发送给哪个Exchange是需要由RoutingKey决定的，生产者需要将Exchange与哪个队列绑定时需要由BindingKey决定的。

3、应用场景

RabbitMQ适用于各种需要异步通信、解耦、流量削峰、任务调度、日志处理等场景

1. 异步处理：RabbitMQ通过消息队列实现异步处理，允许应用程序发送消息后继续执行其他任务，而不需要等待接收者的响应。这种异步性提高了系统的吞吐量和响应速度。
2. 应用解耦：RabbitMQ用于在分布式系统中存储和转发消息，实现发送者和接收者之间的解耦。这有助于提高系统的可扩展性和可维护性。
3. 流量削峰：RabbitMQ能够应对短时间内流量激增的情况，如秒杀、抢购等场景。通过使用RabbitMQ消息队列，可以避免后端应用被大量请求压垮。
4. 任务调度：RabbitMQ可以用于任务调度，例如定时任务、批量任务等。通过将任务发送到消息队列中，可以实现任务的异步处理，同时可以通过设置消息的优先级、超时时间等属性来实现任务调度。
5. 消息通知：在分布式系统中，各个模块之间需要相互通信，例如用户注册、支付成功、物流状态等。通过使用RabbitMQ，可以将消息发送到消息队列中，由消费者接收并处理，实现消息通知功能。
6. 日志处理：RabbitMQ可以用于日志处理，例如日志收集、日志分析等。通过将日志消息发送到消息队列中，可以实现日志的异步处理，同时可以通过设置消息的属性、路由规则等来实现不同类型的日志处理。
7. 跨系统通信：RabbitMQ可用于跨系统的异步通信，所有需要异步交互的地方都可以使用消息队列。在企业应用集成（EAI）中，文件传输、共享数据库、消息队列、远程过程调用都可以作为集成的方法。
8. 应用内同步变异步：比如订单处理，就可以由前端应用将订单信息放到队列，后端应用从队列里依次获得消息处理，高峰时的大量订单可以积压在队列里慢慢处理掉。由于同步通常意味着阻塞，而大量线程的阻塞会降低计算机的性能。
9. 消息驱动的架构（EDA）：系统分解为消息队列和消息制造者和消息消费者，一个处理流程可以根据需要拆成多个阶段（Stage），阶段之间用队列连接起来，前一个阶段处理的结果放入队列，后一个阶段从队列中获取消息继续处理。
10. 需要更灵活的耦合方式：如发布订阅，比如可以指定路由规则。
11. 跨局域网甚至跨城市的通讯（CDN行业）：比如北京机房与广州机房的应用程序的通信。

4、交换机类型

Ⅰ、直连交换机:Directexchange

直连交换机的路由算法非常简单:将消息推送到bindingkey与该消息的routingkey相同的队列。

    直连交换机X上绑定了两个队列。第一个队列绑定了绑定键orange，第二个队列有两个绑定键:black和green。

    在这种场景下，一个消息在布时指定了路由键为orange将会只被路由到队列Q1，路由键为black和green的消息都将被路由到队列Q2。其他的消息都将被丢失。

    同一个绑定键可以绑定到不同的队列上去，可以增加一个交换机X与队列Q2的绑定键，在这种情况下，直连交换机将会和广播交换机有着相同的行为，将消息推送到所有匹配的队列。一个路由键为black的消息将会同时被推送到队列Q1和Q2.

Ⅱ、主题交换机:Topic exchange

①、直连交换机的缺点

直连交换机的routing_key方案非常简单，如果我们希望一条消息发送给多个队列，那么这个交换机需要绑定上非常多的routing_key.假设每个交换机上都绑定一堆的routing_key连接到各个队列上。那么消息的管理就会异常地困难。

②、主题交换机的特点

发送到主题交换机的消息不能有任意的routingkey，必须是由点号分开的一串单词，这些单词可以是任意的，但通常是与消息相关的一些特征。

比如以下是几个有效的routingkey:"stock.usd.nyse","nyse.vmw","quick,orange,rabbit"，routing key的单词可以有很多，最大限制是255 bytes。

Topic交换机的逻辑与direct交换机有点相似,使用特定路由键发送的消息将被发送到所有使用匹配绑定键绑定的队列，然而，绑定键有两个特殊的情况:

• *** **表示匹配任意一个单词
• **# **表示匹配任意一个或多个单词

routing key quick.orange.rabbit -> queue ql,Q2

routing key 1azy.orange.elephant -> queue Ql,Q2

routing key quick.orange.fox-> queue ?

routing keylazy.brown.fox-> queue ?

routing key lazy.pink.rabbit-> queue ?

routing key quick.brown.fox -> queue ?

③、延申

当一个队列的绑定键是”#"，它将会接收所有的消息，而不再考虑所接收消息的路由键。
当一个队列的绑定键没有用到“#“和“*~时，它又像direct交换一样工作。

Ⅲ、扇形交换机:Fanout exchange

扇形交换机是最基本的交换机类型，它所能做的事情非常简单广播消息。

扇形交换机会把能接收到的消息全部发送给绑定在自己身上的队列。因为广播不需要"思考”，所以扇形交换机处理消息的速度也是所有的交换机类型里面最快的。

Ⅳ、首部交换机:Headersexchange

首部交换机和扇形交换机都不需要路由键routingKey,交换机时通过Headers头部来将消息映射到队列的，有点像HTTP的Headers。

Hash结构中要求携带一个键"x-match"，这个键的Value可以是any或者al，这代表消息携带的Hash是需要全部匹配(al)，还是仅匹配一个键(any)就可以了

相比直连交换机，首部交换机的优势是匹配的规则不被限定为字符串(string)而是Object类型。

• all:在发布消息时携带的所有Entry必须和绑定在队列上的所有Entry完全匹配。
• any:只要在发布消息时携带的有一对键值对headers满足队列定义的多个参数arguments的其中一个就能匹配上，注意这里是键值对的完全匹配，只匹配到键了，值却不一样是不行的;

Ⅴ、默认交换机

实际上是一个由RabbitMQ预先声明好的名字为空字符串的直连交换机(direct exchange)。

它有一个特殊的属性使得它对于简单应用特别有用处:那就是每个新建队列(queue)都会自动绑定到默认交换机上，绑定的路由键(routing key)名称与队列名称相同。

①、绑定

默认交换

默认交换隐式绑定到每个队列，路由键等于队列名称。无法显式绑定到默认交换或取消绑定畎认交换。它也无法删除

②、Bindings(1)

当你声明了一个名为"hello”的队列，RabbitMQ会自动将其绑定到默认交换机上，绑定(binding)的路由键名称也是为"hello”。当携带着名为"helo”的路由键的消息被发送到默认交换机的时候，此消息会被默认交换机路由至名为"hello”的队列中类似amq.*的名称的交换机:这些是RabbitMQ默认创建的交换机。这些队列名称被预留做RabbitMQ内部使用，不能被应用使用，否则抛出403(ACCESS_REFUSED)错误

Ⅵ、Dead Letter Exchange(死信交换机)

    RabbitMQ中的一个重要概念，也有人称之为**死信邮箱**。当消息在一个队列中由于过期、被拒绝等原因变成死信之后，它会被重新发送到另一个交换机中，这个交换机就是死信交换机。绑定死信交换机的队列则被称为**死信队列**。

当消息在一个正常的队列中变成死信后，RabbitMQ会自动将这个消息重新发布到设置的死信交换机上，进而被路由到另一个队列，即死信队列。在RabbitMQ中，死信交换机和一般的交换机没有区别，它可以在任何队列上被指定。

使用死信队列时，只需要在定义队列的时候设置队列参数“x-dead-letter-exchange”来指定交换机即可。通过这种方式，可以实现一些特殊的应用逻辑，例如重试机制、日志审计等。

消息变成死信一般是以下三种情况:

• 消息被拒绝，并且设置 requeue 参数为 false
• 消息过期(默认情况下 Rabbit 中的消息不过期，但是可以设置队列的过期时间和消息的过期时间以达到消息过期的效果)
• 队列达到最大长度(一般当设置了最大队列长度或大小并达到最大值时)

当满足上面三种情况时，消息会变成死信消息,并通过死信交换机投递到相应的队列中。

我们只需要监听相应队列,就可以对死信消息进行最后的处理。

5、交换机的属性

Name:交换机名称

Type:交换机类型，direct、topic、fanout、headers

Durabi1ity:是否需要持久化,如果持久性，则RabbitMQ重启后，交换机还存在

Auto Delete:当最后一个绑定到Exchange上的队列删除后，自动删除该

ExchangeInterna1:当前Exchange是否用于RabbitMQ内部使用，默认为False

Arguments:扩展参数，用于扩展AMQP协议定制化使用

二、实操案例

1、准备

消费者

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
 
@Component
@SuppressWarnings("all")
@Slf4j
@RabbitListener(queues = "Queue01")
public class ReceiverQ1 {
 
        // 接收directExchange01交换机中Queue01队列消息的方法
        @RabbitHandler
        public void Queue01(String msg) {
            log.warn("Queue01，接收到信息:" + msg);
        }
 
}

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
 
@Component
@SuppressWarnings("all")
@Slf4j
@RabbitListener(queues = "Queue02")
public class ReceiverQ2 {
 
        // 接收directExchange01交换机中Queue02队列消息的方法
        @RabbitHandler
        public void Queue02(String msg) {
            log.warn("Queue02，接收到信息:" + msg);
        }
 
}

2、直连

RabbitConfig

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;

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

    //    直连交换机

    /**
     * 直连交换机
     * /
     * 创建两个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");
    }

}

在Controller层编写

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

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

测试结果

消费者接收的

3、主题

RabbitConfig

 /**
     * 主题交换机
     * /
     * 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.#");
    }

Controller层

   @RequestMapping("test05")
    public String test05(String rex) {
        // 向topicExchange发送消息，路由键为rex，消息内容为"Hello,topicExchange:Queue!"
        rabbitTemplate.convertAndSend("topicExchange", rex, "Hello,topicExchange:Queue!");
        return "🐉";
    }

4、扇形

RabbitConfig

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

    // 创建一个绑定，将Queue01绑定到fanoutExchange
    @Bean
    public Binding binding07() {
        return BindingBuilder
                .bind(Queue01())
                .to(fanoutExchange());
    }
    // 创建一个绑定，将Queue02绑定到fanoutExchange
    @Bean
    public Binding binding08() {
        return BindingBuilder
                .bind(Queue02())
                .to(fanoutExchange());
    }

Controller层

    @RequestMapping("test06")
   public String test06() {
        // 向fanoutExchange发送消息，消息内容为"Hello,fanoutExchange:Queue!"
        rabbitTemplate.convertAndSend("fanoutExchange","","Hello,fanoutExchange:Queue!");
        return "👌🐉";
    }

测试结果

三、总结

直连交换机适用于简单的单播路由场景。

主题交换机适用于实现各种分发/订阅模式及其变种。

扇形交换机则适用于需要快速广播消息的场景。

直连交换机：

• 特点：直连交换机的routing_key方案非常简单。如果我们希望一条消息发送多个队列，那么这个交换机需要绑定上非常多的routing_key。假设每个交换机上都绑定一堆的routing_key连接到各个队列上，那么消息管理就会异常困难。
• 应用场景：适用于简单的消息路由场景，如单播路由。

主题交换机：

• 特点：主题交换机根据消息的routing key和队列到交换机的绑定模式之间的匹配，将消息路由给一个或多个队列。这提供了多种分发/订阅模式及其变种。
• 应用场景：适用于实现各种分发/订阅模式及其变种，例如多播路由、广播路由等。

扇形交换机：

• 特点：扇形交换机是最基本的交换机类型，其功能非常简单——广播消息。它会把能接收到的消息全部发送给绑定在自己身上的队列，因为广播不需要“思考”，所以扇形交换机处理消息的速度也是所有的交换机类型里面最快的。此外，扇形交换机没有路由键概念，即使绑定了路由键也会被无视。
• 应用场景：适用于需要快速广播消息的场景，例如日志处理、事件通知等。