SpringCloud：初识RabbitMQ及快速入门

逃离我推掉我的手 2024-03-25 20:53 70阅读 0赞

## 1.初识`MQ` ##

### 1.1.同步和异步通讯 ###

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

同步通讯：就像打电话，需要实时响应。

异步通讯：就像发邮件，不需要马上回复。

![在这里插入图片描述][0e1012d8278649a7b654153a9f4e14e7.png_pic_center]

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

#### 1.1.1.同步通讯 ####

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

![在这里插入图片描述][4475872396c14fa2ac12fc491742db1b.png_pic_center]

总结：

同步调用的优点：

*  时效性较强，可以立即得到结果

同步调用的问题：

*  耦合度高
 *  性能和吞吐能力下降
 *  有额外的资源消耗
 *  有级联失败问题

#### 1.1.2.异步通讯 ####

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

我们以购买商品为例，用户支付后需要调用订单服务完成订单状态修改，调用物流服务，从仓库分配响应的库存并准备发货。

在事件模式中，支付服务是事件发布者（`publisher`），在支付完成后只需要发布一个支付成功的事件（`event`），事件中带上订单`id`。

订单服务和物流服务是事件订阅者（`Consumer`），订阅支付成功的事件，监听到事件后完成自己业务即可。

为了解除事件发布者与订阅者之间的耦合，两者并不是直接通信，而是有一个中间人（`Broker`）。发布者发布事件到`Broker`，不关心谁来订阅事件。订阅者从`Broker`订阅事件，不关心谁发来的消息。

![在这里插入图片描述][494398bb1ad0448cacf5733fb5aec7fd.png_pic_center]

`Broker`是一个像数据总线一样的东西，所有的服务要接收数据和发送数据都发到这个总线上，这个总线就像协议一样，让服务间的通讯变得标准和可控。

好处：

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

缺点：

*  架构复杂了，业务没有明显的流程线，不好管理
 *  需要依赖于`Broker`的可靠、安全、性能

好在现在开源软件或云平台上`Broker`的软件是非常成熟的，比较常见的一种就是`MQ`技术。

### 1.2.技术对比： ###

`MQ`，中文是消息队列（`MessageQueue`），字面来看就是存放消息的队列。也就是事件驱动架构中的`Broker`。

比较常见的`MQ`实现：

*  `ActiveMQ`
 *  `RabbitMQ`
 *  `RocketMQ`
 *  `Kafka`

几种常见`MQ`的对比：

<table> 
 <thead> 
  <tr> 
   <th></th> 
   <th><code>RabbitMQ</code></th> 
   <th><code>ActiveMQ</code></th> 
   <th><code>RocketMQ</code></th> 
   <th><code>Kafka</code></th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>公司/社区</td> 
   <td><code>Rabbit</code></td> 
   <td><code>Apache</code></td> 
   <td>阿里</td> 
   <td><code>Apache</code></td> 
  </tr> 
  <tr> 
   <td>开发语言</td> 
   <td><code>Erlang</code></td> 
   <td><code>Java</code></td> 
   <td><code>Java</code></td> 
   <td><code>Scala&amp;Java</code></td> 
  </tr> 
  <tr> 
   <td>协议支持</td> 
   <td><code>AMQP</code>，<code>XMPP</code>，<code>SMTP</code>，<code>STOMP</code></td> 
   <td><code>OpenWire</code>，<code>STOMP</code>，<code>REST</code>，<code>XMPP</code>，<code>AMQP</code></td> 
   <td>自定义协议</td> 
   <td>自定义协议</td> 
  </tr> 
  <tr> 
   <td>可用性</td> 
   <td>高</td> 
   <td>一般</td> 
   <td>高</td> 
   <td>高</td> 
  </tr> 
  <tr> 
   <td>单机吞吐量</td> 
   <td>一般</td> 
   <td>差</td> 
   <td>高</td> 
   <td>非常高</td> 
  </tr> 
  <tr> 
   <td>消息延迟</td> 
   <td>微秒级</td> 
   <td>毫秒级</td> 
   <td>毫秒级</td> 
   <td>毫秒以内</td> 
  </tr> 
  <tr> 
   <td>消息可靠性</td> 
   <td>高</td> 
   <td>一般</td> 
   <td>高</td> 
   <td>一般</td> 
  </tr> 
 </tbody> 
</table>

追求可用性：`Kafka`、 `RocketMQ` 、`RabbitMQ`

追求可靠性：`RabbitMQ`、`RocketMQ`

追求吞吐能力：`RocketMQ`、`Kafka`

追求消息低延迟：`RabbitMQ`、`Kafka`

## 2.快速入门 ##

### 2.1.安装`RabbitMQ` ###

安装`RabbitMQ`，`Docker`在线拉取：

docker pull rabbitmq:3-management

执行下面的命令来运行`MQ`容器：

docker run \
     -e RABBITMQ_DEFAULT_USER=dcxuexi \
     -e RABBITMQ_DEFAULT_PASS=dcxuexi1234 \
     --name mq \
     --hostname mq1 \
     -p 15672:15672 \
     -p 5672:5672 \
     -d \
     rabbitmq:3-management

`MQ`的基本结构：

![在这里插入图片描述][f7b886714fcb45e089f9444b4044f429.png_pic_center]

`RabbitMQ`中的一些角色：

*  `publisher`：生产者
 *  `consumer`：消费者
 *  `exchange`：交换机，负责消息路由
 *  `queue`：队列，存储消息
 *  `virtualHost`：虚拟主机，隔离不同租户的`exchange`、`queue`、消息的隔离

### 2.2.`RabbitMQ`消息模型 ###

`RabbitMQ`官方提供了`5`个不同的`Demo`示例，对应了不同的消息模型：

![在这里插入图片描述][0488568d08a14922bdf04e98d71c7b88.png_pic_center]

### 2.3.创建`mq-demo`工程 ###

创建后可以看到结构如下：

![在这里插入图片描述][96b1911ee80a4ebc9f099cb0341ca426.png_pic_center]

包括三部分：

*  `mq-demo`：父工程，管理项目依赖
 *  `publisher`：消息的发送者
 *  `consumer`：消息的消费者

### 2.4.入门案例 ###

简单队列模式的模型图：

![在这里插入图片描述][577fd94e9e4b45da8e68500bfc3ac043.png_pic_center]

官方的`HelloWorld`是基于最基础的消息队列模型来实现的，只包括三个角色：

*  `publisher`：消息发布者，将消息发送到队列`queue`
 *  `queue`：消息队列，负责接受并缓存消息
 *  `consumer`：订阅队列，处理队列中的消息

#### 2.4.1.`publisher`实现 ####

思路：

*  建立连接
 *  创建`Channel`
 *  声明队列
 *  发送消息
 *  关闭连接和`channel`

代码实现：

package com.dcxuexi.mq.helloworld;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import org.junit.Test;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class PublisherTest {
        
        @Test
        public void testSendMessage() throws IOException, TimeoutException {
        
            // 1.建立连接
            ConnectionFactory factory = new ConnectionFactory();
            // 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
            factory.setHost("124.222.106.122");
            factory.setPort(5672);
            factory.setVirtualHost("/");
            factory.setUsername("dcxuexi");
            factory.setPassword("123456");
            // 1.2.建立连接
            Connection connection = factory.newConnection();
    
            // 2.创建通道Channel
            Channel channel = connection.createChannel();
    
            // 3.创建队列
            String queueName = "simple.queue";
            channel.queueDeclare(queueName, false, false, false, null);
    
            // 4.发送消息
            String message = "hello, rabbitmq!";
            channel.basicPublish("", queueName, null, message.getBytes());
            System.out.println("发送消息成功：【" + message + "】");
    
            // 5.关闭通道和连接
            channel.close();
            connection.close();
    
        }
    }

#### 2.4.2.`consumer`实现 ####

代码思路：

*  建立连接
 *  创建`Channel`
 *  声明队列
 *  订阅消息

代码实现：

package com.dcxuexi.mq.helloworld;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class ConsumerTest {
        
    
        public static void main(String[] args) throws IOException, TimeoutException {
        
            // 1.建立连接
            ConnectionFactory factory = new ConnectionFactory();
            // 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
            factory.setHost("124.222.106.122");
            factory.setPort(5672);
            factory.setVirtualHost("/");
            factory.setUsername("dcxuexi");
            factory.setPassword("123456");
            // 1.2.建立连接
            Connection connection = factory.newConnection();
    
            // 2.创建通道Channel
            Channel channel = connection.createChannel();
    
            // 3.创建队列
            String queueName = "simple.queue";
            channel.queueDeclare(queueName, false, false, false, null);
    
            // 4.订阅消息
            channel.basicConsume(queueName, true, new DefaultConsumer(channel){
        
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope,
                                           AMQP.BasicProperties properties, byte[] body) throws IOException {
        
                    // 5.处理消息
                    String message = new String(body);
                    System.out.println("接收到消息：【" + message + "】");
                }
            });
            System.out.println("等待接收消息。。。。");
        }
    }

### 2.5.总结 ###

基本消息队列的消息发送流程：

1.  建立`connection`
2.  创建`channel`
3.  利用`channel`声明队列
4.  利用`channel`向队列发送消息

基本消息队列的消息接收流程：

1.  建立`connection`
2.  创建`channel`
3.  利用`channel`声明队列
4.  定义`consumer`的消费行为`handleDelivery()`
5.  利用`channel`将消费者与队列绑定

[0e1012d8278649a7b654153a9f4e14e7.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/b9fbec48f2be4d85ba730b401397fabf.png
[4475872396c14fa2ac12fc491742db1b.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/f3fb87483fc946b4b46d0fbd7f98685d.png
[494398bb1ad0448cacf5733fb5aec7fd.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/79c57fddd00f4ab9b9923c3c53644d8b.png
[f7b886714fcb45e089f9444b4044f429.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/133967f77a26434c880a9f28e8d347a9.png
[0488568d08a14922bdf04e98d71c7b88.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/1f1f7a7d5cc84eb8afb4b4d376df7f15.png
[96b1911ee80a4ebc9f099cb0341ca426.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/101d01243fd746e7b5cada043277e024.png
[577fd94e9e4b45da8e68500bfc3ac043.png_pic_center]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/25/80d8ec36a5724234afc292521ad7d1c1.png