RabbitMQ基础篇

• RabbitMQ介绍及快速入门

一、基础篇

1.1 初识MQ

1.1.1 同步和异步通讯

• 微服务间通讯有同步和异步两种方式：

  • 同步通讯：就像打电话，需要实时响应。
  • 异步通讯：就像发邮件，不需要马上回复。

  两种方式各有优劣，打电话可以立即得到响应，但是你却不能跟多个人同时通话。发送邮件可以同时与多个人收发邮件，但是往往响应会有延迟。

同步通讯

• 我们之前学习的 Feign 调用就属于同步方式，虽然调用可以实时得到结果，但存在下面的问题：

  • 耦合度高

  • 性能和吞吐能力下降

  • 有额外的资源消耗

  • 有级联失败问题

  • 异步调用则可以避免上述问题。

异步通讯

• 异步调用方式其实就是基于消息通知的方式，一般包含三个角色：

  • 消息发送者：投递消息的人，就是原来的调用方。

  • 消息代理：管理、暂存、转发消息。

  • 消息接收者：接收和处理消息的人，就是原来的服务提供方。

  • 好处：

    • 吞吐量提升：无需等待订阅者处理完成，响应更快速
    • 故障隔离：服务没有直接调用，不存在级联失败问题
    • 调用间没有阻塞，不会造成无效的资源占用
    • 耦合度低，扩展性强：每个服务都可以灵活插拔，可替换
    • 缓存消息，流量削峰填谷：不管发布事件的流量波动多大，都由 Broker 接收，订阅者可以按照自己的速度去处理事件。

  • 缺点：

    • 不能立即得到调用结果，时效性差
    • 架构复杂了，业务没有明显的流程线，不好管理
    • 需要依赖于 Broker 的可靠、安全、性能

1.1.2 MQ技术选型

• MQ（Message Queue），中文是消息队列，字面来看就是存放消息的队列，也就是异步调用中的 Broker。

  • 比较常见的 MQ 实现：

    • ActiveMQ
    • RabbitMQ
    • RocketMQ
    • Kafka

  • 几种常见 MQ 的对比：

    • 追求可用性：Kafka、 RocketMQ 、RabbitMQ
    • 追求可靠性：RabbitMQ、RocketMQ
    • 追求吞吐能力：RocketMQ、Kafka
    • 追求消息低延迟：RabbitMQ、Kafka

1.2 RabbitMQ快速入门

1.2.1 介绍和安装

• RabbitMQ 是基于 Erlang 语言开发的开源消息通信中间件。

  • RabbitMQ 的整体架构及核心概念：

    • publisher：生产者
    • consumer：消费者
    • exchange：交换机，负责消息路由
    • queue：队列，存储消息
    • virtualHost：虚拟主机，起到数据隔离的作用，不同的 virtualhost 之间的数据是隔离的。

• Centos7安装RabbitMQ与使用（超详细）-CSDN博客

1.2.2 快速入门

• 是在 rabbitmq 管理界面的演示，看下就行了。

1.2.3 数据隔离

• 同 1.2.2。就是演示了不同的 virtualhost 之间的数据是隔离的。

1.3 Java客户端

1.3.1 快速入门

• AMQP（Advanced Message Queuing Protocol），高级消息队列协议，是用于在应用程序之间传递业务消息的开放标准。该协议与语言和平台无关，更符合微服务中独立性的要求。

  • Spring AMQP，是基于 AMQP 协议定义的一套 API 规范，提供了模板来发送和接收消息。包含两部分，其中 spring-amqp 是基础抽象，
    spring-rabbit 是底层的默认实现。

    Spring AMQP 提供了三个功能：

    • 自动声明队列、交换机及其绑定关系
    • 基于注解的监听器模式，异步接收消息
    • 封装了 RabbitTemplate 工具，用于发送消息

  • 在 Springboot 中使用 RabbitMQ 的步骤如下：

    1. 引入相关依赖：

       <!--AMQP依赖，包含RabbitMQ-->
       <dependency>
           <groupId>org.springframework.boot</groupId>
           <artifactId>spring-boot-starter-amqp</artifactId>
       </dependency>

    2. 在 application.yml 文件中配置 rabbitmq-server 相关信息（publisher 端和 consumer 端都要配置）：

       spring:
         rabbitmq:
           host: 192.168.44.134 # 主机名
           port: 5672 # 端口
           virtual-host: / # 虚拟主机
           username: admin # 用户名
           password: admin # 密码

    3. 使用 RabbitMQ 进行消息的首发：

       • 利用 RabbitTemplate 发送消息 —> publisher

         @SpringBootTest
         public class SpringAmqpTest {
             @Autowired
             private RabbitTemplate rabbitTemplate;
             @Test
             public void testSimpleQueue() {
                 // 队列名称
                 String queueName = "simple.queue";
                 // 消息
                 String message = "hello, spring amqp!";
                 // 发送消息
                 rabbitTemplate.convertAndSend(queueName, message);
             }
         }

       • 利用 @RabbitListener 注解声明要监听的队列，监听消息 —> consumer

         @Component
         public class SpringRabbitListener {
             @RabbitListener(queues = "simple.queue")
             public void listenSimpleQueueMessage(String msg) throws InterruptedException {
                 System.out.println("spring 消费者接收到消息：【" + msg + "】");
             }
         }

1.3.2 work模型

• Work queues，任务模型。简单来说就是让多个消费者绑定到同一个队列，共同消费队列中的消息。

  • 当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。

    • 此时就可以使用 work 模型，多个消费者共同处理消息处理，速度就能大大提高了。

    默认情况下，RabbitMQ 会将消息依次轮询投递给绑定在队列上的每一个消费者，同一条消息只会被一个消费者处理（所有消费者最终消费的消息量是相等的）。但这并没有考虑到消费者处理消息的速度，可能出现消息堆积。

    • 在 Spring 中有一个简单的配置，可以解决这个问题。修改 consumer 服务的 application.yml 文件，添加配置：

      spring:
          rabbitmq:
            listener:
              simple:
                prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息

      • 这样，消费能力强的消费者就可以消费更多的消息，俗称“能者多劳”。

• 总结 work 模型的使用：

  • 多个消费者绑定到一个队列，队列中同一条消息只会被一个消费者处理。
  • 通过设置 prefetch 来控制消费者预取的消息数量，能者多劳。

1.3.3 发布/订阅

• 发布/订阅的模型如图：

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

  • Publisher：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）
  • Exchange：交换机，一方面，接收生产者发送的消息；另一方面，将消息按照规则路由到与之绑定的队列，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于 Exchange 的类型。Exchange 有以下 3 种类型：
    • Fanout：广播，将消息交给所有绑定到交换机的队列。
    • Direct：定向，把消息交给符合指定 routing key 的队列。
    • Topic：通配符，把消息交给符合 routing pattern（路由模式） 的队列。
  • Queue：消息队列也与以前一样，接收消息、缓存消息。
  • Consumer：消费者，与以前一样，订阅队列，没有变化。

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

1.3.3.1 Fanout交换机

• Fanout Exchange 会将接收到的消息广播到每一个跟其绑定的 queue，所以也叫广播模式。

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

1.3.3.2 Direct交换机

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

  Direct Exchange 会将接收到的消息根据规则路由到指定的 Queue，因此称为定向路由。

  • 在 Direct 模型下：
    • 每一个 Queue 都与 Exchange 设置一个 BindingKey。
    • 发布者发送消息时，需要指定消息的 RoutingKey。
    • Exchange：将消息路由到 BindingKey 与消息 RoutingKey 一致的队列。

1.3.3.3 Topic交换机

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

  • Routingkey 一般都是由一个或多个单词组成，多个单词之间以 “.” 分割，例如： item.insert

  • 通配符规则：

    • #：匹配零个或多个单词
    • *：匹配一个单词

    例如，item.#：能够匹配 item.spu.insert 或者 item.spu 或者 item；item.*：只能匹配 item.spu。

1.3.4 声明队列和交换机的方式

基于Bean的方式

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

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

  @Configuration
  public class FanoutConfig {
      /**
       * 声明交换机
       * @return Fanout类型交换机
       */
      @Bean
      public FanoutExchange fanoutExchange(){
          // return ExchangeBuilder.fanoutExchange("itcast.fanout").build();
          return new FanoutExchange("itcast.fanout");
      }
  
      /**
       * 第1个队列
       */
      @Bean
      public Queue fanoutQueue1(){
          // return QueueBuilder.durable("fanout.queue1").build()
          return new Queue("fanout.queue1");
      }
  
      /**
       * 绑定队列和交换机
       */
      @Bean
      public Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
          return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
      }
  
      /**
       * 第2个队列
       */
      @Bean
      public Queue fanoutQueue2(){
          return new Queue("fanout.queue2");
      }
  
      /**
       * 绑定队列和交换机
       */
      @Bean
      public Binding bindingQueue2(){
          return BindingBuilder.bind(fanoutQueue2()).to(fanoutExchange());
      }
  }

基于注解的方式

• SpringAMQP 还提供了基于 @RabbitListener 注解来声明队列和交换机的方式：

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

1.3.5 消息转换器

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

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

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

• 如果要使用 JSON 序列化代替默认的 JDK 序列化，要做两件事：

  • 在 publisher 和 consumer 中都要引入 jackson 依赖：

    <dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
    </dependency>

  • 在 publisher 和 consumer 中都要配置 MessageConverter：

    @Bean
    public MessageConverter messageConverter(){
    	return new Jackson2JsonMessageConverter();
    }