时间戳

首页 > 技术文章 > HM-SpringCloud微服务系列4.3.3【发布/订阅模型及其三种交换机】

HM-SpringCloud微服务系列4.3.3【发布/订阅模型及其三种交换机】

yppah 2022-01-28 13:13 原文

1 发布/订阅模型简介

image

  1. 可以看到，在订阅模型中，多了一个exchange角色，而且过程略有变化：
    • Publisher：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是exchange交换机
    • Exchange：交换机。一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。Exchange有以下3种类型：
      • Fanout：广播，将消息交给所有绑定到交换机的队列
      • Direct：定向，把消息交给符合指定routing key 的队列
      • Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列
    • Consumer：消费者，与以前一样，订阅队列，没有变化
    • Queue：消息队列也与以前一样，接收消息、缓存消息。
  2. Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！

2 发布/订阅模型：Fanout-Exchange广播交换机

2.1 Fanout-Exchange

  1. Fanout Exchange 会将接收到的消息广播到每一个跟其绑定的queue
    image
  2. 在广播模式下，消息发送流程是这样的：
    1. 可以有多个队列
    2. 每个队列都要绑定到Exchange（交换机）
    3. 生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定
    4. 交换机把消息发送给绑定过的所有队列
    5. 订阅队列的消费者都能拿到消息

2.2 案例

image

  • 创建一个交换机 itcast.fanout，类型是Fanout
  • 创建两个队列fanout.queue1和fanout.queue2，绑定到交换机itcast.fanout

2.2.1 声明交换机、队列和绑定

  1. Spring提供了声明交换机、队列、绑定关系的API，其中Exchange接口表示所有不同类型的交换机
    image
  2. 在consumer服务中创建一个类，添加@Configuration注解，并声明FanoutExchange、Queue和绑定关系对象Binding
package cn.itcast.mq.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.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class FanoutConfig {

    // 声明交换机：itcast.fanout
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("itcast.fanout");
    }

    // 声明队列1：fanout.queue1
    @Bean
    public Queue fanoutQueue1() { // 注意Queue来自amqp.core包
        return new Queue("fanout.queue1");
    }

    // 绑定队列1到交换机
    @Bean
    public Binding fanoutBinding1(Queue fanoutQueue1, FanoutExchange fanoutExchange) { // 注意Binding来自amqp.core包
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }

    // 声明队列2：fanout.queue2
    @Bean
    public Queue fanoutQueue2() {
        return new Queue("fanout.queue2");
    }

    // 绑定队列2到交换机
    @Bean
    public Binding fanoutBinding2(Queue fanoutQueue2, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }
}

image
image
image
3. 启动测试
image
image
image
image
image

2.2.2 消息接收

  • 在consumer服务的SpringRabbitListener类中，添加两个方法，分别监听fanout.queue1和fanout.queue2：
    @RabbitListener(queues = "fanout.queue1")
    public void listenFanoutQueue1(String msg) {
        System.out.println("消费者1接收到fanout.queue1消息：【" + msg +"】");
    }

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

image

2.2.3 消息发送

  • 在publisher服务的SpringAmqpTest类中添加测试方法
    @Test
    public void testSentFanoutExchange() {
        // 交换机名称
        String exchangeName = "itcast.fanout";
        // 消息
        String message = "hi, every one";
        // 发送消息
        rabbitTemplate.convertAndSend(exchangeName, "", message);
    }

image

2.2.4 测试

image
image
image
image
image
image

2.3 小结

  1. 交换机的作用是什么？
    • 接收publisher发送的消息
    • 将消息按照规则路由到与之绑定的队列
    • 不能缓存消息，路由失败，消息丢失
    • FanoutExchange的会将消息路由到每个绑定的队列
  2. 声明队列、交换机、绑定关系的Bean是什么？
    • Queue
    • FanoutExchange
    • Binding

3 发布/订阅模型：Direct-Exchange定向交换机

3.1 Direct-Exchange

  • 在Fanout模式中，一条消息，会被所有订阅的队列都消费。但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。
    image
  • 在Direct模型下：
    • 队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）
    • 消息的发送方在 向 Exchange发送消息时，也必须指定消息的 RoutingKey
    • Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey与消息的 Routing key完全一致，才会接收到消息

3.2 案例

image

  1. 利用@RabbitListener声明Exchange、Queue、RoutingKey
  2. 在consumer服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2
  3. 在publisher中编写测试方法，向itcast. direct发送消息

3.2.1 基于注解声明队列和交换机

  • 基于@Bean的方式声明队列和交换机比较麻烦，Spring还提供了基于注解方式来声明。
  • 在consumer的SpringRabbitListener中添加两个消费者方法，分别监听direct.queue1和direct.queue2。
  • 同时基于注解@RabbitListener声明Exchange、Queue、RoutingKey。
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue1"),
            exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT), //type也可以直接赋值为"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 = "itcast.direct", type = "direct"),
            key = {"red", "yellow"}
    ))
    public void listenDirectQueue2(String msg) {
        System.out.println("消费者2接收到direct.queue2消息：【" + msg +"】");
    }

image

3.2.2 消息发送

  • 在publisher服务发送消息到DirectExchange：在publisher服务的SpringAmqpTest类中添加测试方法
@Test
    public void testSendDirectExchange() {
        // 交换机名称
        String exchangeName = "itcast.direct";
        // 消息
        String message = "hi red"; // blue yellow red
        // 发送消息
        rabbitTemplate.convertAndSend(exchangeName, "red", message);
    }

3.2.3 测试

image
image
image
image
image
image
image
image
image
image

3.3 小结

  1. 描述下Direct交换机与Fanout交换机的差异？
    • Fanout交换机将消息路由给每一个与之绑定的队列
    • Direct交换机根据RoutingKey判断路由给哪个队列
    • 如果多个队列具有相同的RoutingKey，则与Fanout功能类似
  2. 基于@RabbitListener注解声明队列和交换机有哪些常见注解？
    • @Queue
    • @Exchange

4 发布/订阅模型：Topic-Exchange通配符交换机

4.1 Topic-Exchange

  • Topic：主题、话题
  • TopicExchange与DirectExchange类似，都可以根据RoutingKey把消息路由到不同的队列。
  • 区别在于TopicExchange可以让队列在绑定RoutingKey的时候使用通配符。
  • Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert。
  • Queue与Exchange指定BindingKey时可以使用通配符（通配符规则）：
    • #：匹配一个或多个词（代指0个或多个单词）
    • *：匹配不多不少恰好1个词（代指一个单词）
  • 举例：
    • item.#：能够匹配item.spu.insert 或者 item.spu
    • item.*：只能匹配item.spu
  • 图示：
    image
  • 解释：
    • Queue1：绑定的是china.# ，因此凡是以 china.开头的routing key 都会被匹配到。包括china.news和china.weather
    • Queue2：绑定的是#.news ，因此凡是以 .news结尾的 routing key 都会被匹配。包括china.news和japan.news

4.2 案例

image

  1. 利用@RabbitListener声明Exchange、Queue、RoutingKey
  2. 在consumer服务中，编写两个消费者方法，分别监听topic.queue1和topic.queue2
  3. 在publisher中编写测试方法，向itcast. topic发送消息

4.2.1 消息接收

  • 在consumer服务的SpringRabbitListener中添加方法：
    • 在consumer服务中，编写两个消费者方法，分别监听topic.queue1和topic.queue2，
    • 并利用@RabbitListener声明Exchange、Queue、RoutingKey
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue1"),
            exchange = @Exchange(name = "itcast.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 = "itcast.topic", type = "topic"),
            key = "#.news"
    ))
    public void listenTopicQueue2(String msg) {
        System.out.println("消费者2接收到topic.queue2消息：【" + msg +"】");
	}

image

4.2.2 消息发送

  • 在publisher服务的SpringAmqpTest类中添加测试方法：
@Test
    public void testSendTopicExchange() {
        // 交换机名称
        String exchangeName = "itcast.topic";
        // 消息
        String message = "itheima在哪上市了？";
        // 发送消息
        rabbitTemplate.convertAndSend(exchangeName, "china.news", message);
    }

4.2.3 测试

image
image
image
image
image
image
image
image

4.3小结

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

推荐阅读