RabbitMq交换机详解

2023-12-17 16:47:25

1.交换机类型

在之前的两个测试案例中直发消息队列，都没有交换机，生产者直接发送消息到队列。而一旦引入交换机，消息发送的模式会有很大变化：

可以看到，在订阅模型中，多了一个exchange角色，而且过程略有变化：

Publisher：生产者，不再发送消息到队列中，而是发给交换机
Exchange：交换机，一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。
Queue：消息队列也与以前一样，接收消息、缓存消息。不过队列一定要与交换机绑定。
Consumer：消费者，与以前一样，订阅队列，没有变化

Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！

交换机的类型有四种：

Fanout：广播，将消息交给所有绑定到交换机的队列。我们最早在控制台使用的正是Fanout交换机
Direct：订阅，基于RoutingKey（路由key）发送给订阅了消息的队列
Topic：通配符订阅，与Direct类似，只不过RoutingKey可以使用通配符
Headers：头匹配，基于MQ的消息头匹配，用的较少。

2.Fanout交换机

Fanout，英文翻译是扇出，我觉得在MQ中叫广播更合适。
在广播模式下，消息发送流程是这样的：
在这里插入图片描述

1）可以有多个队列
2）每个队列都要绑定到Exchange（交换机）
3）生产者发送的消息，只能发送到交换机
4）交换机把消息发送给绑定过的所有队列
5）订阅队列的消费者都能拿到消息

我们的计划是这样的：
在这里插入图片描述

创建一个名为 shen.fanout的交换机，类型是Fanout
创建两个队列fanout.queue1和fanout.queue2，绑定到交换机shen.fanout

2.1.声明队列和交换机

在控制台创建队列fanout.queue1:
在这里插入图片描述

在创建一个队列fanout.queue2：
在这里插入图片描述

然后再创建一个交换机：
在这里插入图片描述

然后绑定两个队列到交换机：
在这里插入图片描述

在这里插入图片描述

2.2.消息发送

在publisher服务的SpringAmqpTest类中添加测试方法：

@Test
void testFanoutQueue() throws InterruptedException {
	String exchangeName = "shen.fanout";
	for (int i = 0; i < 50; i++) {
		String msg = "hello,everyone,message_" + i;
		rabbitTemplate.convertAndSend(exchangeName, null,msg);
		Thread.sleep(20);
	}
}

2.3.消息接收

在consumer服务的SpringRabbitListener中添加两个方法，作为消费者：

@RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(String msg) {
    System.out.println("消费者1接收到Fanout消息：【" + msg + "】");
}

@RabbitListener(queues = "fanout.queue2")
public void listenFanoutQueue2(String msg) {
    System.out.println("消费者2接收到Fanout消息：【" + msg + "】");
}

2.4.总结

交换机的作用是什么？

接收publisher发送的消息
将消息按照规则路由到与之绑定的队列
不能缓存消息，路由失败，消息丢失
FanoutExchange的会将消息路由到每个绑定的队列

3.Direct交换机

在Fanout模式中，一条消息，会被所有订阅的队列都消费。但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。
在这里插入图片描述

在Direct模型下：

队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）
消息的发送方在向 Exchange发送消息时，也必须指定消息的 RoutingKey。
Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey与消息的 Routing key完全一致，才会接收到消息

案例需求如图：
在这里插入图片描述

声明一个名为shen.direct的交换机
声明队列direct.queue1，绑定shen.direct，bindingKey为blud和red
声明队列direct.queue2，绑定shen.direct，bindingKey为yellow和red
在consumer服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2
在publisher中编写测试方法，向shen.direct发送消息

3.1.声明队列和交换机

首先在控制台声明两个队列direct.queue1和direct.queue2，这里不再展示过程：
在这里插入图片描述

然后声明一个direct类型的交换机，命名为shen.direct:
在这里插入图片描述

然后使用red和blue作为key，绑定direct.queue1到shen.direct：
在这里插入图片描述

同理，使用red和yellow作为key，绑定direct.queue2到shen.direct，步骤略，最终结果：
在这里插入图片描述

3.2.消息接收

在consumer服务的SpringRabbitListener中添加方法：

@RabbitListener(queues = "direct.queue1")
public void listenDirectQueue1(String msg) {
    System.out.println("消费者1接收到direct.queue1的消息：【" + msg + "】");
}

@RabbitListener(queues = "direct.queue2")
public void listenDirectQueue2(String msg) {
    System.out.println("消费者2接收到direct.queue2的消息：【" + msg + "】");
}

3.3.消息发送

在publisher服务的SpringAmqpTest类中添加测试方法：

@Test
public void testSendDirectExchange() {
    // 交换机名称
    String exchangeName = "shen.direct";
    // 消息
    String message = "红色警报！日本乱排核废水，导致海洋生物变异，惊现哥斯拉！";
    // 发送消息
    rabbitTemplate.convertAndSend(exchangeName, "red", message);
}

由于使用的red这个key，所以两个消费者都收到了消息：
在这里插入图片描述

我们再切换为blue这个key：

@Test
public void testSendDirectExchange() {
    // 交换机名称
    String exchangeName = "shen.direct";
    // 消息
    String message = "最新报道，哥斯拉是居民自治巨型气球，虚惊一场！";
    // 发送消息
    rabbitTemplate.convertAndSend(exchangeName, "blue", message);
}

你会发现，只有消费者1收到了消息：
在这里插入图片描述

3.4.总结

描述下Direct交换机与Fanout交换机的差异？

Fanout交换机将消息路由给每一个与之绑定的队列
Direct交换机根据RoutingKey判断路由给哪个队列
如果多个队列具有相同的RoutingKey，则与Fanout功能类似
可以看出Direct交换机的功能更加强大，想发送给全部队列就给他们一个相同的RoutingKey，根据分类发送就指定一个RoutingKey。

4.Topic交换机

4.1.说明

Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。
只不过Topic类型Exchange可以让队列在绑定BindingKey 的时候使用通配符！

BindingKey 一般都是有一个或多个单词组成，多个单词之间以.分割，例如： item.insert

通配符规则：

#：匹配一个或多个词
*：匹配不多不少恰好1个词

举例：

item.#：能够匹配item.spu.insert 或者 item.spu
item.*：只能匹配item.spu

图示：
在这里插入图片描述

假如此时publisher发送的消息使用的RoutingKey共有四种：

china.news 代表有中国的新闻消息；
china.weather 代表中国的天气消息；
japan.news 则代表日本新闻
japan.weather 代表日本的天气消息；

解释：

topic.queue1：绑定的是china.# ，凡是以 china.开头的routing key 都会被匹配到，包括：
- china.news
- china.weather
topic.queue2：绑定的是#.news ，凡是以 .news结尾的 routing key 都会被匹配。包括:
- china.news
- japan.news

接下来，我们就按照上图所示，来演示一下Topic交换机的用法。
首先，在控制台按照图示例子创建队列、交换机，并利用通配符绑定队列和交换机。此处步骤略。最终结果如下：
在这里插入图片描述

4.2.消息发送

在publisher服务的SpringAmqpTest类中添加测试方法：

/**
 * topicExchange
 */
@Test
public void testSendTopicExchange() {
    // 交换机名称
    String exchangeName = "shen.topic";
    // 消息
    String message = "喜报！孙悟空大战哥斯拉，胜!";
    // 发送消息
    rabbitTemplate.convertAndSend(exchangeName, "china.news", message);
}

4.3.消息接收

在consumer服务的SpringRabbitListener中添加方法：

@RabbitListener(queues = "topic.queue1")
public void listenTopicQueue1(String msg){
    System.out.println("消费者1接收到topic.queue1的消息：【" + msg + "】");
}

@RabbitListener(queues = "topic.queue2")
public void listenTopicQueue2(String msg){
    System.out.println("消费者2接收到topic.queue2的消息：【" + msg + "】");
}

4.4.总结

描述下Direct交换机与Topic交换机的差异？

Topic交换机接收的消息RoutingKey必须是多个单词，以 **.** 分割
Topic交换机与队列绑定时的bindingKey可以指定通配符
#：代表0个或多个词
*：代表1个词

5.Headers交换机

5.1.说明

在RabbitMQ中使用Headers交换器时，你可以根据消息头里的键值对来进行路由，而不是像direct或topic交换器那样依赖路由键。一个Headers交换器允许你定义一个或多个键值对作为绑定条件，只有当消息的头信息满足这些条件时，它才会被路由到相应的队列。
在这里插入图片描述
Headers也支持一个称作“x-match”的特殊属性，这个属性决定了多个头信息之间是“all”匹配还是“any”匹配：

all表示所有的键值对都必须匹配。这是默认值。
any表示消息只要有任何一个头信息符合条件就可以被路由。
这些属性设置在绑定队列到Headers交换机时的参数中。

接下来，我们就按照上图所示，来演示一下Headers交换机的用法。
首先，在控制台按照图示例子创建队列、交换机，并定义键值绑定队列和交换机。此处步骤略。最终结果如下：
在这里插入图片描述

5.2.消息发送

@Test
public void sendAnimalMessage() {
    String msg = "最新报道，哥斯拉是居民自治巨型气球，虚惊一场！" ;
    Map<String, Object> headers = new HashMap<>();
    headers.put("category", "animal");
    headers.put("type", "rabbit");
    rabbitTemplate.convertAndSend("shen.headers", "", msg, m -> {
        headers.forEach((key, value) -> m.getMessageProperties().setHeader(key, value));
        return m;
    });
}

5.3.消息接收

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class HeaderMessageReceiver {

    @RabbitListener(queues = "headers.queue")
    public void receiveMessage(String message) {
        System.out.println("Received with Headers Exchange: " + message);
    }
}

5.4.总结

RabbitMQ的Headers交换机和Direct和Topic交换机稍有不同。他们的路由行为是基于消息头部（header）中的键值对进行匹配的，这意味着消息是根据它提供的header信息路由到不同的队列中的。
headers的交换机关注消息header中的内容来决定消息该被送往哪个或哪些队列。这种类型的交换机提供了最为灵活的路由方式，允许你通过多个属性来定义路由规则。
当你绑定一个队列到Headers交换机时，你可以指定一个或多个header作为匹配规则。与Direct交换机的路由键相比，Headers交换机使用的是一组键值对。
Headers交换机行为
匹配：如果设置的x-match属性为all，那么只有当消息的header与绑定时指定的所有键值对都相匹配时，消息才会被路由到对应的队列。如果设置为any，那么只需消息header中的任一键值对与绑定时指定的某一个键值对匹配，消息就会被路由到对应的队列。
忽略路由键：与Direct或Topic交换机不同，发送到Headers交换机的消息在使用basic.publish方法时通常设置路由键为一个空字符串，因为Headers交换机并不使用这个字段来决定消息的路由。

Headers交换机是RabbitMQ中较为高级且灵活的特性，它为路由提供了额外的维度，但这也意味着使用和理解上会更复杂一些。在实际应用中，如果消息的路由决策需要根据多个属性来做更复杂的判断，那么Headers交换机可能会是个不错的选择。

5.5.交换机路由键（Routing Key）的处理方式

在RabbitMQ中，交换器（Exchange）的类型决定了消息如何路由到队列中。所有交换器在发送消息时都可以接收一个路由键（Routing Key）参数，但是不同类型的交换器对路由键的处理方式各不相同。fanout、direct、topic 和 headers 是几种常见的交换器类型，它们对路由键的处理逻辑有所区别：

Fanout Exchange:
- Routing Key：在 fanout 交换器上，路由键会被忽略。不管发布到交换器上的消息的路由键是什么，消息都会被发送到所有绑定到该交换器的队列。
- 用途：当您想要将消息广播到所有队列时使用。
Direct Exchange:
- Routing Key：在 direct 交换器上，一个消息会被路由到与消息的路由键完全匹配的队列。
- 用途：当你想要定点发送消息到指定的队列时使用。
Topic Exchange:
- Routing Key：topic 交换器允许使用通配符进行模糊匹配。路由键可以有多个词（words），用点（.）隔开。通配符可以是星号（*）匹配一个词或者井号（#）匹配零个或多个词。
- 用途：适用于发送到多个队列，但队列订阅不是完全相同的情况，实现了模式匹配。
Headers Exchange:
- Routing Key：headers 交换器不依赖于路由键的匹配规则。它使用头信息中的键值对进行匹配。
- 用途：当你想根据消息内容的多个属性来路由消息时使用。

即使 fanout 交换器能让你设定路由键，但它并不会使用这个值来决定消息的路由。在 fanout 交换机的情况下，设置路由键不会产生任何影响。

direct 和 topic 交换器会根据路由键来决定目标队列，而 headers 交换器则完全忽略路由键，转而使用消息头的键值对来进行路由决定。

要体会它们的区别，最好是实际去设置不同的交换器，并发布带有不同路由键或头信息的消息，然后观察消息如何被路由到不同的队列。这样通过实践，你将更容易理解每种类型的交换器和路由行为的差异。

6.声明队列和交换机

在之前我们都是基于RabbitMQ控制台来创建队列、交换机。但是在实际开发时，队列和交换机是程序员定义的，将来项目上线，又要交给运维去创建。那么程序员就需要把程序中运行的所有队列和交换机都写下来，交给运维。在这个过程中是很容易出现错误的。
因此推荐的做法是由程序启动时检查队列和交换机是否存在，如果不存在自动创建。

6.1.基本API

SpringAMQP提供了几个类，用来声明队列、交换机及其绑定关系:

Queue: 用于声明队列，可以用工厂类QueueBuilder构建。
Exchange:用于声明交换机，可以用工厂类ExchangeBuilder构建。
Binding:用于声明队列和交换机的绑定关系，可以用工厂类BindingBuilder构建。

SpringAMQP提供了一个Queue类，用来创建队列：
在这里插入图片描述

SpringAMQP还提供了一个Exchange接口，来表示所有不同类型的交换机：
在这里插入图片描述

我们可以自己创建队列和交换机，不过SpringAMQP还提供了ExchangeBuilder来简化这个过程：
在这里插入图片描述

而在绑定队列和交换机时，则需要使用BindingBuilder来创建Binding对象：
在这里插入图片描述

6.2.fanout示例

注意，发送方只关心发送就行了，一般都是消费者关心是什么样子的，所以在消费方声明。

在consumer中创建一个类，声明队列和交换机：

package com.example.consumer.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class FanoutConfiguration {
    /**
     * 声明交换机
     *
     * @return Fanout类型交换机
     */
    @Bean
    public FanoutExchange fanoutExchange() {
        //ExchangeBuilder.fanoutExchange("").build();
        return new FanoutExchange("shen.fanout");
    }

    /**
     * 第1个队列
     */
    @Bean(name = "queue1")
    public Queue fanoutQueue1() {
        //QueueBuilder.durable("").build();//持久的，带向磁盘。
        return new Queue("fanout.queue1");//默认持久
    }

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding fanoutBuiding1(@Qualifier(value = "queue1") Queue fanoutQueue1, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }

    /**
     * 第2个队列
     */
    @Bean
    public Queue fanoutQueue2() {
        //QueueBuilder.durable("").build();//持久的，带向磁盘。
        return new Queue("fanout.queue2");//默认持久
    }

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding fanoutBuiding2() {
    /*
    在spring中所有加了 Bean 的方法都会被动态代理，所以此处并不是真正的调用了这个方法 fanoutQueue4()，
    而是检查 spring容器 中有没有代理这个 fanoutQueue4()方法，代理了，直接返回这个 bean 对象，
    而不是真正的去执行。
    */
        return BindingBuilder.bind(fanoutQueue2()).to(fanoutExchange());
    }

}

6.3.direct示例

direct模式由于要绑定多个KEY，会非常麻烦，每一个Key都要编写一个binding：

package com.example.consumer.config;

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

@Configuration
public class DirectConfiguration {
    /**
     * 声明交换机
     *
     * @return Direct类型交换机
     */
    @Bean
    public DirectExchange directExchange() {
        return ExchangeBuilder.directExchange("shen.direct").build();
    }

    /**
     * 第1个队列
     */
    @Bean
    public Queue directQueue1() {
        return new Queue("direct.queue1");
    }

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue1WithRed(Queue directQueue1, DirectExchange directExchange) {
        return BindingBuilder.bind(directQueue1).to(directExchange).with("red");
    }

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue1WithBlue(Queue directQueue1, DirectExchange directExchange) {
        return BindingBuilder.bind(directQueue1).to(directExchange).with("blue");
    }

    /**
     * 第2个队列
     */
    @Bean
    public Queue directQueue2() {
        return new Queue("direct.queue2");
    }

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue2WithRed() {
        return BindingBuilder.bind(directQueue2()).to(directExchange()).with("red");
    }

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue2WithYellow(Queue directQueue2, DirectExchange directExchange) {
        return BindingBuilder.bind(directQueue2).to(directExchange).with("yellow");
    }

}

6.4.基于注解声明

基于@Bean的方式声明队列和交换机比较麻烦，Spring还提供了基于注解方式来声明。

例如，我们同样声明Direct模式的交换机和队列：

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "direct.queue1"),
    exchange = @Exchange(name = "shen.direct", type = ExchangeTypes.DIRECT),
    key = {"red", "blue"}
))
public void listenDirectQueue1(String msg){
    System.out.println("消费者1接收到direct.queue1的消息：【" + msg + "】");
}

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "direct.queue2"),
    exchange = @Exchange(name = "shen.direct", type = ExchangeTypes.DIRECT),
    key = {"red", "yellow"}
))
public void listenDirectQueue2(String msg){
    System.out.println("消费者2接收到direct.queue2的消息：【" + msg + "】");
}

是不是简单多了。
再试试Topic模式：

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "topic.queue1"),
    exchange = @Exchange(name = "shen.topic", type = ExchangeTypes.TOPIC),
    key = "china.#"
))
public void listenTopicQueue1(String msg){
    System.out.println("消费者1接收到topic.queue1的消息：【" + msg + "】");
}

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "topic.queue2"),
    exchange = @Exchange(name = "shen.topic", type = ExchangeTypes.TOPIC),
    key = "#.news"
))
public void listenTopicQueue2(String msg){
    System.out.println("消费者2接收到topic.queue2的消息：【" + msg + "】");
}

7.消息转换器

Spring的消息发送代码接收的消息体是一个Object：
在这里插入图片描述

而在数据传输时，它会把你发送的消息序列化为字节发送给MQ，接收消息的时候，还会把字节反序列化为Java对象。
只不过，默认情况下Spring采用的序列化方式是JDK序列化。众所周知，JDK序列化存在下列问题：

数据体积过大
有安全漏洞
可读性差

我们来测试一下。

7.1.测试默认转换器

1）创建测试队列
首先，我们在consumer服务中声明一个新的配置类：
在这里插入图片描述

利用@Bean的方式创建一个队列，具体代码：

package com.example.consumer.config;

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

@Configuration
public class MessageConfig {

    @Bean
    public Queue objectQueue() {
        return new Queue("object.queue");
    }
}

注意，这里我们先不要给这个队列添加消费者，我们要查看消息体的格式。

重启consumer服务以后，该队列就会被自动创建出来了：
在这里插入图片描述

2）发送消息
我们在publisher模块的SpringAmqpTest中新增一个消息发送的代码，发送一个Map对象：

@Test
public void testSendMap() throws InterruptedException {
    // 准备消息
    Map<String,Object> msg = new HashMap<>();
    msg.put("name", "jack");
    msg.put("age", 21);
    // 发送消息
    rabbitTemplate.convertAndSend("object.queue", msg);
}

发送消息后查看控制台：
在这里插入图片描述

可以看到消息格式非常不友好。

7.2.配置JSON转换器

显然，JDK序列化方式并不合适。我们希望消息体的体积更小、可读性更高，因此可以使用JSON方式来做序列化和反序列化。

在publisher和consumer两个服务中都引入依赖：

<dependency>
    <groupId>com.fasterxml.jackson.dataformat</groupId>
    <artifactId>jackson-dataformat-xml</artifactId>
    <version>2.9.10</version>
</dependency>

注意，如果项目中引入了spring-boot-starter-web依赖，则无需再次引入Jackson依赖。

配置消息转换器，在publisher和consumer两个服务的启动类中添加一个Bean即可：

@Bean
public MessageConverter messageConverter(){
    // 1.定义消息转换器
    Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();
    // 2.配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息
    jackson2JsonMessageConverter.setCreateMessageIds(true);
    return jackson2JsonMessageConverter;
}

消息转换器中添加的messageId可以便于我们将来做幂等性判断。

此时，我们到MQ控制台删除object.queue中的旧的消息。然后再次执行刚才的消息发送的代码，到MQ的控制台查看消息结构：
在这里插入图片描述
可以看到，消息大小小了很多，只要24bytes了。

7.3.消费者接收Object

我们在consumer服务中定义一个新的消费者，publisher是用Map发送，那么消费者也一定要用Map接收，格式如下：

@RabbitListener(queues = "object.queue")
public void listenSimpleQueueMessage(Map<String, Object> msg) throws InterruptedException {
    System.out.println("消费者接收到object.queue消息：【" + msg + "】");
}

文章来源:https://blog.csdn.net/m0_62943934/article/details/135036290
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若内容造成侵权/违法违规/事实不符，请联系我的编程经验分享网邮箱：veading@qq.com进行投诉反馈，一经查实，立即删除！

RabbitMq交换机详解

目录

1.交换机类型

2.Fanout交换机

2.1.声明队列和交换机

2.2.消息发送

2.3.消息接收

2.4.总结

3.Direct交换机

3.1.声明队列和交换机

3.2.消息接收

3.3.消息发送

3.4.总结

4.Topic交换机

4.1.说明

4.2.消息发送

4.3.消息接收

4.4.总结

5.Headers交换机

5.1.说明

5.2.消息发送

5.3.消息接收

5.4.总结

5.5.交换机路由键（Routing Key）的处理方式

6.声明队列和交换机

6.1.基本API

6.2.fanout示例

6.3.direct示例

6.4.基于注解声明

7.消息转换器

7.1.测试默认转换器

7.2.配置JSON转换器

7.3.消费者接收Object