RabbitMQ入门之基础理论

今天药忘吃喽~ 2022-12-27 09:07 75阅读 0赞

### RabbitMQ基础理论 ###

*  RabbitMQ入门篇
 *  前言
 *   *  什么是MQ？
     *  常见的MQ
     *   *  ActiveMQ
         *  Kafka
         *  RocketMQ
         *  RabbitMQ
 *  RabbitMQ
 *   *  概念
     *   *  什么是AMQP？
         *  AMQP协议模型
     *  安装与配置
     *   *  安装
         *  配置
     *  RabbitMQ配置
     *   *  RabbitMQ命令行管理
         *  Web管理界面
         *   *  Admin用户和虚拟主机管理
             *   *  创建用户
                 *  创建虚拟主机
                 *  绑定虚拟主机和用户
     *  RabbitMQ支持的几种消息模型
     *   *  消息模型概览
         *  Hello World模型
         *   *  概念
             *  程序实现
             *  细节优化
         *  Work quenes模型
         *   *  概念
             *  程序实现
         *  fanout模型
         *   *  概念
             *  程序实现
         *  Routing模型
         *   *  Routing 之订阅模型-Direct(直连)
             *   *  概念
                 *  程序实现
             *  Routing 之订阅模型-Topics
             *   *  概念
                 *  程序实现

# RabbitMQ入门篇 #

按照惯例，先上[官网][Link 1]：[https://www.rabbitmq.com/][Link 1]

[官方教程][Link 2]：[https://www.rabbitmq.com/\#getstarted][Link 2]

# 前言 #

## 什么是MQ？ ##

\*\*MQ(Message Quene) \*\*: 翻译为`消息队列`,通过典型的`生产者`和`消费者`模型,生产者不断向消息队列中生产消息，消费者不断的从队列中获取消息。因为消息的生产和消费都是异步的，而且只关心消息的发送和接收，没有业务逻辑的侵入，轻松的实现系统间解耦。别名为`消息中间件`通过利用高效可靠的消息传递机制进行平台无关的数据交流，并基于数据通信来进行分布式系统的集成

## 常见的MQ ##

当今市面上有很多主流的消息中间件，如老牌的`ActiveMQ`、`RabbitMQ`，炙手可热的`Kafka`，阿里巴巴自主开发`RocketMQ`等

### ActiveMQ ###

`ActiveMQ`是Apache出品，最流行、能力强劲的开源消息总线。它是一个完全支持JMS规范的的消息中间件。丰富的API，多种集群架构模式让ActiveMQ在业界成为老牌的消息中间件，在中小型企业颇受欢迎!

### Kafka ###

`Kafka`是LinkedIn开源的分布式发布-订阅消息系统，目前归属于Apache顶级项目。Kafka主要特点是基于Pull的模式来处理消息消费，追求高吞吐量，一开始的目的就是用于日志收集和传输。0.8版本开始支持复制，不支持事务，对消息的重复、丢失、错误没有严格要求，适合产生大量数据的互联网服务的数据收集业务。

### RocketMQ ###

`RocketMQ`是阿里开源的消息中间件，它是纯Java开发，具有高吞吐量、高可用性、适合大规模分布式系统应用的特点。RocketMQ思路起源于Kafka，但并不是Kafka的一个Copy，它对消息的可靠传输及事务性做了优化，目前在阿里集团被广泛应用于交易、充值、流计算、消息推送、日志流式处理、binglog分发等场景

### RabbitMQ ###

`RabbitMQ`是使用Erlang语言开发的开源消息队列系统，基于AMQP协议来实现。AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。AMQP协议更多用在企业系统内对数据一致性、稳定性和可靠性要求很高的场景，对性能和吞吐量的要求还在其次

> RabbitMQ比Kafka可靠，Kafka更适合IO高吞吐的处理，一般应用在大数据日志处理或对实时性（少量延迟），可靠性（少量丢数据）要求稍低的场景使用，比如ELK日志收集

# RabbitMQ #

## 概念 ##

**RabbitMQ**：基于`AMQP`协议，erlang语言开发，是部署最广泛的开源消息中间件,是最受欢迎的开源消息中间件之一。这里说一下`AMQP`协议

### 什么是AMQP？ ###

### AMQP协议模型 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center]

## 安装与配置 ##

安装之前，我们需要按照之前的文章[Linux–04、虚拟机的克隆与配置][Linux_04]克隆出一台新的虚拟机，修改主机名为`Christy020(RabbitMQ)`， 修改ip地址为192.168.10.200

### 安装 ###

1.  下载安装包
    
    `官网下载地址`: https://www.rabbitmq.com/download.html
    
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 1]
    
    上面说了RabbitMQ是用erlang语言开发的，需要安装erlang的依赖，所以需要下面两个安装包
    
    ![在这里插入图片描述][2020120923474669.png_pic_center]
    
    我们的Linux操作系统时CentOS7.6，不要下载错了
2.  上传安装包到linux
    
    进入到`/usr/tools`目录，上传安装包
    
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 2]
3.  安装Erlang依赖包
    
    `rpm -ivh erlang-22.0.7-1.el7.x86_64.rpm`
    
    ![在这里插入图片描述][20201209234822456.png_pic_center]
4.  安装RabbitMQ安装包
    
    `yum install -y rabbitmq-server-3.7.18-1.el7.noarch.rpm`，安装RabbitMQ的过程中是需要联网  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 3]

### 配置 ###

上述安装包安装完成后，接下来进入到配置环节

1.  复制配置文件
    
    RabbitMQ安装完成后配置文件模板默认生成位置在`/usr/share/doc/rabbitmq-server-3.7.18/`下，名称为`rabbitmq.config.example`，执行复制命令`cp /usr/share/doc/rabbitmq-server-3.7.18/rabbitmq.config.example /etc/rabbitmq/rabbitmq.config`拷贝配置文件到`/etc/rabbitmq/`下
    
    ![在这里插入图片描述][20201209234915861.png_pic_center]
2.  修改配置文件
    
    执行命令`vim /etc/rabbitmq/rabbitmq.config`，修改配置文件如下
    
    这个主要是打开来宾账户，否则小无法进入后面说的后台管理界面
    
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 4]
3.  启动RabbitMQ的插件管理功能
    
    执行以下命令`rabbitmq-plugins enable rabbitmq_management`，执行结果如下图则说明插件管理服务启动成功
    
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 5]
    
    这里启动插件服务的时候报了一个错误`/usr/lib/rabbitmq/bin/rabbitmq-env: eval:行180: 未预期的符号`(’ 附近有语法错误\`，如下图

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

我们`cd`到这个目录，找到这个文件执行`vim`命令，`:set nu`显示出行号后找到180行

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 6]

之前是因为`path`变量前后的引号和$没有转义造成的，上图是修改过的，保存退出重新执行启动插件管理服务就行了

1.  启动RabbitMQ服务
    
    执行命令`systemctl start rabbitmq-server`启动RabbitMQ服务
    
        ## 常用的启动关停RabbitMQ服务的命令
        systemctl restart rabbitmq-server
        systemctl stop rabbitmq-server
2.  查看服务状态
    
    执行命令`systemctl status rabbitmq-server`，如果出现下图则说明RabbitMQ服务启动成功
    
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 7]
3.  服务启动成功后我们就可以访问RabbitMQ的Web管理界面了
    
    这里访问的地址是`http://192.168.10.200:15672`，RabbitMQ默认的端口号是**15672**
    
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 8]
    
    RabbitMQ首次启动，我们需要以来兵账户登录，默认的在登录账号和密码：`guest/guest`
    
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 9]
    
    至于页面上的选项和标签都是什么意思，下面我们来详细介绍一下

## RabbitMQ配置 ##

这里的配置区别于上面安装时的配置，这里的配置指的是安装成功以后为了方便使用或者提升性能的配置。RabbitMQ的配置有两种方式：其一是命令行的配置方式；其二是Web管理界面的配置方式

### RabbitMQ命令行管理 ###

### Web管理界面 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 10]

*  Overview：概览，当前使用的RabbitMQ的大致数据
 *  Connections：无论生产者还是消费者，都需要与RabbitMQ建立连接后才可以完成消息的生产和消费，在这里可以查看连接情况
 *  Channels：通道，建立连接后，会形成通道，消息的投递获取依赖通道
 *  Exchanges：交换机，用来实现消息的路由
 *  Queues：队列，即消息队列，消息存放在队列中，等待消费，消费后被移除队列
 *  Admin：用户管理，这里可以管理用户，设置用户权限，新建|绑定虚拟主机等一系列有关用户的操作

#### Admin用户和虚拟主机管理 ####

##### 创建用户 #####

点开上面的`Admin`选项卡，如下图

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 11]

上面的Tags选项，其实是指定用户的角色，可选的有以下几个：

*  `超级管理员(administrator)`
    
    可登陆管理控制台，可查看所有的信息，并且可以对用户，策略(policy)进行操作
 *  `监控者(monitoring)`
    
    可登陆管理控制台，同时可以查看rabbitmq节点的相关信息(进程数，内存使用情况，磁盘使用情况等)
 *  `策略制定者(policymaker)`
    
    可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红框标识的部分)
 *  `普通管理者(management)`
    
    仅可登陆管理控制台，无法看到节点信息，也无法对策略进行管理
 *  `其他`
    
    无法登陆管理控制台，通常就是普通的生产者和消费者

##### 创建虚拟主机 #####

什么是虚拟主机？

为了让各个用户可以互不干扰的工作，RabbitMQ添加了虚拟主机（Virtual Hosts）的概念。其实就是一个独立的访问路径，不同用户使用不同路径，各自有自己的队列、交换机，互相不会影响

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 12]

##### 绑定虚拟主机和用户 #####

点击添加好的虚拟主机，上面是`/tide`，进入虚拟机设置界面按下图设置

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 13]

设置完后返回到user里面可以看到创建的`christy`用户有访问虚拟主机`/tide`的权限了

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 14]

## RabbitMQ支持的几种消息模型 ##

### 消息模型概览 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 15]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 16]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 17]

### Hello World模型 ###

#### 概念 ####

对应上述七种模型中的第一种

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

在该模型中

*  P：生产者，也就是要发送消息的程序
 *  C：消费者：消息的接受者，会一直等待消息到来
 *  queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。

#### 程序实现 ####

1.  新建一个普通的maven工程，`pom.xml`中引入rabbitmq的客户端依赖
    
        <dependencies>
            <dependency>
                <groupId>junit</groupId>
                <artifactId>junit</artifactId>
                <version>4.11</version>
            </dependency>
        
            <dependency>
                <groupId>com.rabbitmq</groupId>
                <artifactId>amqp-client</artifactId>
                <version>5.7.2</version>
            </dependency>
        </dependencies>
2.  生产者–Provider.java
    
        package com.christy.rabbitmq.hello;
        
        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 Provider { 
            /** 生产消息 **/
            @Test
            public void testSendMessage() throws IOException, TimeoutException { 
                // 创建连接mq的工厂对象
                ConnectionFactory connectionFactory = new ConnectionFactory();
                // 设置连接RabbitMQ的主机
                connectionFactory.setHost("192.168.10.200");
                // 设置端口号
                connectionFactory.setPort(5672);
                // 设置连接哪个虚拟主机
                connectionFactory.setVirtualHost("/tide");
                // 设置访问虚拟主机的用户名和密码
                connectionFactory.setUsername("christy");
                connectionFactory.setPassword("123456");
        
                // 获取连接对象
                Connection connection = connectionFactory.newConnection();
                // 获取连接通道
                Channel channel = connection.createChannel();
        
                // 绑定消息队列 这里要与生产者的参数保持一致，否则会出现问题
                /** 参数一：队列名称，如果队列不存在则创建 * 参数二：用来定义是否持久化队列 true-持久化 false-不持久化 * 参数三：exclusive 当前通道是否独占该队列 true-独占 false-不独占 * 参数四：autoDelete 是否在消息消费完成后（消费者解除与生产者的连接，即通道关闭并且连接关闭）自动删除队列 true-自动删除 false-不删除 * 参数五：额外的附加参数 **/
                channel.queueDeclare("Dear Tide",false,false,false,null);
        
                // 发布消息
                // 参数一：交换机名称
                // 参数二：队列名称
                // 参数三：传递消息额外设置，比如设置消息持久化MessageProperties.PERSISTENT_TEXT_PLAIN
                // 参数四：消息的具体内容
                channel.basicPublish("","Dear Tide",null,"Tide, Christy love you!".getBytes());
        
                channel.close();
                connection.close();
            }
        }
3.  测试运行
    
    程序运行成功后，我们刷新Web管理界面，查看`Queues`选项卡
    
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 18]
    
    可以看到队列`/tide`中有一条准备好待消费的消息
4.  消费者–Customer.java
    
    上面生产者已经可以成功生产消息放入队列了，下面我们来编写消费者的代码来实成功消费队列中的消息
    
        package com.christy.rabbitmq.hello;
        
        import com.rabbitmq.client.*;
        
        import java.io.IOException;
        import java.util.concurrent.TimeoutException;
        
        /** * 消息消费者 */
        public class Customer { 
            public static void main(String[] args) throws IOException, TimeoutException { 
                /** 消费者要消费生产者产生的消息就必须跟生产者的主机|虚拟主机配置相同 **/
                // 创建连接mq的工厂对象
                ConnectionFactory connectionFactory = new ConnectionFactory();
                // 连接mq的主机
                connectionFactory.setHost("192.168.10.200");
                // 设置端口
                connectionFactory.setPort(5672);
                // 设置虚拟主机
                connectionFactory.setVirtualHost("/tide");
                // 设置虚拟主机的用户名和密码
                connectionFactory.setUsername("christy");
                connectionFactory.setPassword("123456");
        
                // 获取连接对象
                Connection connection = connectionFactory.newConnection();
                // 获取连接通道
                Channel channel = connection.createChannel();
        
                // 绑定消息队列 这里要与生产者的参数保持一致，否则会出现问题
                /** 参数一：队列名称，如果队列不存在则创建 * 参数二：用来定义是否持久化队列 true-持久化 false-不持久化 * 参数三：exclusive 当前通道是否独占该队列 true-独占 false-不独占 * 参数四：autoDelete 是否在消息消费完成后（消费者解除与生产者的连接，即通道关闭并且连接关闭）自动删除队列 true-自动删除 false-不删除 * 参数五：额外的附加参数 **/
                channel.queueDeclare("Dear Tide",false,false,false,null);
        
                // 消费消息
                channel.basicConsume("Dear Tide",true,new DefaultConsumer(channel){ 
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                        System.out.println("================消费消息=============="+ new String(body));
                    }
                });
        
                /** 关闭通道和连接的话只会消费一次就结束，通过情况下消费是一直在监听生产者是否有消息需要消费，故在此将其注释掉 **/
                /*channel.close(); connection.close();*/
            }
        }
5.  测试
    
    ![在这里插入图片描述][20201209235935364.gif_pic_center]

#### 细节优化 ####

针对上述生产者和消费者的代码对比可以发现创建连接`Connection`的代码部分存在冗余，这里我们来优化一下

package com.christy.rabbitmq.utils;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    
    public class RabbitMQUtil { 
        
        private static ConnectionFactory connectionFactory;
        static { 
            connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("192.168.10.200");
            connectionFactory.setPort(5672);
            connectionFactory.setVirtualHost("/tide");
            connectionFactory.setUsername("christy");
            connectionFactory.setPassword("123456");
        }
        
        public static Connection getConnection(){ 
            try { 
                return connectionFactory.newConnection();
            } catch (Exception e) { 
                e.printStackTrace();
            }
            
            return null;
        }
        
        public static void closeConnectionAndChannel(Channel channel, Connection conn){ 
            try { 
                if (null != channel) channel.close();
                if (null != conn) conn.close();
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }
    }

### Work quenes模型 ###

#### 概念 ####

**Work queues**也被称之为`任务队列(Task queues)`。对应上述七种模型的第二种；

上面我们讲的`Hello World`模型，一旦遇到消息处理比较耗时的情况，这时候生产消息速度会远远大于消息的消费速度；随着时间流逝队列中的消息就会堆积的越来越多；这时候就需要`Work queues`消息模型：**让多个消费者绑定到一个队列，共同消费队列中的消息**。队列中的消息一旦被消费就会从队列中删除，因此任务是不会被重复执行的

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 19]

#### 程序实现 ####

1.  生产者–provider.java
    
        package com.christy.rabbitmq.work;
        
        import com.christy.rabbitmq.utils.RabbitMQUtil;
        import com.rabbitmq.client.Channel;
        import com.rabbitmq.client.Connection;
        import com.rabbitmq.client.MessageProperties;
        import org.junit.Test;
        
        import java.io.IOException;
        
        public class Provider { 
            @Test
            public void testProvider() throws IOException { 
                // 创建连接
                Connection connection = RabbitMQUtil.getConnection();
                // 创建通道
                Channel channel = connection.createChannel();
                // 绑定队列
                channel.queueDeclare("work queues",true,false,false,null);
        
                /** 由于是work模型，需要多个消费者消费多条消息，我们这里用for模拟生产者生产的多条消息 **/
                for (int i = 1; i <= 20; i++){ 
                    channel.basicPublish("","work queues", MessageProperties.PERSISTENT_TEXT_PLAIN,(i+ "---->Hello, Work queues!").getBytes());
                }
        
                RabbitMQUtil.closeConnectionAndChannel(channel,connection);
            }
        }
2.  消费者–Customer01.Java&Customer02.java
    
        package com.christy.rabbitmq.work;
        
        import com.christy.rabbitmq.utils.RabbitMQUtil;
        import com.rabbitmq.client.*;
        
        import java.io.IOException;
        import java.sql.SQLOutput;
        
        public class Customer01 { 
            public static void main(String[] args) throws IOException { 
                // 创建连接
                Connection connection = RabbitMQUtil.getConnection();
                // 创建通道
                Channel channel = connection.createChannel();
                // 绑定消息队列
                channel.queueDeclare("Work queues",true,false,false,null);
                // 消费消息
                channel.basicConsume("Work queues",true,new DefaultConsumer(channel){ 
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                        System.out.println("消费者1："+ new String(body));
                    }
                });
            }
        }
    
        package com.christy.rabbitmq.work;
        
        import com.christy.rabbitmq.utils.RabbitMQUtil;
        import com.rabbitmq.client.*;
        
        import java.io.IOException;
        
        public class Customer02 { 
            public static void main(String[] args) throws IOException { 
                // 创建连接
                Connection connection = RabbitMQUtil.getConnection();
                // 创建通道
                Channel channel = connection.createChannel();
                // 绑定消息队列
                channel.queueDeclare("Work queues",true,false,false,null);
                // 消费消息
                channel.basicConsume("Work queues",true,new DefaultConsumer(channel){ 
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                        System.out.println("消费者2："+ new String(body));
                    }
                });
            }
        }
3.  测试
    
    ![在这里插入图片描述][20201210000035993.gif_pic_center]
    
    以上测试可以看出`默认情况下，RabbitMQ将按顺序将每个消息发送给下一个使用者。平均而言，每个消费者都会收到相同数量的消息。这种分发消息的方式称为循环`
    
    **那么问题来了**
    
    问：上述代码有什么问题？怎么解决？如果我的消费者01消费消息的速度较消费者02来讲要慢，我希望的是消费者01消费的少一些，消费者02消费的多一些，该怎么办？
    
    答：上述代码实现是有问题的。原因：上述代码由于是默认情况下的实现并且消费时设置了`autoACK`为true即自动确认，也就是说队列中的消息会按照`循环`的方式将消息一次性的平均分配给他的消费者，将消息保存到每个消费者的channel中，一旦消息到达channel，不管消费者有没有消费都会自动发送确认给队列，队列中就会标记该消息已经被消费且不再保存该消息。此时一旦某个消费者宕机，那么该消费者未消费的消息就会丢失。
    
    解决方案：想要解决上面的问题必须保证以下两步
    
     *  设置通道一次只能消费一个消息
     *  关闭消息消费时的自动确认，开启手动确认
    
    实现以上两步既能保证消息不丢失，还能实现后面能者多劳的问题
4.  代码优化
    
    为了实现上面的`能者多劳`，我们让**Customer01**每次消费消息之前睡眠两秒，**Cusomer02**正常
    
        package com.christy.rabbitmq.work;
        
        import com.christy.rabbitmq.utils.RabbitMQUtil;
        import com.rabbitmq.client.*;
        
        import java.io.IOException;
        
        public class Customer01 { 
            public static void main(String[] args) throws IOException { 
                // 创建连接
                Connection connection = RabbitMQUtil.getConnection();
                // 创建通道
                final Channel channel = connection.createChannel();
                // 绑定消息队列
                channel.queueDeclare("Work queues", true, false, false, null);
                /** 设置通道一次只接受一条未确认的消息 **/
                channel.basicQos(1);
                // 消费消息 设置autoAck为false
                channel.basicConsume("Work queues", false, new DefaultConsumer(channel) { 
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                        try { 
                            Thread.sleep(2000);
                        } catch (InterruptedException e) { 
                            e.printStackTrace();
                        }
                        System.out.println("消费者1：" + new String(body));
                        /** 开启手动确认 * 参数1：消息的标识 * 参数2：是否一次确认多条消息 true-是 false-否 **/
                        channel.basicAck(envelope.getDeliveryTag(),false);
                    }
                });
            }
        }
    
        package com.christy.rabbitmq.work;
        
        import com.christy.rabbitmq.utils.RabbitMQUtil;
        import com.rabbitmq.client.*;
        
        import java.io.IOException;
        
        public class Customer02 { 
            public static void main(String[] args) throws IOException { 
                // 创建连接
                Connection connection = RabbitMQUtil.getConnection();
                // 创建通道
                final Channel channel = connection.createChannel();
                // 绑定消息队列
                channel.queueDeclare("Work queues",true,false,false,null);
                /** 设置通道一次只接受一条未确认的消息 **/
                channel.basicQos(1);
                // 消费消息 设置autoAck为false
                channel.basicConsume("Work queues",false,new DefaultConsumer(channel){ 
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                        System.out.println("消费者2："+ new String(body));
                        /** 开启手动确认 * 参数1：消息的标识 * 参数2：是否一次确认多条消息 true-是 false-否 **/
                        channel.basicAck(envelope.getDeliveryTag(),false);
                    }
                });
            }
        }
5.  测试
    
    ![在这里插入图片描述][20201210000116471.gif_pic_center]
    
    测试可知该方案能解决上述提出的问题并且实现了能者多劳

### fanout模型 ###

#### 概念 ####

**fanout模型**也称之为`广播模型`，对应上面七种消息模型的第三种

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

在广播模式下，消息发送流程是这样的：

*  可以有多个消费者
 *  每个**消费者有自己的queue**（队列）
 *  每个**队列都要绑定到Exchange**（交换机）
 *  **生产者发送的消息，只能发送到交换机**，交换机来决定要发给哪个队列，生产者无法决定。
 *  交换机把消息发送给绑定过的所有队列
 *  队列的消费者都能拿到消息。实现一条消息被多个消费者消费

#### 程序实现 ####

1.  生产者–provider.java
    
        package com.christy.rabbitmq.fanout;
        
        import com.christy.rabbitmq.utils.RabbitMQUtil;
        import com.rabbitmq.client.Channel;
        import com.rabbitmq.client.Connection;
        
        import java.io.IOException;
        
        public class Provider { 
            public static void main(String[] args) throws IOException { 
                // 创建连接
                Connection connection = RabbitMQUtil.getConnection();
                Channel channel = connection.createChannel();
        
                /** 声明交换机 * 参数1：交换机名称 * 参数2：交换机类型 fanout 广播 */
                channel.exchangeDeclare("orders","fanout");
                
                /** 发布消息 * 参数一：交换机名称 * 参数二：队列名称 * 参数三：传递消息额外设置 * 参数四：消息的具体内容 **/
                channel.basicPublish("orders","",null,"fanout publish message".getBytes());
                
                RabbitMQUtil.closeConnectionAndChannel(channel,connection);
            }
        }
2.  消费者（多个）
    
        package com.christy.rabbitmq.fanout;
        
        import com.christy.rabbitmq.utils.RabbitMQUtil;
        import com.rabbitmq.client.*;
        
        import java.io.IOException;
        
        public class Customer01 { 
            public static void main(String[] args) throws IOException { 
                // 创建连接
                Connection connection = RabbitMQUtil.getConnection();
                // 创建通道
                Channel channel = connection.createChannel();
        
                // 通道绑定交换机
                channel.exchangeDeclare("orders","fanout");
                // 创建临时队列
                String queue = channel.queueDeclare().getQueue();
                // 临时队列绑定到交换机
                channel.queueBind(queue,"orders","");
                // 消费消息
                channel.basicConsume(queue,true,new DefaultConsumer(channel){ 
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                        System.out.println("消费者01："+ new String(body));
                    }
                });
            }
        }
    
        package com.christy.rabbitmq.fanout;
        
        import com.christy.rabbitmq.utils.RabbitMQUtil;
        import com.rabbitmq.client.*;
        
        import java.io.IOException;
        
        public class Customer02 { 
            public static void main(String[] args) throws IOException { 
                // 创建连接
                Connection connection = RabbitMQUtil.getConnection();
                // 创建通道
                Channel channel = connection.createChannel();
        
                // 通道绑定交换机
                channel.exchangeDeclare("orders","fanout");
                // 创建临时队列
                String queue = channel.queueDeclare().getQueue();
                // 临时队列绑定到交换机
                channel.queueBind(queue,"orders","");
                // 消费消息
                channel.basicConsume(queue,true,new DefaultConsumer(channel){ 
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                        System.out.println("消费者02："+ new String(body));
                    }
                });
            }
        }
    
        package com.christy.rabbitmq.fanout;
        
        import com.christy.rabbitmq.utils.RabbitMQUtil;
        import com.rabbitmq.client.*;
        
        import java.io.IOException;
        
        public class Customer03 { 
            public static void main(String[] args) throws IOException { 
                // 创建连接
                Connection connection = RabbitMQUtil.getConnection();
                // 创建通道
                Channel channel = connection.createChannel();
        
                // 通道绑定交换机
                channel.exchangeDeclare("orders","fanout");
                // 创建临时队列
                String queue = channel.queueDeclare().getQueue();
                // 临时队列绑定到交换机
                channel.queueBind(queue,"orders","");
                // 消费消息
                channel.basicConsume(queue,true,new DefaultConsumer(channel){ 
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                        System.out.println("消费者03："+ new String(body));
                    }
                });
            }
        }
3.  测试
    
    ![在这里插入图片描述][20201210000201900.gif_pic_center]
    
    上面经过测试可以发现三个消费者都消费到了消息

### Routing模型 ###

#### Routing 之订阅模型-Direct(直连) ####

##### 概念 #####

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

在Direct模型下：

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

流程:

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 20]

图解：

*  P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。
 *  X：Exchange（交换机），接收生产者的消息，然后把消息递交给 与routing key完全匹配的队列
 *  C1：消费者，其所在队列指定了需要routing key 为 error 的消息
 *  C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息

##### 程序实现 #####

1.  生产者
    
        package com.christy.rabbitmq.direct;
        
        import com.christy.rabbitmq.utils.RabbitMQUtil;
        import com.rabbitmq.client.Channel;
        import com.rabbitmq.client.Connection;
        
        import java.io.IOException;
        
        public class Provider { 
            public static void main(String[] args) throws IOException { 
                // 创建连接
                Connection connection = RabbitMQUtil.getConnection();
                // 创建通道
                Channel channel = connection.createChannel();
        
                String exchangeName = "logs_direct";
                /** 绑定交换机 * 参数1：交换机名称 * 参数2：交换机类型 direct 基于指令的Routing key转发这里不能设置成fanout，否则routing key将没有效果 */
                channel.exchangeDeclare(exchangeName,"direct");
                // 生产消息
                String routingKey = "info";
                channel.basicPublish(exchangeName,routingKey,null,("direct模型基于routing key["+routingKey+"]生产的消息").getBytes());
        
                RabbitMQUtil.closeConnectionAndChannel(channel,connection);
            }
        }
2.  消费者（多个）
    
        package com.christy.rabbitmq.direct;
        
        import com.christy.rabbitmq.utils.RabbitMQUtil;
        import com.rabbitmq.client.*;
        
        import java.io.IOException;
        
        public class Consumer01 { 
            public static void main(String[] args) throws IOException { 
                // 创建连接
                Connection connection = RabbitMQUtil.getConnection();
                // 创建通道
                Channel channel = connection.createChannel();
        
                String exchangeName = "logs_direct";
                // 通道绑定交换机
                channel.exchangeDeclare(exchangeName,"direct");
                // 创建临时队列
                String queue = channel.queueDeclare().getQueue();
                // 临时队列绑定交换机
                channel.queueBind(queue,exchangeName,"error");
                // 消费消息
                channel.basicConsume(queue,true,new DefaultConsumer(channel){ 
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                        System.out.println("消费者01：" + new String(body));
                    }
                });
            }
        }
    
        package com.christy.rabbitmq.direct;
        
        import com.christy.rabbitmq.utils.RabbitMQUtil;
        import com.rabbitmq.client.*;
        
        import java.io.IOException;
        
        public class Consumer02 { 
            public static void main(String[] args) throws IOException { 
                // 创建连接
                Connection connection = RabbitMQUtil.getConnection();
                // 创建通道
                Channel channel = connection.createChannel();
        
                String exchangeName = "logs_direct";
                // 通道绑定交换机
                channel.exchangeDeclare(exchangeName,"direct");
                // 创建临时队列
                String queue = channel.queueDeclare().getQueue();
                // 临时队列绑定交换机
                channel.queueBind(queue,exchangeName,"info");
                channel.queueBind(queue,exchangeName,"error");
                // 消费消息
                channel.basicConsume(queue,true,new DefaultConsumer(channel){ 
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                        System.out.println("消费者02：" + new String(body));
                    }
                });
            }
        }
3.  测试
    
    ![在这里插入图片描述][20201210000259182.gif_pic_center]
    
    经过测试我们可以看到**Customer01**只能消费`error`级别的消息，**Customer02**可以消费`info`与`error`级别的消息，这个结果符合我们的预期

#### Routing 之订阅模型-Topics ####

##### 概念 #####

`Topic`类型的`Exchange`与`Direct`相比，都是可以根据`RoutingKey`把消息路由到不同的队列。只不过`Topic`类型`Exchange`可以让队列在绑定`Routing key`的时候使用通配符！这种模型`Routingkey` 一般都是由一个或多个单词组成，多个单词之间以”.”分割，例如： `item.insert`

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 21]

# 通配符
    	* (star) can substitute for exactly one word.    匹配不多不少恰好1个词
    	# (hash) can substitute for zero or more words.  匹配一个或多个词
    # 如:
    	audit.#    匹配audit.irs.corporate或者 audit.irs 等
    	audit.*   只能匹配 audit.irs

##### 程序实现 #####

1.  生产者
    
        package com.christy.rabbitmq.topics;
        
        import com.christy.rabbitmq.utils.RabbitMQUtil;
        import com.rabbitmq.client.Channel;
        import com.rabbitmq.client.Connection;
        
        import java.io.IOException;
        
        public class Provider { 
            public static void main(String[] args) throws IOException { 
                // 创建连接
                Connection connection = RabbitMQUtil.getConnection();
                // 创建通道
                Channel channel = connection.createChannel();
        
                String exchangeName = "topics";
                /** 绑定交换机 * 参数1：交换机名称 * 参数2：交换机类型 topic 动态路由 */
                channel.exchangeDeclare(exchangeName,"topic");
                // 生产消息
                String routingKey = "admin.user";
                channel.basicPublish(exchangeName,routingKey,null,("topic模型基于routing key["+routingKey+"]生产的消息").getBytes());
        
                RabbitMQUtil.closeConnectionAndChannel(channel,connection);
            }
        }
2.  消费者（多个）
    
        package com.christy.rabbitmq.topics;
        
        import com.christy.rabbitmq.utils.RabbitMQUtil;
        import com.rabbitmq.client.*;
        
        import java.io.IOException;
        
        public class Consumer01 { 
            public static void main(String[] args) throws IOException { 
                // 创建连接
                Connection connection = RabbitMQUtil.getConnection();
                // 创建通道
                Channel channel = connection.createChannel();
        
                String exchangeName = "topics";
                // 通道绑定交换机
                channel.exchangeDeclare(exchangeName,"topic");
                // 创建临时队列
                String queue = channel.queueDeclare().getQueue();
                // 临时队列绑定交换机
                channel.queueBind(queue,exchangeName,"admin.*");
                // 消费消息
                channel.basicConsume(queue,true,new DefaultConsumer(channel){ 
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                        System.out.println("消费者01：" + new String(body));
                    }
                });
            }
        }
    
        package com.christy.rabbitmq.topics;
        
        import com.christy.rabbitmq.utils.RabbitMQUtil;
        import com.rabbitmq.client.*;
        
        import java.io.IOException;
        
        public class Consumer02 { 
            public static void main(String[] args) throws IOException { 
                // 创建连接
                Connection connection = RabbitMQUtil.getConnection();
                // 创建通道
                Channel channel = connection.createChannel();
        
                String exchangeName = "topics";
                // 通道绑定交换机
                channel.exchangeDeclare(exchangeName,"topic");
                // 创建临时队列
                String queue = channel.queueDeclare().getQueue();
                // 临时队列绑定交换机
                channel.queueBind(queue,exchangeName,"admin.#");
                // 消费消息
                channel.basicConsume(queue,true,new DefaultConsumer(channel){ 
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                        System.out.println("消费者02：" + new String(body));
                    }
                });
            }
        }
3.  测试
    
    ![在这里插入图片描述][20201210000349821.gif_pic_center]
    
    以上测试可以发现**Customer01**只能消费`routing key`为`admin.user`的消息，而**Customer02**则不同，这也符合我们先前的预期

[Link 1]: https://www.rabbitmq.com/
[Link 2]: https://www.rabbitmq.com/#getstarted
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/c98cb7735f054948a75b180da4840a56.png
[Linux_04]: https://blog.csdn.net/bbxylqf126com/article/details/105468650
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/40153cf7ef60432abf6d018df84fe42f.png
[2020120923474669.png_pic_center]: /images/20221120/5e139f06de514303abfbd00a55964c72.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221120/613eb7d1b30b45829715224ca8980265.png
[20201209234822456.png_pic_center]: /images/20221120/467b15f2c40644b096b255fc6d266430.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221120/a2a7852056654cda8d810a332e59f661.png
[20201209234915861.png_pic_center]: /images/20221120/668ec91ec1af4724a745a56164bb1950.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221120/88b36e6617a34cf5b5d034a9c7232c23.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 5]: /images/20221120/3143cf88f63945f6948d4f4c060b6f56.png
[20201209235251461.png_pic_center]: /images/20221120/b85f92bb215b42baaf61f46daff9105d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 6]: /images/20221120/5ecf6c4fda664110a002239c491374fb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 7]: /images/20221120/97bcdc150b364a51a3d604a485d0dbca.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 8]: /images/20221120/5bac64d190c34dad8fc05617f3612bae.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 9]: /images/20221120/80c0bf518b9349c49332dd0811fcdebc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 10]: /images/20221120/33b6187abb404866b57e5e61100f87b9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 11]: /images/20221120/ba6049943feb4a8b8800b5176437a4e8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 12]: /images/20221120/bde6e99e696c4fe681947e61d334de1f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 13]: /images/20221120/af099b074a544c008bb19de5b930a156.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 14]: /images/20221120/387b3b9b76d4494b8ab647b65e488d98.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 15]: /images/20221120/ff46e1edeffb4aaea63c310c7f79dcac.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 16]: /images/20221120/561acfd42080473a904ce36994f6a1f5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 17]: /images/20221120/995fa3bdf7254dc293c9a332b79d5bb6.png
[20201209235841731.png_pic_center]: /images/20221120/1b1a31c0f7204ce5909ccd0313e31303.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 18]: /images/20221120/8db95601a7e94b39b10c5d0e0762731f.png
[20201209235935364.gif_pic_center]: /images/20221120/1a9f2454303e432a88db2d82a2ceb70b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 19]: /images/20221120/139e2ac0fa7146668d6f033dde9eba03.png
[20201210000035993.gif_pic_center]: /images/20221120/a0da337cd0d7498c8c73e0c49bd847d2.png
[20201210000116471.gif_pic_center]: /images/20221120/97e62b823dfd42659ccaac49799cf514.png
[20201210000133962.png_pic_center]: /images/20221120/62f25509f5b3491ca0d4a2f6b07c92c3.png
[20201210000201900.gif_pic_center]: /images/20221120/9e38648ba10d4b11b3f9b6aad2babf0d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 20]: /images/20221120/0127e9100683467ab75f1d915c24a0ea.png
[20201210000259182.gif_pic_center]: /images/20221120/0c5ae752cb9b44e5a267230e10617366.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JieHlscWYxMjZjb20_size_16_color_FFFFFF_t_70_pic_center 21]: /images/20221120/a75fc2eb6a8d46b5a0b65307f1bc355d.png
[20201210000349821.gif_pic_center]: /images/20221120/e7c3a2dd615644ec993018a6938101d1.png