RabbitMQ中交换机的应用，原理，案例的实现

🎉🎉欢迎来到我的CSDN主页！🎉🎉

🏅我是平顶山大师，一个在CSDN分享笔记的博主。📚📚

🌟推荐给大家我的博客专栏《RabbitMQ中交换机的应用及原理，案例的实现》。🎯🎯

🎁如果感觉还不错的话请给我关注加三连吧！🎁🎁

一、RabbitMQ中交换机的介绍

1.基本概念

1.1. 概述

1.2.RabbitMQ交换机类型

2. 作用及优势

2.1 作用

2.2 优势

3. 工作原理

二、交换机Exchange

1. Direct

2. Topic

3. Fanout

三、代码案例

1. 直连direct

生产者代码

消费者代码：

直连测试：

2.Topic

生产者代码

测试

3. 扇形fanout

生产者代码

测试

总结：

一、RabbitMQ中交换机的介绍

1.基本概念

1.1. 概述

RabbitMQ中的交换机（exchange）是消息的分发中心，它接收来自生产者的消息，并将这些消息路由到一个或多个队列中。交换机根据消息的路由键（routing key）将消息发送到相应的队列中。

1.2.RabbitMQ交换机类型

RabbitMQ一共四种交换机，如下所示：

直连交换机（direct exchange）：直连交换机根据消息的路由键将消息发送到与之匹配的队列中。如果消息的路由键与队列的绑定键（binding key）完全匹配，那么消息将被发送到该队列中。

主题交换机（topic exchange）：主题交换机根据消息的路由键和队列的绑定键的模式进行匹配。可以使用通配符（*和#）来匹配多个路由键，从而实现更灵活的消息路由。符号#表示一个或多个词，*表示一个词。

扇形交换机（fanout exchange）：扇出交换机将消息发送到所有与之绑定的队列中，无论消息的路由键是什么。它实现了一对多的消息分发。

头部交换机（headers exchange）：头部交换机根据消息的头部信息进行匹配，而不是路由键。可以根据消息的头部属性来决定消息的路由。

交换机和队列之间通过绑定（binding）进行关联，生产者将消息发送到交换机，交换机根据路由键将消息发送到相应的队列中。交换机和队列的绑定关系可以通过管理界面或者命令行工具进行配置。交换机它负责将消息路由到相应的队列中，实现了灵活的消息分发机制。不同类型的交换机可以满足不同的业务需求，开发者可以根据实际情况选择合适的交换机类型来实现消息的路由和分发。

2. 作用及优势

2.1 作用

交换机在项目中的主要作用包括：

1. 消息路由：交换机负责将消息路由到一个或多个队列中，根据消息的路由键和交换机的类型进行匹配和分发，确保消息能够准确地到达目标队列。

2. 消息分发：交换机可以根据不同的规则将消息分发到不同的队列中，实现灵活的消息分发机制。这对于实现消息的多播、广播等场景非常有用。

3. 解耦：通过交换机，生产者和消费者之间可以完全解耦，生产者只需要将消息发送到交换机中，而不需要关心消息具体发送到哪个队列中，消费者也只需要从队列中接收消息，而不需要关心消息的来源。

4. 消息过滤：通过不同类型的交换机和绑定规则，可以实现消息的过滤和选择性接收，确保消费者只接收到其关心的消息。

5. 实现消息通道：交换机是消息在RabbitMQ中的通道，通过交换机可以将消息从生产者传递给消费者，实现了消息的传递和通信。

交换机在项目中起到了消息路由、分发、解耦和过滤等重要作用，是实现消息传递和通信的关键组件。通过合理使用不同类型的交换机，可以实现灵活、高效的消息传递机制，满足不同业务场景的需求。

2.2 优势

交换机在消息传递系统中具有以下优势：

1. 灵活的消息路由：交换机可以根据消息的路由键将消息发送到不同的队列中，实现了灵活的消息路由机制。这样可以根据消息的不同属性将消息发送到不同的消费者或处理逻辑中，提高了系统的灵活性和可扩展性。

2. 解耦和分布式系统支持：通过交换机，生产者和消费者之间可以完全解耦，生产者只需要将消息发送到交换机中，而不需要关心消息具体发送到哪个队列中，消费者也只需要从队列中接收消息，而不需要关心消息的来源。这对于构建分布式系统和微服务架构非常有用。

3. 多播和广播支持：通过扇出交换机（fanout exchange），交换机可以将消息发送到所有与之绑定的队列中，实现了一对多的消息分发，支持了多播和广播的消息传递方式。

4. 消息过滤和选择性接收：通过不同类型的交换机和绑定规则，可以实现消息的过滤和选择性接收，确保消费者只接收到其关心的消息，提高了系统的效率和性能。

5. 实现消息通道：交换机是消息在消息队列系统中的通道，通过交换机可以将消息从生产者传递给消费者，实现了消息的传递和通信，为系统中的消息传递提供了可靠的通道。

总的来说，交换机在消息传递系统中具有灵活的消息路由、解耦和分布式系统支持、多播和广播支持、消息过滤和选择性接收等优势，为构建高效、灵活的消息传递系统提供了重要的支持。

3. 工作原理

RabbitMQ的交换机（Exchange）是消息路由的核心组件，负责消息的分发和路由。下面是RabbitMQ交换机的工作原理：

1. 发布消息：生产者将消息发送到RabbitMQ的交换机中，同时指定一个路由键（Routing Key）。

2. 交换机根据类型进行路由：RabbitMQ的交换机有四种类型，分别是直连交换机（direct exchange）、扇出交换机（fanout exchange）、主题交换机（topic exchange）和头部交换机（headers exchange）。不同类型的交换机根据不同的路由规则进行消息的路由和分发。

3. 路由规则：直连交换机根据消息的路由键将消息发送到与之绑定的队列中；扇出交换机将消息发送到所有与之绑定的队列中；主题交换机根据消息的路由键和队列的绑定规则进行匹配，将消息发送到匹配的队列中；头部交换机根据消息的头部属性进行匹配，将消息发送到匹配的队列中。

4. 绑定队列：交换机需要和队列进行绑定，指定绑定的路由键或者其他条件，确保消息能够被正确地路由到目标队列中。

5. 发送到队列：一旦消息被交换机路由到目标队列，消费者就可以从队列中接收并处理消息。

总的来说，RabbitMQ的交换机根据不同的类型和路由规则，将消息发送到目标队列中，实现了消息的路由和分发。通过合理使用不同类型的交换机和绑定规则，可以实现灵活、高效的消息传递机制，满足不同业务场景的需求。

二、交换机Exchange

1. Direct

直连交换机（Direct Exchange）是RabbitMQ中最简单的交换机类型之一，它使用消息的路由键（Routing Key）来决定将消息发送到哪个队列。

2. Topic

主题交换机（Topic Exchange）是RabbitMQ中一种灵活且强大的交换机类型，它使用消息的路由键和通配符模式来决定将消息发送到哪个队列。Queue 通过 routingkey 绑定到TopicExchange 上，当消息到达 TopicExchange 后，TopicExchange 根据消息的 routingkey 将消息路由到一个或者多个 Queue 上。

3. Fanout

扇形交换机是RabbitMQ中的一种消息路由方式，FanoutExchange 的数据交换策略是把所有到达 FanoutExchange 的消息转发给所有与它绑定的 Queue 上，在这种策略中，routingkey 将不起任何作用，不管消息的路由键是什么。

进入RabbitMQ的管理页面：

这里会展示交换机的各种信息：

Type 表示交换机的类型。

Features 有两个取值 D 和 I。

D 表示交换机持久化，将交换机的属性在服务器内部保存，当 MQ 的服务器发生意外或关闭之后，重启 RabbitMQ 时不需要重新手动或执行代码去建立交换机，交换机会自动建立，相当于一直存在。

I 表示这个交换机不可以被消息生产者用来推送消息，仅用来进行交换机和交换机之间的绑定。

Message rate in 表示消息进入的速率。 Message rate out 表示消息出去的速率。

点击下方的 Add a new exchange 可以创建一个新的交换机。

三、代码案例

如果你掌握了上一篇博客的基础上，再来写这一次的案例，就是soeasy

3.1开启docker服务

1. 直连direct

生产者代码

在生产者项目中的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");
    }

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

//    -------------------------直连------------------------------------

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

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

消费者代码：

ReceiverQ1接收交换机中Queue01队列消息的方法：

package com.example.consumer;
 
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);
        }
 
}

ReceiverQ2接收交换机中Queue02队列消息的方法

package com.example.consumer;
 
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);
        }
 
}

这里在备注一下；生产者根据key来找对应的消息队列

直连测试：

2.Topic

生产者代码

RabbitConfig中增加以下代码：

 /**
     * 主题交换机
     *
     *  '*'代表一个单词，
     *  '#'代表任意数量的字符，也代表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("平顶山大师.#");
    }
    @Bean
    public Binding binding06() {
        return BindingBuilder
                .bind(Queue02())
                .to(topicExchange())
                .with("平顶山大师.#");
    }

TestController中增加以下代码：

 //    -------------------------主题topic ------------------------------------

    @RequestMapping("sender5")
    @ResponseBody
    public String test05(String rex) {
        rabbitTemplate.convertAndSend("topicExchange",rex,"Hello,topicExchange:Queue!");
        return "🎈5";
    }

    @RequestMapping("sender6")
    @ResponseBody
    public String test06() {
        rabbitTemplate.convertAndSend("fanoutExchange","","Hello,fanoutExchange:Queue!");
        return "💕6";
    }

测试

3. 扇形fanout

生产者代码

RabbitConfig中增加以下代码：

    /**
     * 扇形交换机
     *
     * 定义了一个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());
    }

TestController中增加以下代码：

    @RequestMapping("sender6")
    @ResponseBody
    public String test06() {
        rabbitTemplate.convertAndSend("fanoutExchange","","Hello,fanoutExchange:Queue!");
        return "💕6";
    }

这里会直接发送到Q1,Q2

测试

全部代码：

RabbitConfig：

package com.example.provide;

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

@Configuration
@SuppressWarnings("all")
public class RabbitConfig {
    @Bean
    public Queue firstQueue() {
        return new Queue("firstQueue");
    }

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

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

    /**
     * 主题交换机
     *
     *  '*'代表一个单词，
     *  '#'代表任意数量的字符，也代表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("平顶山大师.#");
    }
    @Bean
    public Binding binding06() {
        return BindingBuilder
                .bind(Queue02())
                .to(topicExchange())
                .with("平顶山大师.#");
    }

    /**
     * 扇形交换机
     *
     * 定义了一个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());
    }
}

TestController：

package com.example.provide;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

/**
 * @author 程序猿-小李哥
 * @site www.xiaolige.com
 * @company 猪八戒有限集团
 * @create 2024-01-19-18:46
 */
@Controller
public class Testcontrol {
    @Autowired
    private AmqpTemplate rabbitTemplate;
    @Autowired
    private ObjectMapper objectMapper;

    @RequestMapping("/sender1")
    @ResponseBody
    public String sendFirst() {
        rabbitTemplate.convertAndSend("firstQueue", "Hello World");
        return "😶‍🌫️1";
    }

    @RequestMapping("/sender2")
    @ResponseBody
    public String sender2() throws JsonProcessingException {
        User user = new User("1", "1");
        // 序列化对象转换为JSON字符串
        String json = objectMapper.writeValueAsString(user);
        rabbitTemplate.convertAndSend("secondQueue", json);
        return "😎2";
    }

//    -------------------------直连------------------------------------

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

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

    //    -------------------------主题topic ------------------------------------

    @RequestMapping("sender5")
    @ResponseBody
    public String test05(String rex) {
        rabbitTemplate.convertAndSend("topicExchange",rex,"Hello,topicExchange:Queue!");
        return "🎈5";
    }

    @RequestMapping("sender6")
    @ResponseBody
    public String test06() {
        rabbitTemplate.convertAndSend("fanoutExchange","","Hello,fanoutExchange:Queue!");
        return "💕6";
    }

}

总结：

问题：为啥要使用RabbitMQ中交换机？？？

消息路由：交换机用于将消息从生产者发送到消费者之间建立起路由规则。它可以根据不同的路由键（routing key）将消息传递给不同的队列或者其他交换机，实现灵活的消息传递方式。

解耦和扩展性：通过使用交换机，生产者和消费者之间可以解耦。生产者只需要将消息发送到交换机上，并不需要关心具体的消费者。消费者可以根据自己的需求，绑定到交换机上接收特定类型的消息，这样可以实现系统的灵活扩展和升级。

发布-订阅模式：交换机支持发布-订阅模式，可以将消息广播给多个消费者。当多个消费者同时绑定到同一个交换机上时，交换机会将消息复制并传递给所有绑定的队列，实现消息的多播。