【云原生进阶之PaaS中间件】第四章RabbitMQ-1-简介及工作模式

江中散人_Jun

发布于 2024-02-18 08:37:25

1240

发布于 2024-02-18 08:37:25

文章被收录于专栏：云原生布道专栏云原生布道专栏

1 RabbitMQ简介

1.1 基本介绍

RabbitMQ 是一个由 Erlang 语言开发的 AMQP 的开源实现。AMQP（Advanced Message Queue：高级消息队列协议）它是应用层协议的一个开放标准，为面向消息的中间件设计，基于此协议的客户端与消息中间件可传递消息，并不受产品、开发语言等条件的限制。RabbitMQ 最初起源于消息系统，用于在分布式系统中存储转发消息，具体有如下一些特点：

可靠性： RabbitMQ 使用一些机制来保证可靠性，比如持久化、传输确认机制（ack）和发布确认等。
灵活的路由策略：在消息进入队列之前，通过 Exchange 来路由消息，对于典型的路由功能，RabbitMQ 已经提供了一些内置的 Exchange 来实现。针对复杂的路由功能，可以将多个 Exchange 绑在一起，也通过插件机制实现自己的 Exchange。
消息集群：多个 RabbitMQ 服务器可以组成一个集群，形成一个逻辑 Broker。
高可用：队列可以在集群中的集群上进行镜像，使得在部分节点出问题的情况下队列仍然可用。
多种协议： RabbitMQ 支持多种消息队列协议，比如 STOMP、MQTT 等。
多语言客户端： RabbitMQ 支持多种常用的语言，比如：Java、.NET 等
管理界面： RabbitMQ 提供了一个易用的用户界面，使得用户可以监控和管理消息 Broker 的许多方面。

1.2 业务场景

1.2.1 异步处理

如：用户注册发送，注册邮件、注册短信，

传统做法：

1、串行（先发送邮件、再发短信）。问题：持续时间长

2、并行（将注册信息写入数据库后，同时发送邮件、短信），速度快、但不能满足高吞吐需求。

消息队列做法：

将数据写入数据库、同时发送消息给发送邮件和注册，异步处理

1.2.2 应用解耦

如：双十一购物节，用户下单后、订单系统通知库存系统。

传统做法:

订单系统调用库存系统接口。问题：库存接口故障，订单就会失败，而损失大量订单

消息队列做法

订单系统：下单，订单系统完成持久化，将消息写入队列，返回下单成功给用户

库存系统：订阅下单的消息，获取下单消息，进行库操作，就算库存系统故障，消息队列也能保证消息可靠投递，不会导致消息丢失。

1.2.3 流量削峰

如：秒杀活动、一般会因为流量过大，导致应用挂掉，一般在应用前端加入消息队列。

作用：

1、可以控制活动人数，超过一定阈值，订单直接丢弃。

2、可以缓解短时间的高流量压垮应用（应用程序按自己的最大处理能力获取订单）

消息队列做法:

1、用户的请求，服务器收到后，首先写入消息队列，加入消息队列长度最大值，则直接抛弃用户请求或跳转到错误页面

2、秒杀业务根据消息队列中的请求信息，再做后续处理

2 RabbitMQ系统架构

2.1 基本概念

2.1.1 Message（消息）

消息是不具名的，它由消息头和消息体组成。消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括routing-key（路由键）、priority（相对于其他消息的优先权）、delivery-mode（传输模式，指出该消息可能需要持久化存储）等。

2.1.2 Publisher

消息生产者，也是一个向交换器发布消息的客户端应用程序，就是投递消息的程序。

2.1.3 Consumer

消息消费者，表示一个从消息队列中取得消息的客户端应用程序，就是接受消息的程序。

2.1.4 Broker

表示消息队列服务器实体。它提供一种传输服务，它的角色就是维护一条从生产者到消费者的路线，保证数据能按照指定的方式进行传输。

2.1.5 Virtual Host（虚拟主机）

Virtual host是一个虚拟主机的概念，一个Broker中可以有多个Virtual host，每个Virtual host都有一套自己的Exchange和Queue，同一个Virtual host中的Exchange和Queue不能重名，不同的Virtual host中的Exchange和Queue名字可以一样。这样，不同的用户在访问同一个RabbitMQ Broker时，可以创建自己单独的Virtual host，然后在自己的Virtual host中创建Exchange和Queue，很好地做到了不同用户之间相互隔离的效果。每个vhost本质上就是一个mini版的rabbitmq服务器，拥有自己的队列、交换器、绑定和权限机制。vhost是AMQP概念的基础，必须在连接时指定，rabbitmq默认的vhost是 / 。

2.1.6 Exchange

交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。消息交换机，它指定消息按什么规则，路由到哪个队列。

2.1.7 Routing Key

路由关键字，exchange根据这个关键字进行消息投递。

2.1.8 Binding（绑定）

用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。

它的作用就是把exchange和queue按照路由规则绑定起来。

绑定其实就是关联了exchange和queue，或者这么说：queue对exchange的内容感兴趣，exchange要把它的Message deliver到queue。

2.1.9 Queue（消息队列）

消息的载体，每个消息都会被投到一个或多个队列，等待消费者连接到这个队列将其取走。它是消息的容器，也是消息的终点。

2.1.10 Connection

每个producer（生产者）或者consumer（消费者）要通过RabbitMQ发送与消费消息，首先就要与RabbitMQ建立连接，这个连接就是Connection。Connection是一个TCP长连接。

2.1.11 Channel（信道，通道）

Channel是在Connection的基础上建立的虚拟连接，RabbitMQ中大部分的操作都是使用Channel完成的，比如：声明Queue、声明Exchange、发布消息、消费消息等。

看到此处，你是否有这样一个疑问：既然已经有了Connection，我们完全可以使用Connection完成Channel的工作，为什么还要引入Channel这样一个虚拟连接的概念呢？因为现在的程序都是支持多线程的，如果没有Channel，那么每个线程在访问RabbitMQ时都要建立一个Connection这样的TCP连接，对于操作系统来说，建立和销毁TCP连接是非常大的开销，在系统访问流量高峰时，会严重影响系统性能。

Channel就是为了解决这种问题，通常情况下，每个线程创建单独的Channel进行通讯，每个Channel都有自己的channel id帮助Broker和客户端识别Channel，所以Channel之间是完全隔离的。Connection与Channel之间的关系可以比作光纤电缆，如果把Connection比作一条光纤电缆，那么Channel就相当于是电缆中的一束光纤，属于一种典型的多路复用连接技术。

2.1.12 集群节点

RabbitMQ的集群节点包括内存节点、磁盘节点。顾名思义内存节点就是将所有数据放在内存，磁盘节点将数据放在磁盘。不过，如果在投递消息时，打开了消息的持久化，那么即使是内存节点，数据还是安全的放在磁盘。

一个rabbitmq集群中可以共享 user，vhost，queue，exchange等，所有的数据和状态都是必须在所有节点上复制的，一个例外是，那些当前只属于创建它的节点的消息队列，尽管它们可见且可被所有节点读取。rabbitmq节点可以动态的加入到集群中，一个节点它可以加入到集群中，也可以从集群进行一个基本的负载均衡。

集群中有两种节点：

1.内存节点：只保存状态到内存（一个例外的情况是：持久的queue的持久内容将被保存到disk）

2.磁盘节点：保存状态到内存和磁盘。

内存节点虽然不写入磁盘，但是它执行比磁盘节点要好。集群中，只需要一个磁盘节点来保存状态就足够了如果集群中只有内存节点，那么不能停止它们，否则所有的状态，消息等都会丢失。

2.2 重要组件实现

2.2.1 Exchange机制（交换器）

Exchange是一个比较重要的概念，它是消息到达RabbitMQ的第一站，主要负责根据不同的分发规则将消息分发到不同的Queue，供订阅了相关Queue的消费者消费到指定的消息。那Exchange有哪些分发消息的规则呢？这就要说到Exchange的4种类型了：direct、fanout、topic、headers。

在介绍这4种类型的Exchange之前，我们先来了解一下另外一个比较重要的概念：Routing key，翻译成中文就是路由键。当我们创建好Exchange和Queue之后，需要使用Routing key（通常叫作Binding key）将它们绑定起来，producer在向Exchange发送一条消息的时候，必须指定一个Routing key，然后Exchange接收到这条消息之后，会解析Routing key，然后根据Exchange和Queue的绑定规则，将消息分发到符合规则的Queue中。

接下来，我们根据上面的流程再来详细介绍下面4种类型的Exchange。

2.2.1.1 direct类型转发

direct的意思是直接的，direct类型的Exchange会将消息转发到指定Routing key的Queue上，Routing key的解析规则为精确匹配。也就是只有当producer发送的消息的Routing key与某个Binding key相等时，消息才会被分发到对应的Queue上。

比如我们现在有一个direct类型的Exchange，它下面绑定了三个Queue，Binding key分别是ORDER/GOODS/STOCK：

然后我们向该Exchange中发送一条消息，消息的Routing key是ORDER：

按照规则分析，这条消息应该被路由到MY_EXCHANGE_ORDER_QUEUE这个Queue。消息发送成功之后，我们去Queues中查看，发现确实只有MY_EXCHANGE_ORDER_QUEUE这个QUEUE接收到了一条消息。

进入这个队列，通过getMessage取出消息查看，确实是我们刚才手动发送的那条消息。

所以，direct类型的Exchange在分发消息时，必须保证producer发送消息的Routing key与Exchange和Queue绑定的Binding key相等才可以。

2.2.1.2 fanout 类型（广播）转发

fanout是扇形的意思，该类型通常叫作广播类型。fanout类型的Exchange不处理Routing key，而是会将发送给它的消息路由到所有与它绑定的Queue上。

比如我们现在有一个fanout类型的Exchange，它下面绑定了三个Queue，Binding key分别是ORDER/GOODS/STOCK：

然后我们向该Exchange中发送一条消息，消息的Routing key随便填一个值abc：

按照规则分析，这条消息应该被路由到所有与该Exchange绑定的Queue，即三个Queue都应该会受到消息。消息发送成功之后，我们去Queues中查看，发现确实每个QUEUE都接收到了一条消息。

进入这三个QUEUE，通过getMessage取出消息查看，确实是我们刚才手动发送的那条消息。

所以，fanout类型的Exchange不管Routing key是什么，它都会将接收到的消息分发给所有与自己绑定了的Queue上。

2.2.1.3 topic类型转发

topic的意思是主题，topic类型的Exchange会根据通配符对Routing key进行匹配，只要Routing key满足某个通配符的条件，就会被路由到对应的Queue上。通配符的匹配规则如下：

Routing key必须是一串字符串，每个单词用“.”分隔；
符号“#”表示匹配一个或多个单词；
符号“*”表示匹配一个单词。

例如：“*.123” 能够匹配到 “abc.123”，但匹配不到 “abc.def.123”；“#.123” 既能够匹配到 “abc.123”，也能匹配到 “abc.def.123”。

比如我们现在有一个topic类型的Exchange，它下面绑定了4个Queue，Binding key分别是 *.ORDER、GOODS.*、#.STOCK、USER.#。

然后我们向该Exchange中发送一条消息，消息的Routing key为：USER.ABC.ORDER。

按照规则分析，USER.ABC.ORDER这个Routing key只可以匹配到 “USER.#” ，所以，这条消息应该被路由到MY_TOPIC_USER_QUEUE这个Queue中。消息发送成功之后，我们去Queues中查看，发现结果符合我们的预期。

进入这个QUEUE，通过getMessage取出消息查看，确实是我们刚才手动发送的那条消息。

2.2.1.4 headers类型转发

日常工作中，以上三种类型的Exchange已经能够满足我们基本上所有的需求了，headers模式并不经常使用，我们只需要对headers Exchange有一个基本的了解就可以了。

headers Exchange中，Exchange与Queue之间的绑定不再通过Binding key绑定，而是通过Arguments绑定。比如我们现在有一个headers类型的Exchange，下面通过不同的Arguments绑定了三个Queue：

producer在发送消息时可以添加headers属性，Exchange接收到消息后，会解析headers属性，只要我们上面配置的Arguments中的所有属性全部被包含在Headers中并且值相等，那么这条消息就会被路由到对应的Queue中。

比如我们向上面的Exchange中发送一条消息，消息的Headers中添加“x=1”：

根据规则，只有queue1这个队列满足x=1的条件，queue2中的y=2条件不满足，所以，消息应该只被路由到queue1队列中。消息发送成功后，我们可以看到queue1确实收到了消息：

并且这条消息就是我们刚才手动发送的消息：

然后我们再发送一条消息，消息的headers中有两个属性：x=1，y=2：

根据规则，queue1的x=1的条件满足，queue2的x=1、y=2的条件满足，queue3的y=2的条件满足，所以，这三个Queue应该都能够收到这条消息。消息发送成功后，结果符合预期：

这条消息就是我们刚才手动发送的消息：

3 五种工作模型示例

springboot依赖配置：

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

yml配置：

spring:
  application:
    name: rabbitmq
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    virtual-host: /

3.1 Hello World简单模型

一对一消费，只有一个消费者能接收到：

消费者

@Component
public class HolloWordListener {
    // @RabbitListener(queues = ("simple.queue")) // queues需手动先创建队列
    @RabbitListener(queuesToDeclare = @Queue("simple.queue"))  // queuesToDeclare 自动声明队列
    public void holloWordListener(String message){
        System.out.println("message = " + message);
    }
}

生产者

@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void testSimpleQueue() {
    String queueName = "simple.queue"; // 队列名称
    String message = "heel，simple.queue"; // 要发送的消息
    rabbitTemplate.convertAndSend(queueName,message);
}

3.2 Work queues工作队列

多个消费者，你一个我一个分配消费消息，有预取机制，默认公平消费，可配置能者多劳模式，谁完成的快，谁多做一点。

消费者

@Component
public class WoekWordListener {
    @RabbitListener(queuesToDeclare = @Queue("workQueue")) // queuesToDeclare 自动声明队列
    public void holloWordListener(String message) throws InterruptedException {
        Thread.sleep(200);
        System.out.println("message1 = " + message);
    }

    @RabbitListener(queuesToDeclare = @Queue("workQueue")) // queuesToDeclare 自动声明队列
    public void holloWordListener1(String message) throws InterruptedException {
     Thread.sleep(400);
     System.out.println("message2 = " + message);
    }
}

生产者

@Autowired
private RabbitTemplate rabbitTemplate;

@Test
public void testWorkQueue(){
    String queueName = "workQueue";
    String message = "hello，work.queue__";
    for (int i = 0; i < 10; i++) {
     rabbitTemplate.convertAndSend(queueName,message+i);
     System.out.println("i = " + i);
    }
}

取消预取机制，能者多劳配置：

spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    virtual-host: /
    listener:
      simple:
        prefetch: 1 # 每次只能获取一条，处理完成才能获取下一条

3.3 Publish/Subscribe发布订阅模型

发布订阅模式与之前案例的区别就是允许将同一消息发送给多个消费者。实现方式是加入了exchange(交换机)，注意：交换机是不缓存消息的。

使用fanout交换机，会将接收到的消息路由到每一个跟其绑定的queue(队列)。

消费者

// 消费者直接绑定交换机，指定类型为fanout
@Component
public class FanoutExchangeListener {
    // 不指定队列，消息过了就没了
    //  @RabbitListener(bindings = {@QueueBinding(value = @Queue,exchange = @Exchange(value = "fanoutTest",type = ExchangeTypes.FANOUT))})
    // 指定队列，可以接收缓存到队列里的消息
    @RabbitListener(bindings = {@QueueBinding(value = @Queue(value ="test",durable = "true" ),exchange = @Exchange(value = "fanoutTest",type = ExchangeTypes.FANOUT))})
    public void reveivel(String message){
        System.out.println("message = " + message);
    }
    
    @RabbitListener(bindings = {@QueueBinding(value = @Queue,exchange = @Exchange(value = "fanoutTest",type = ExchangeTypes.FANOUT))})
    public void reveivel2(String message){
        System.out.println("message1 = " + message);
    }
}

生产者

@Autowired
private RabbitTemplate rabbitTemplate;

@Test
public void tesyPubSubQueue(){
    // 参数1：交换机名称 ， 参数2routingKey,（fanout类型可不写） , 参数3，消息内容
    rabbitTemplate.convertAndSend("fanoutTest","","消息内容");
}

3.4 Routing路由模型

routing模型也是将消息发送到交换机。使用的是Direct类型的交换机，会将接收到的消息根据规则路由到指定的Queue(队列)，因此称为路由模式。

消费者

// 消费者直接绑定交换机，指定类型为direct，并指定key表示能消费的key
@Component
public class RoutingExchangeListener {
    // 不指定队列，消息过了就没了
    //  @RabbitListener(bindings = {@QueueBinding(value = @Queue,exchange = @Exchange(value = "direstTest",type = ExchangeTypes.DIRECT),key = {"info","error"})})
    // 指定队列，可以接收缓存到队列里的消息
    // key = {"info","error"} 表示我能接收到routingKey为 info和error的消息
    @RabbitListener(bindings = {@QueueBinding(value = @Queue(value ="test1",durable = "true" ),exchange = @Exchange(value = "direstTest",type = ExchangeTypes.DIRECT),key = {"info","error"})})
    public void receivel(String message){
    System.out.println("message = " + message);
    }

    // key = {"error"} 表示我只能接收到routingKey为 error的消息
    @RabbitListener(bindings = {@QueueBinding(value = @Queue,exchange = @Exchange(value = "direstTest",type = ExchangeTypes.DIRECT),key = {"error"})})
    public void receivel1(String message){
    System.out.println("message1 = " + message);
    }
}

生产者

@Autowired
private RabbitTemplate rabbitTemplate;

// 路由模型
@Test
public void direstExchangeTest(){
    rabbitTemplate.convertAndSend("direstTest","info","发送info的key的路由消息");
}

// 路由模型
@Test
public void direstExchangeTest1(){
    rabbitTemplate.convertAndSend("direstTest","error","发送error的key的路由消息");
}

3.5 Topics主题模型

topicExchange与directExchange类型，区别在于routingKey必须是多个单词的列表，并且以 . 分隔：

*（代表通配符，任意一个字段）

#（号代表一个或多个字段)

消费者

@Componentpublic class TopicsExchangeListener {
    // 不指定队列，消息过了就没了
    //  @RabbitListener(bindings = {@QueueBinding(value = @Queue,exchange = @Exchange(name = "topicList",type = ExchangeTypes.TOPIC),key = {"user.save","user."})})
    // 指定队列，可以接收缓存到队列里的消息
    // key = {"user.save","user.*"} 表示能消费 routingkey为  user.save 和 user.任意一个字符  的消息
    @RabbitListener(bindings = {@QueueBinding(value = @Queue(value ="test2",durable = "true" ),exchange = @Exchange(name = "topicList",type = ExchangeTypes.TOPIC),key = {"user.save","user.*"})})
    public void recevicel(String message){
        System.out.println("message = " + message);
    }

    // key = {"order.#","user.*"} 表示能消费 routingkey为  order.一个或多个字符   和  user.任意一个字符  的消息
    @RabbitListener(bindings = {@QueueBinding(value = @Queue,exchange = @Exchange(name = "topicList",type = ExchangeTypes.TOPIC),key = {"order.#","user.*"})})
    public void recevicel1(String message){
        System.out.println("message1 = " + message);
    }
}

生产者

@Autowired
private RabbitTemplate rabbitTemplate;

@Test
public void topicTest(){
    rabbitTemplate.convertAndSend("topicTest","user.save","topic路由消息,use.save");
}

@Test
public void topicTest1(){
    rabbitTemplate.convertAndSend("topicTest","order.select.getone","topic路由消息,order.select.getone");
}

3.6 消息转换器

代码里直接发送对象，虽然接收的到消息，但是rabbitmq的界面上看到的消息会是乱码。

依赖

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

配置

@Configuration
public class rabbitmqConfig {
    // 消息转换配置
    @Bean
    public MessageConverter jsonMessageConverter(){
        return new Jackson2JsonMessageConverter();
    }
}

再次发送就会是转换好的消息。