Rabbit MQ 笔记整理(含 RPC 模型、整合 SpringBoot 2.x 实战、集群)

1. MQ 引言

1.1 什么是 MQ？

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

1.2 MQ 的作用

消息队列已经逐渐成为企业IT系统内部通信的核心手段。它具有低耦合、可靠投递、广播、流量控制、最终一致性等一系列功能，成为异步RPC的主要手段之一。当今市面上有很多主流的消息中间件，如老牌的ActiveMQ、RabbitMQ，炙手可热的Kafka，阿里巴巴自主开发RocketMQ等。

1.2.1 应用解耦

场景：双11是购物狂节,用户下单后,订单系统需要通知库存系统,传统的做法就是订单系统调用库存系统的接口.

这种做法有一个缺点:

当库存系统出现故障时,订单就会失败。 订单系统和库存系统高耦合. 引入消息队列

• 订单系统:用户下单后,订单系统完成持久化处理,将消息写入消息队列,返回用户订单下单成功。

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

---

1.2.2 流量消峰

1.2.2.1 思路一

如果订单系统最多能处理一万次订单，这个处理能力应付正常时段的下单时绰绰有余，正常时段我们下单一秒后就能返回结果。但是在高峰期，如果有两万次下单操作系统是处理不了的，只能限制订单超过一万后不允许用户下单。

使用消息队列做缓冲，我们可以取消这个限制，把一秒内下的订单分散成一段时间来处理，这事有些用户可能在下单十几秒后才能收到下单成功的操作，但是比不能下单的体验要好。

1.2.2.2 思路二

场景: 秒杀活动，一般会因为流量过大，导致应用挂掉,为了解决这个问题，一般在应用前端加入消息队列。

作用:

? 1.可以控制活动人数，超过此一定阀值的订单直接丢弃(我为什么秒杀一次都没有成功过呢^^)

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

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

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

---

1.2.3 消息分发

多个服务对数据感兴趣，只需要监听同一类消息即可处理。

例如A产生数据，B对数据感兴趣。如果没有消息的队列A每次处理完需要调用一下B服务。过了一段时间C对数据也感性，A就需要改代码，调用B服务，调用C服务。只要有服务需要，A服务都要改动代码。很不方便。

有了消息队列后，A只管发送一次消息，B对消息感兴趣，只需要监听消息。C感兴趣，C也去监听消息。A服务作为基础服务完全不需要有改动。

---

1.2.4 异步消息

场景说明：用户注册后，需要发注册邮件和注册短信,传统的做法有两种 1.串行的方式 2.并行的方式

• 串行方式: 将注册信息写入数据库后,发送注册邮件,再发送注册短信,以上三个任务全部完成后才返回给客户端。 这有一个问题是,邮件,短信并不是必须的,它只是一个通知,而这种做法让客户端等待没有必要等待的东西.

  

• 并行方式:将注册信息写入数据库后,发送邮件的同时,发送短信,以上三个任务完成后,返回给客户端,并行的方式能提高处理的时间。

  

• 消息队列:假设三个业务节点分别使用50ms,串行方式使用时间150ms,并行使用时间100ms。虽然并行已经提高的处理时间,但是,前面说过,邮件和短信对我正常的使用网站没有任何影响，客户端没有必要等着其发送完成才显示注册成功,应该是写入数据库后就返回. 消息队列: 引入消息队列后，把发送邮件,短信不是必须的业务逻辑异步处理

  

由此可以看出,引入消息队列后，用户的响应时间就等于写入数据库的时间+写入消息队列的时间(可以忽略不计),引入消息队列后处理后,响应时间是串行的3倍,是并行的2倍。

参考资料1：https://www.jianshu.com/p/9a0e9ffa17dd

参考资料2：《MQ消息中间件之RabbitMQ以及整合SpringBoot2.x实战教程》—— B站up主：编程不良人

1.3 几大 MQ 的对比

# 1.ActiveMQActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。它是一个完全支持JMS规范的的消息中间件。丰富的API,多种集群架构模式让ActiveMQ在业界成为老牌的消息中间件,在中小型企业颇受欢迎!# 2.KafkaKafka是LinkedIn开源的分布式发布-订阅消息系统，目前归属于Apache顶级项目。Kafka主要特点是基于Pull的模式来处理消息消费，追求高吞吐量，一开始的目的就是用于日志收集和传输。0.8版本开始支持复制，不支持事务，对消息的重复、丢失、错误没有严格要求，适合产生大量数据的互联网服务的数据收集业务。# 3.RocketMQRocketMQ是阿里开源的消息中间件，它是纯Java开发，具有高吞吐量、高可用性、适合大规模分布式系统应用的特点。RocketMQ思路起源于Kafka，但并不是Kafka的一个Copy，它对消息的可靠传输及事务性做了优化，目前在阿里集团被广泛应用于交易、充值、流计算、消息推送、日志流式处理、binglog分发等场景。# 4.RabbitMQRabbitMQ是使用Erlang语言开发的开源消息队列系统，基于AMQP协议来实现。AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。AMQP协议更多用在企业系统内对数据一致性、稳定性和可靠性要求很高的场景，对性能和吞吐量的要求还在其次。

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

参考：《MQ消息中间件之RabbitMQ以及整合SpringBoot2.x实战教程》—— B站up主：编程不良人

2. RabbitMQ 引言

2.1 RabbitMQ

官网：RabbitMQ

RabbitMQ是实现了高级消息队列协议（AMQP）的开源消息代理软件（亦称面向消息的中间件）。RabbitMQ服务器是用Erlang语言编写的，而群集和故障转移是构建在开放电信平台框架上的。所有主要的编程语言均有与代理接口通讯的客户端库)。

2.2 AMQP

AMQP，即Advanced Message Queuing Protocol，一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。

AMQP在2003年时被提出，最早用于解决金融领不同平台之间的消息传递交互问题。更准确的说是一种binary wire-level protocol（链接协议）。这是其和JMS的本质差别，AMQP不从API层进行限定，而是直接定义网络交换的数据格式。这使得实现了AMQP的provider天然性就是跨平台的。

AMQP 协议模型：

3. RabbitMQ 的安装（CentOS 7.x）

参考：https://www.cnblogs.com/fengyumeng/p/11133924.html

3.1 安装 perl、wget 工具

yum -y install gcc glibc-devel make ncurses-devel openssl-devel xmlto perl wget gtk2-devel binutils-devel

3.2 安装 erlang

• 下载安装包

wget http://erlang.org/download/otp_src_22.0.tar.gz

这里会非常非常慢，这里提供网盘链接（包括 erlang、rabbitmq、socat），可以直接下载，不过如果版本变了那就没办法了，可以去找找看镜像站有没有。

链接: https://pan.baidu.com/s/1foq6lg9GG31pywsVgQuurQ 密码: t5da

• 解压

tar -zxvf otp_src_22.0.tar.gz

• 移动解压后得到的文件夹到 /usr/local/

mv otp_src_22.0 /usr/local/

• 切换目录

cd /usr/local/otp_src_22.0/

• 创建即将安装 erlang 的目录

mkdir ../erlang

• 配置安装环境

./configure --prefix=/usr/local/erlang

这里有报错的话别管。

• 安装

make install

• 检查是否安装成功

ll /usr/local/erlang/bin

• 配置环境变量

echo 'export PATH=$PATH:/usr/local/erlang/bin' >> /etc/profile

• 刷新配置文件

source /etc/profile

• 进入 erlang 编辑器

erl

如果出现以下界面，就说明安装成功啦~

• 退出

halt().

3.3 安装 socat

socat作用是在两个流之间建立双向的通道，且支持众多协议和链接方式：ip，tcp，udp，ipv6，pipe，exec，system，open，proxy，openssl，socket 等。

将上面提供的 socat 安装包上传到 Centos 中

然后然后在对应的目录下执行命令：

rpm -ivh socat-1.7.3.2-2.el7.x86_64.rpm

3.4 安装 RabbitMQ

• 下载

wget https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.7.15/rabbitmq-server-generic-unix-3.7.15.tar.xz

如果嫌太慢也可以将上述网盘提供的后缀为tar.xz RabbitMQ 安装包上传到 CentOS 中

• 安装 xz 解压工具

yum install -y xz

• 解压tar.xz压缩包，得到 tar 包

/bin/xz -d rabbitmq-server-generic-unix-3.7.15.tar.xz

• 解压 tar 包

tar -xvf rabbitmq-server-generic-unix-3.7.15.tar

• 移动到 /usr/local 并改名为 rabbitmq

mv rabbitmq_server-3.7.15/ /usr/local/rabbitmq

• 配置环境变量

echo 'export PATH=$PATH:/usr/local/rabbitmq/sbin' >> /etc/profile

• 刷新配置文件

source /etc/profile

• 创建配置目录

mkdir /etc/rabbitmq

• 将 RabbitMQ 配置文件模板复制到 /etc/rabbitmq/ 目录下

我这里安装的时候没找到这个模板文件，所以需要从 github 上下载，链接如下：

https://github.com/rabbitmq/rabbitmq-server/blob/master/docs/rabbitmq.conf.example

复制到 /etc/rabbitmq/ 目录下并改名为 rabbitmq.config

• 开启来宾账户

开启这个来宾访问的原因是为了能访问 RabbitMQ 的 Web 管理页面。

vim /etc/rabbitmq/rabbitmq.config

• 开启 RabbitMQ

rabbitmq-server -detached

• 关闭 RabbitMQ

rabbitmqctl stop

• 查看 RabbitMQ 状态

rabbitmqctl status

4. Rabbit MQ 的用户管理

4.1 命令行形式

4.1.1 添加一个用户

rabbitmqctl add_user 账户名 密码

如：

rabbitmqctl add_user hedon hedon

4.1.2 设置权限

rabbitmqctl set_permissions -p "/" 用户名 ".*" ".*" ".*"

4.1.3 查看用户权限

rabbitmqctl list_user_permissions 用户名

4.1.4 设置tag

rabbitmqctl set_user_tags 用户名 administrator

4.1.5 查看所有用户

rabbitmqctl list_users

4.1.6 删除用户

rabbitmqctl delete_user 用户名

4.2 Web 界面管理模式

详见下面的 Web 管理。

5. RabbitMQ 的 Web 管理

5.1 在开启 RabbitMQ 的基础上开启 web 插件

rabbitmq-plugins enable rabbitmq_management

5.2 访问 web 管理界面

5.2.1 本机的话访问 http://localhost:15672

默认的账号密码都是 guest（就是我们刚刚开启的来宾访问）。登录后如下：

5.2.2 非本机的话访问 http://虚拟机ip地址:15672

这里会出现一个问题：guest 账户只能在本机使用，所以这里我们需要先增加用户。

在创建了新的账户 hedon 之后，我们就可以在非本机访问 RabbitMQ 的 Web 界面了：

5.3 Overview

5.4 Connections

显示当前有多少个客户端连接到 RabbitMQ。

5.5 Channels

显示当前有多少个通道。

5.6 Exchanges

5.7 Queues

5.8 Admin

当点击某个用户名时，可以跳转到管理该用户的界面：

6. RabbitMQ 支持的消息模型

6.0 模型总览

注意：使用的时候需要先在项目中注入 rabbitmq 的依赖坐标：

<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.9.0</version>
</dependency>

6.1 Hello World 模型

在上图的模型中，有以下概念：

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

6.1.1 生产者

public class Provider {
    public static void main(String[] args) throws IOException, TimeoutException {
    //1. 创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();//2. 配置连接工厂connectionFactory.setHost("172.16.208.138");connectionFactory.setPort(5672);connectionFactory.setUsername("hedon");connectionFactory.setPassword("hedon");connectionFactory.setVirtualHost("/hedon");//3. 获取连接对象Connection connection = connectionFactory.newConnection();//4. 创建通道Channel channel = connection.createChannel();/*** 5. 申明队列** 参数1（queue）：队列名，没有的话会自动创建* 参数2（durable）：是否持久化，在 RabbitMQ 重启时是否会自动删除该队列（即使保存队列，队列中的消息还是会被清空）* 参数3（exclusive）：是否独占队列* 参数4（autoDelete）：当队列中没有消息时是否自动删除* 参数5（arguments）：队列的其他属性，这里可以填构造方法中的一些参数*/channel.queueDeclare("hello",false,false,false,null);//*** 6. 发布信息** 参数1（exchange）：交换机，直连模式不需要交换机* 参数2（routingKey）：路由key，直连模式就是对列名称* 参数3（props）：其他属性* 参数4（body）：要发布的信息，需要转为字节码*/channel.basicPublish("","hello",null,"hello rabbitmq".getBytes());//7. 关闭通道channel.close();//8. 关闭连接connection.close();}
}

生产者运行6次，会发送6条消息到通道中：

6.1.2 消费者

public class Consumer {
    public static void main(String[] args) throws IOException, TimeoutException {
    //1. 创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();//2. 配置连接工厂connectionFactory.setHost("172.16.208.138");connectionFactory.setPort(5672);connectionFactory.setUsername("hedon");connectionFactory.setPassword("hedon");connectionFactory.setVirtualHost("/hedon");//3. 获取连接对象Connection connection = connectionFactory.newConnection();//4. 创建通道Channel channel = connection.createChannel();/*** 5. 申明队列 => 这里需要跟发布者一一对应* * 参数1（queue）：队列名，没有的话会自动创建* 参数2（durable）：是否持久化，在 RabbitMQ 重启时是否会自动删除该队列（即使保存队列，队列中的消息还是会被清空）* 参数3（exclusive）：是否独占队列* 参数4（autoDelete）：当队列中没有消息时是否自动删除* 参数5（arguments）：队列的其他属性，这里可以填构造方法中的一些参数*/channel.queueDeclare("hello",false,false,false,null);/*** 6. 获取消息** 参数1（queue）：队列名* 参数2（autoAck）：如果服务器应考虑消息传递后已确认，则为true* 参数3（callback）：一个实现了 Consumer 接口的对象*/channel.basicConsume("hello",true,new DefaultConsumer(channel){
    /*** 7. 处理信息的接收** @param consumerTag 消费者标签* @param envelope 消息的打包数据* @param properties AMQP的属性，消息的内容头信息* @param body 信息的内容* @throws IOException*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    //输出消息System.out.println(new String(body));}});//这里不关闭的话会一直监听，一但有消息进入，就会输出。//channel.close();//connection.close();}
}

消费者运行，会一直监听，把所有消息都输出，如果有新的消息进入，也会输出：

6.2 Work Queues 模型

Work queues，也被称为（Task queues），任务模型。当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。此时就可以使用work 模型：让多个消费者绑定到一个队列，共同消费队列中的消息。队列中的消息一旦消费，就会消失，因此任务是不会被重复执行的。

6.2.1 平均消费

6.2.1.1 生产者

public class Provider {
    public static void main(String[] args) throws IOException {
    //1. 通过自行封装的方法获取连接对象Connection connection = RabbitMqUtils.getConnection();//2. 获取通道对象Channel channel = connection.createChannel();//3. 申明队列channel.queueDeclare("work",true,false,false,null);//4. 发布信息for (int i = 0; i < 100; i++) {
    channel.basicPublish("","work",null,("发布消息"+i).getBytes());}//5. 通过自行封装的方法关闭资源RabbitMqUtils.closeChannelAndConnection(channel,connection);}
}

6.2.1.2 消费者一

public class ConsumerOne {
    public static void main(String[] args) throws IOException {
    //1. 通过自行封装的方法获取连接对象Connection connection = RabbitMqUtils.getConnection();//2. 获取通道对象Channel channel = connection.createChannel();//3. 申明队列channel.queueDeclare("work",true,false,false,null);//4. 消费消息channel.basicConsume("work",new DefaultConsumer(channel){
    @Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    System.out.println("这是消费者1在消费信息：" + new String(body));}});}
}

6.2.1.3 消费者二

public class ConsumerTwo {
    public static void main(String[] args) throws IOException {
    //1. 通过自行封装的方法获取连接对象Connection connection = RabbitMqUtils.getConnection();//2. 获取通道对象Channel channel = connection.createChannel();//3. 申明队列channel.queueDeclare("work",true,false,false,null);//4. 消费消息channel.basicConsume("work",new DefaultConsumer(channel){
    @Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    System.out.println("这是消费者2在消费信息：" + new String(body));}});}
}

6.2.1.4 测试

先运行两个消费者，然后运行生产者产生消息，观察两个消费者消费消息的情况：

通过上图我们可以发现 work 模型在默认情况下，RabbitMQ 会按顺序将每个消息发送给下一个消费者者。平均而言，每个消费者都会收到相同数量的消息。这种分发消息的方式称为轮询。

6.2.2 能者多劳

6.2.2.1 平均消费可能出现的问题

在平均消费模式中，消费者只要从队列中拿到消息，就立刻发送确认机制，有可能在处理消息的时候就突然宕机了或者出现意外了，这样消息还没来得及消费就遗失了，就造成业务数据的丢失。

另外，也有可能两个消费者处理消息的效率不一样，就有可能造成一个消费者已经消费完消息然后闲着，而另外一个消费者拿到了消息，却一直处于处理消息的状态，造成资源的浪费。

所以我们需要做三件事：

• 关闭消息自动确认机制

  //第二个参数 = false：关闭消息自动确认机制
  channel.basicConsume(队列名,false,new DefaultConsumer(channel){
        ....
  })
  

• 不能一次性把消息交给消费者

  channel.basicQos(1); //每一次只消费一个消息
  

• 主动进行确认

  //参数1：确认的是队列中的哪个具体消息
  //参数2：是否开启多个消息同时确认
  channel.basicAck(envelope.getDeliveryTag(),false);
  

6.2.2.2 修改消费者

6.2.2.3 再次测试

从上述结果中可以发现，由于我先前修改了消费者1的代码，让它的执行时间边长，所以它就只来得及接收并处理一条信息，其他信息都被消费者2接收并处理了，这样就达到了“能者多劳”的效果了。

6.3 Fanout 模型

fanout 模型是一种广播模型，也就是一个生产者可以发一个消息，进行广播，让多个消费者消费同一个消息。

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

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

6.3.1 生产者

public class Provider {
    public static void main(String[] args) throws IOException {
    //1. 通过自行封装的方法获取连接对象Connection connection = RabbitMqUtils.getConnection();//2. 获取通道对象Channel channel = connection.createChannel();//3. 声明交换机 => fanout 模式的交换机 type 需要制定为 fanoutchannel.exchangeDeclare("logs","fanout");//4. 广播消息channel.basicPublish("logs","",null,"这是一条广播".getBytes());//5. 关闭资源RabbitMqUtils.closeChannelAndConnection(channel,connection);}
}

6.3.2 消费者一

public class ConsumerOne {
    public static void main(String[] args) throws IOException {
    //1. 通过自行封装的方法获取连接对象Connection connection = RabbitMqUtils.getConnection();//2. 获取通道对象Channel channel = connection.createChannel();//3. 绑定交换机 => fanout 模式的交换机 type 需要制定为 fanoutchannel.exchangeDeclare("logs","fanout");//4. 创建临时队列，这里会随机生成一个队列名称String queue = channel.queueDeclare().getQueue();//5. 将临时队列绑定到交换机上，fanout模式暂时还不需要routingKeychannel.queueBind(queue,"logs","");//6. 消费消息channel.basicConsume(queue,true,new DefaultConsumer(channel){
    @Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    System.out.println("消费者1接收到消息："+new String(body));}});}
}

消费者二同理。

6.3.3 测试

6.4 Routing 模型

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

在 Direct 模型下：

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

图解：

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

6.6.1 生产者

public class Provider {
    public static void main(String[] args) throws IOException {
    //1. 通过自行封装的方法获取连接对象Connection connection = RabbitMqUtils.getConnection();//2. 获取通道对象Channel channel = connection.createChannel();//3. 声明交换机 => routing 模型的交换机类型是 directchannel.exchangeDeclare("ex_direct","direct");//4. 广播消息 => 一条是 error，另一条是 infochannel.basicPublish("ex_direct","error",null,"这是一条error广播".getBytes());channel.basicPublish("ex_direct","info",null,"这是一条info广播".getBytes());//5. 关闭资源RabbitMqUtils.closeChannelAndConnection(channel,connection);}
}

6.6.2 消费者一

public class ConsumerOne {
    public static void main(String[] args) throws IOException {
    //1. 通过自行封装的方法获取连接对象Connection connection = RabbitMqUtils.getConnection();//2. 获取通道对象Channel channel = connection.createChannel();//3. 绑定交换机 => routing 模型的交换机类型是 directchannel.exchangeDeclare("ex_direct","direct");//4. 创建临时队列String queue = channel.queueDeclare().getQueue();//5. 将临时队列绑定到交换机上 => 消费者1只能消费 routingKey 为 error 的消息channel.queueBind(queue,"ex_direct","error");//6. 消费消息channel.basicConsume(queue,true,new DefaultConsumer(channel){
    @Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    System.out.println("消费者1接收到消息："+new String(body));}});}
}

6.6.3 消费者二

public class ConsumerTwo {
    public static void main(String[] args) throws IOException {
    Connection connection = RabbitMqUtils.getConnection();Channel channel = connection.createChannel();channel.exchangeDeclare("ex_direct","direct");String queue = channel.queueDeclare().getQueue();//消费者2可以消费 routingKey 为 info、error 和 warning 的消息channel.queueBind(queue,"ex_direct","info");channel.queueBind(queue,"ex_direct","error");channel.queueBind(queue,"ex_direct","warning");channel.basicConsume(queue,true,new DefaultConsumer(channel){
    @Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    System.out.println("这是消费者2在消费消息："+new String(body));}});}
}

6.6.4 测试

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-8Y5es6wm-1601029214130)(…/…/…/…/…/…/…/…/Users/hedon-/Library/Application%20Support/typora-user-images/image-20200922223137183.png)]

6.5 Topics 模型

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

# 统配符*   匹配不多不少恰好1个词#   匹配一个或多个词
# 如:audit.#    匹配audit.irs.corporate或者 audit.irs 等audit.*    只能匹配 audit.irs

6.5.1 生产者

public class Provider {
    public static void main(String[] args) throws IOException {
    //1. 通过自行封装的方法获取连接对象Connection connection = RabbitMqUtils.getConnection();//2. 获取通道对象Channel channel = connection.createChannel();//3. 声明交换机 => topics 模型的交换机类型为 topicchannel.exchangeDeclare("ex_topic","topic");//4. 广播消息channel.basicPublish("ex_topic","logs",null,"这是一条 logs 广播".getBytes());channel.basicPublish("ex_topic","logs.a",null,"这是一条 logs.a 信息".getBytes());channel.basicPublish("ex_topic","logs.a.b",null,"这是一条 logs.a.b 广播".getBytes());channel.basicPublish("ex_topic","logs.a.b.c",null,"这是一条 logs.a.b.c 广播".getBytes());//5. 关闭资源RabbitMqUtils.closeChannelAndConnection(channel,connection);}
}

6.5.2 消费者一

public class ConsumerOne {
    public static void main(String[] args) throws IOException {
    //1. 通过自行封装的方法获取连接对象Connection connection = RabbitMqUtils.getConnection();//2. 获取通道对象Channel channel = connection.createChannel();//3. 绑定交换机 => topics 模型的交换机类型为 topicchannel.exchangeDeclare("ex_topic","topic");//4. 创建临时队列String queue = channel.queueDeclare().getQueue();//5. 将临时队列绑定到交换机上 => 消费者1只能消费 routingKey 为 error 的消息//routingKey 的每个单词之间用"."来分割，通配符 * 表示只能匹配1个单词，如 logs.achannel.queueBind(queue,"ex_topic","logs.*");//6. 消费消息channel.basicConsume(queue,true,new DefaultConsumer(channel){
    @Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    System.out.println("消费者1接收到消息："+new String(body));}});}
}

6.5.3 消费者二

public class ConsumerTwo {
    public static void main(String[] args) throws IOException {
    Connection connection = RabbitMqUtils.getConnection();Channel channel = connection.createChannel();channel.exchangeDeclare("ex_topic","topic");String queue = channel.queueDeclare().getQueue();//消费者2可以消费 routingKey 为 info、error 和 warning 的消息//每个单词之间用"."来分割，通配符 # 表示可以匹配多个单词，如 logs.a，logs.a.bchannel.queueBind(queue,"ex_topic","logs.#");channel.basicConsume(queue,true,new DefaultConsumer(channel){
    @Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    System.out.println("这是消费者2在消费消息："+new String(body));}});}
}

6.5.4 测试

如下图，消费者1的 routingKey 为“logs.*”，所以 logs. 后面必须有一个单词才能被消费者1消费，而消费者2的 routingKey 为“logs.#”，所以 logs. 后面没有单词或者有多个单词都可以被消费者2消费。

6.6 RPC 模型

在 Work Queues 模型中，我们学习了如何使用工作队列在多个工作人员之间分配耗时的任务。

但是，如果我们需要在远程计算机上运行功能并等待结果怎么办？那就可以用 RPC 模型（远程过程调用）。

我们将使用 RabbitMQ 构建RPC系统：客户端和可伸缩 RPC 服务器。由于我们没有值得分配的耗时任务，因此我们将创建一个虚拟 RPC服务，该服务返回斐波那契数。

图解：

• 对于RPC请求，客户端发送一条消息，该消息具有两个属性：replyTo（设置为仅为该请求创建的匿名互斥队列）和correlationId（设置为每个请求的唯一值）。

• 该请求被发送到rpc_queue队列。

• RPC工作程序（又名：服务器）正在等待该队列上的请求。出现请求时，它会使用 replyTo 字段中的队列来完成工作，并将消息和结果发送回客户端。

• 客户端等待答复队列中的数据。出现消息时，它将检查correlationId属性。如果它与请求中的值匹配，则将响应返回给应用程序。

补充：消息的属性

6.6.1 客户端

public class RPCClient {
    private Connection connection;private Channel channel;private String requestQueueName = "rpc_queue"; //发送请求的队列名称public static void main(String[] args) throws IOException, InterruptedException {
    //初始化 RPCClientRPCClient rpcClient = new RPCClient();rpcClient.connection = RabbitMqUtils.getConnection();rpcClient.channel = rpcClient.connection.createChannel();//发送 request 请求信息for (int i = 0; i < 5; i++) {
    String i_str = Integer.toString(i);System.out.println("现在客户端希望计算 fic("+i+")");String response = rpcClient.call(i_str);System.out.println("现在计算出来 fic("+i+") = "+ response);}//关闭资源close(rpcClient);}//发送请求，希望调用远程的函数public String call(String message) throws IOException, InterruptedException {
    //定义一个相关IDfinal String correlationId = UUID.randomUUID().toString();//回调队列String replyQueueName = channel.queueDeclare().getQueue();//定义消息属性AMQP.BasicProperties basicProperties = new AMQP.BasicProperties.Builder().correlationId(correlationId)  //设置相关ID.replyTo(replyQueueName)       //设置回调队列.build();//存储相应信息final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);//发送消息channel.basicPublish("",requestQueueName,basicProperties,message.getBytes("UTF-8"));//消费消息String ctag = channel.basicConsume(replyQueueName, true//参数3：服务器端传过来的回调对象, new DeliverCallback() {
    @Overridepublic void handle(String consumerTag, Delivery message) throws IOException {
    if (message.getProperties().getCorrelationId().equals(correlationId)) {
    response.offer(new String(message.getBody(), "UTF-8"));}}}, new CancelCallback() {
    @Overridepublic void handle(String consumerTag) throws IOException {
    }});String result = response.take();channel.basicCancel(ctag);return result;}//关闭资源public static void close(RPCClient rpcClient){
    RabbitMqUtils.closeChannelAndConnection(rpcClient.channel,rpcClient.connection);}}

6.6.2 服务端

public class RPCServer {
    //定义队列名称private static final String RPC_QUEUE_NAME = "rpc_queue";//计算斐波那契数列public static int fic(int a){
    if (a == 0) return 0;if (a == 1) return 1;return fic(a-1) + fic(a-2);}public static void main(String[] args) throws IOException {
    //调用自行封装的工具类获取连接对象Connection connection = RabbitMqUtils.getConnection();//获取通道Channel channel = connection.createChannel();//声明队列channel.queueDeclare(RPC_QUEUE_NAME,false,false,false,null);//清除队列channel.queuePurge(RPC_QUEUE_NAME);//每次处理一条信息channel.basicQos(1);//定义一个监听器Object monitor = new Object();//定义回调信息DeliverCallback deliverCallback = new DeliverCallback() {
    /*** @param consumerTag 消费者标签，可以与消费者建立联系* @param message 消费者发送过来的消息(消息属性、消息封装体、消息没人)*/@Overridepublic void handle(String consumerTag, Delivery message) throws IOException {
    //定义信息属性AMQP.BasicProperties basicProperties = new AMQP.BasicProperties.Builder().correlationId(message.getProperties().getCorrelationId()) //指明关联ID.build();//定义相应信息String response = "";//解析request信息try{
    String s = new String(message.getBody(),"UTF-8");int i = Integer.parseInt(s);System.out.println("正在计算 fic("+i+")");response += fic(i);}catch (Exception e){
    e.printStackTrace();}finally {
    //发布 response 消息channel.basicPublish("",message.getProperties().getReplyTo(),basicProperties,response.getBytes("UTF-8"));//手动确认信息channel.basicAck(message.getEnvelope().getDeliveryTag(),false);//RabbitMq 消费者工作线程通知 RPC 服务器所有者线程synchronized (monitor){
    /*** notify()** 唤醒处于等待的线程*/monitor.notify();}}}};//消费客户端发送过来的请求消息channel.basicConsume(RPC_QUEUE_NAME, false, deliverCallback, new CancelCallback() {
    // 等待并准备好使用来自RPC客户端的消息@Overridepublic void handle(String consumerTag) throws IOException {
    while (true){
    synchronized (monitor){
    try {
    /*** wait()** 使得当前线程立刻停止运行，处于等待状态（WAIT），* 并将当前线程置入锁对象的等待队列中，* 直到被通知（notify）或被中断为止。*/monitor.wait();}catch (InterruptedException e){
    e.printStackTrace();}}}}});}
}

6.6.3 测试

6.6.4 图解上述两段代码

上面两段代码是笔者模仿官网代码写出来的，刚开始写的时候也是一头雾水，现尝试画张图来加深理解。

6.7 Publisher Confirms 模型

待补。

---

7. SpringBoot 2.x 整合 RabbitMQ

7.1 Hello World 模型

7.1.1 导入 maven 依赖坐标

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

7.1.2 配置 RabbitMQ

spring:application:name: rabbitmq-springbootrabbitmq:host: 172.16.208.140port: 5672username: hedonpassword: hedonvirtual-host: /hedon

项目启动后，会自动注入 RabbitTemplate，该对象可以简化对 RabbitMQ 的使用。

7.1.3 生产者

关键对象：rabbitTemplate

@SpringBootTest
public class TestRabbitMQ {
    //注入 RabbitTemplate@Autowiredprivate RabbitTemplate rabbitTemplate;/*** ① Hello World 模型*/@Testpublic void testHelloWorld(){
    /*** 参数1：队列名称* 参数2：消息内容** 注意：当没有消费者在监听的时候，生产者的运行是不会有任何效果的。*/rabbitTemplate.convertAndSend("hello","hello world");}
}

7.1.4 消费者

关键注解：

• @RabbitListener：申明监听的队列
  • @Queue：具体指明哪一个队列及其相应的属性
• @RabbitHandler：指定收到消息时的回调方法

@Component
//注明消费者在监听 hello 这个队列
@RabbitListener(queuesToDeclare = @Queue(value = "hello",durable = "false",autoDelete = "true"))
public class HelloConsumer {
    /*** RabbitHandler 注解指明这是接收到消息时的回调方法* message 参数是传过来的消息*/@RabbitHandlerpublic void receive(String message){
    System.out.println("message = " + message);}
}

7.1.5 测试

运行生产者所在的单元测试方法 testHelloWorld()，观察消费者是否接收到消息。

---

7.2 Work Queues 模型

7.2.1 平均消费

默认就是平均消费的。

7.2.1.1 生产者

/*** ② Work Queues 模型*/
@Test
public void testWorkQueues(){
    for (int i = 0; i < 10; i++) {
    rabbitTemplate.convertAndSend("work","work 模型发出的第 "+i+" 消息");}
}

7.2.1.2 消费者

@Component
public class WorkConsumer {
    /*** 如果直接在方法上加 @RabbitListener 注解的话，那就不需要加 @RabbitHandler 注解了，默认就加上了。*/@RabbitListener(queuesToDeclare = @Queue(value = "work"))public void receive1(String message){
    System.out.println("这是 work 模型的消息者1得到的 message = " + message);}@RabbitListener(queuesToDeclare = @Queue(value = "work"))public void receive2(String message){
    System.out.println("这是 work 模型的消息者2得到的 message = " + message);}
}

7.2.1.3 测试

7.2.2 能者多劳

待补。

---

7.3 Fanout 模型

7.3.1 生产者

/*** ③ Fanout 模型*/
@Test
public void testFanout(){
    rabbitTemplate.convertAndSend("logs","","这是 Fanout 模型发送的消息");
}

7.3.2 消费者

@Component
public class FanoutConsumer {
    //因为是临时队列，所以不需要进行 queuesToDeclare@RabbitListener(bindings = {
    @QueueBinding(value = @Queue,                                     //不指定名称的话就会给我们随机创建临时队列exchange = @Exchange(value = "ex_logs",type = "fanout") //指定绑定的交换机)})public void receive1(String message){
    System.out.println("消费者1 message = " + message);}@RabbitListener(bindings = {
    @QueueBinding(value = @Queue,                                     //不指定名称的话就会给我们随机创建临时队列exchange = @Exchange(value = "ex_logs",type = "fanout") //指定绑定的交换机)})public void receive2(String message){
    System.out.println("消费者2 message = " + message);}}

7.3.3 测试

# 掉坑记录
这里笔者遇到了一个报错： ……………… received 'true' but current is 'false'。
# 报错原因
报错的原因其实是 queue 前后两次声明是不一样的，比如第一次声明的 queue 是要持久化的，第二次不想持久化了，这样就会报错。
# 解决方案
解决的办法就是去 web 管理界面删掉原来的 queue，然后再重新声明。

---

7.4 Routing 模型

7.4.1 生产者

/*** ④ Routing 模型*/
@Test
public void testRouting() {
    rabbitTemplate.convertAndSend("routing_directs","info","这是 Routing 模型 发送 info 的 key 的信息");rabbitTemplate.convertAndSend("routing_directs","error","这是 Routing 模型 发送 error 的 key 的信息");rabbitTemplate.convertAndSend("routing_directs","warning","这是 Routing 模型 发送 warning 的 key 的信息");
}

7.4.2 消费者

@Component
public class RouteConsumer {
    @RabbitListener( bindings = {
    @QueueBinding(value =  @Queue,exchange = @Exchange(value = "routing_directs",type = "direct"), //默认就是 direct 类型的key = {
    "info","error","warning"}  //接收 3 种类型的 routingKey 信息)})public void receive1(String message){
    System.out.println("消费者1 接受到的消息 message = "+ message);}@RabbitListener( bindings = {
    @QueueBinding(value =  @Queue,exchange = @Exchange(value = "routing_directs",type = "direct"),key = {
    "info"}          //只接收 info 类型的 routingKey 信息)})public void receive2(String message){
    System.out.println("消费者2 接受到的消息 message = "+ message);}
}

7.4.3 测试

---

7.5 Topics 模型

Topics 模型其实就是 Routing 模型将交换机的类型从 direct 改成 topic，然后支持通配符。

7.5.1 生产者

/*** ⑤ Topics 模型*/
@Test
public void testTopics(){
    rabbitTemplate.convertAndSend("routing_topics","info","这是 Topics 模型发送的 info 类型信息");rabbitTemplate.convertAndSend("routing_topics","info.a","这是 Topics 模型发送的 info.a 类型信息");rabbitTemplate.convertAndSend("routing_topics","info.a.b","这是 Topics 模型发送的 info.a.b 类型信息");rabbitTemplate.convertAndSend("routing_topics","info.a.b.c","这是 Topics 模型发送的 info.a.b.c 类型信息");
}

7.5.2 消费者

@Component
public class TopicsConsumer {
    @RabbitListener(bindings = {
    @QueueBinding(value = @Queue,exchange = @Exchange(value = "routing_topics",type = "topic"),key = {
    "info.#"}  // # 可以匹配 n 个单词)})public void receive1(String message){
    System.out.println("这是消费者1收到的消息 message = "+message);}@RabbitListener(bindings = {
    @QueueBinding(value = @Queue,exchange = @Exchange(value = "routing_topics",type = "topic"),key = {
    "info.*"}  // * 只能匹配 1 个单词)})public void receive2(String message){
    System.out.println("这是消费者2收到的消息 message = "+message);}}

7.5.3 测试

---

7.6 RPC 模型

待补。

7.7 Publisher Confirms 模型

待补。

---

8. RabbitMQ 集群

8.1 普通集群(副本集群)

默认情况下：RabbitMQ 代理操作所需的所有数据/状态都将跨所有节点复制。

这方面的一个例外是消息队列，默认情况下，消息队列仅位于一个节点上，尽管它们可以从所有节点看到和访问

8.1.1 架构图

? 核心解决问题: 当集群中某一时刻master节点宕机,可以对Quene中信息,进行备份。

8.1.2 集群搭建

# 0.集群规划node1: 10.15.0.3  mq1  master 主节点node2: 10.15.0.4  mq2  repl1  副本节点node3: 10.15.0.5  mq3  repl2  副本节点# 1.克隆三台机器主机名和ip映射vim /etc/hosts加入:10.15.0.3 mq110.15.0.4 mq210.15.0.5 mq3node1: vim /etc/hostname 加入:  mq1node2: vim /etc/hostname 加入:  mq2node3: vim /etc/hostname 加入:  mq3# 2.三个机器安装rabbitmq,并同步cookie文件,在node1上执行:scp /var/lib/rabbitmq/.erlang.cookie root@mq2:/var/lib/rabbitmq/scp /var/lib/rabbitmq/.erlang.cookie root@mq3:/var/lib/rabbitmq/# 3.查看cookie是否一致:node1: cat /var/lib/rabbitmq/.erlang.cookie node2: cat /var/lib/rabbitmq/.erlang.cookie node3: cat /var/lib/rabbitmq/.erlang.cookie # 4.后台启动rabbitmq所有节点执行如下命令,启动成功访问管理界面:rabbitmq-server -detached # 5.在node2和node3执行加入集群命令:1.关闭       rabbitmqctl stop_app2.加入集群    rabbitmqctl join_cluster rabbit@mq13.启动服务    rabbitmqctl start_app# 6.查看集群状态,任意节点执行:rabbitmqctl cluster_status# 7.如果出现如下显示,集群搭建成功:Cluster status of node rabbit@mq3 ...[{nodes,[{disc,[rabbit@mq1,rabbit@mq2,rabbit@mq3]}]},{running_nodes,[rabbit@mq1,rabbit@mq2,rabbit@mq3]},{cluster_name,<<"rabbit@mq1">>},{partitions,[]},{alarms,[{rabbit@mq1,[]},{rabbit@mq2,[]},{rabbit@mq3,[]}]}]# 8.登录管理界面,展示如下状态:

# 9.测试集群在node1上,创建队列

# 10.查看node2和node3节点:

# 11.关闭node1节点,执行如下命令,查看node2和node3:rabbitmqctl stop_app

---

8.2 镜像集群

镜像队列机制就是将队列在三个节点之间设置主从关系，消息会在三个节点之间进行自动同步，且如果其中一个节点不可用，并不会导致消息丢失或服务不可用的情况，提升MQ集群的整体高可用性。

8.2.1 架构图

8.2.2 集群搭建

# 0.策略说明rabbitmqctl set_policy [-p <vhost>] [--priority <priority>] [--apply-to <apply-to>] <name> <pattern>  <definition>-p Vhost： 可选参数，针对指定vhost下的queue进行设置Name:     policy的名称Pattern: queue的匹配模式(正则表达式)Definition：镜像定义，包括三个部分ha-mode, ha-params, ha-sync-modeha-mode:指明镜像队列的模式，有效值为 all/exactly/nodesall：表示在集群中所有的节点上进行镜像exactly：表示在指定个数的节点上进行镜像，节点的个数由 ha-params 指定nodes：表示在指定的节点上进行镜像，节点名称通过 ha-params 指定ha-params：ha-mode模式需要用到的参数ha-sync-mode：进行队列中消息的同步方式，有效值为 automatic 和 manual(默认)priority：可选参数，policy 的优先级(数字越高，优先级越高)
# 1.查看当前策略rabbitmqctl list_policies# 2.添加策略rabbitmqctl set_policy ha-all '^hello' '{"ha-mode":"all","ha-sync-mode":"automatic"}' 说明:策略正则表达式为 “^” 表示所有匹配所有队列名称  ^hello:匹配hello开头队列# 3.删除策略rabbitmqctl clear_policy ha-all# 4.测试集群

---