笔触狂放9

这个屌丝很懒，什么也没留下！

热门标签

RabbitMQ消息中间件技术精讲全集

作者：笔触狂放9 | 2024-02-21 21:52:14

踩

RabbitMQ消息中间件技术精讲全集

RabbitMQ消息中间件技术精讲

一. 主流消息中间件介绍

1.1 ActiveMQ

ActiveMQ是Apache出品，最流行的，能力强劲的开源消息总线，并且它是一个完全支持JMS规范的消息中间件。
其丰富的API，多种集群构建模式使得他成为业界老牌消息中间件，在中小型企业中应用广泛！
MQ衡量指标：服务性能、数据存储、集群架构；

优点：老牌，稳定性高，成熟度高，集群架构模式好(Zookeeper)，对性能要求不高的可以使用，它只需要引入依赖即可，SpringBoot自身集成了。
缺点：性能低，延迟高；

集群架构图：

1.2 Kafka

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

只要有足够大的内存，就能承担很大的数据传输；

Kafka集群模式：(集群复制，数据就不容易丢失)

1.3 RocketMQ

RocketMQ是阿里开源的消息中间件，目前也已经孵化为Apache顶级项目，它是纯java开发，具有高吞吐量、高可用性、适合大规模分布式系统应用的特点。RocketMQ思路起源于Kafka，它对消息的可靠传输以及事务性做了优化，目前在阿里集团被广泛应用于交易、充值、流计算、消息推送、日志流式处理、binglog分发等场景。
集群拓扑：(替换了Zookeeper,因为它性能低，换成了Name Server)

它的分布式事务等很多功能是需要商业版才能有的，需要收费。它高性能，可靠性，支持水平扩展，它有一个最大的问题：商业收费。

1.4 RabbitMQ

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

能保障数据不丢失，可做高可用，负载均衡，性能也很高，建议使用。

二. RabbitMQ核心概念及AMQP协议

2.1 本章导航

互联网大厂为什么选择RabbitMQ
RabbitMQ的高性能之道是如何做到的？
什么是AMQP高级协议？
AMQP核心概念是什么？
RabbitMQ整体架构模型是什么样子的？
RabbitMQ消息是如何流转的？
RabbitMQ安装与使用？
命令行与管控台？
RabbitMQ消息生产与消费？
RabbitMQ交换机详解？
RabbitMQ队列、绑定、虚拟主机、消息

2.2 初始RabbitMQ

RabbitMQ是一个开源的消息代理和队列服务器，用来通过普通协议在完全不同的应用之间共享数据，RabbitMQ是使用Erlang语言来编写的，并且RabbitMQ是基于AMQP协议的；
很多大厂如滴滴、美团、头条等，因为：
- 开源、性能优秀、稳定性保障
- 提供可靠性消息投递模式(confirm)、返回模式(return)
- 与SpringAMQP完美的整合、API丰富。
- 集群模式丰富、表达式配置、HA模式、镜像队列模型
- 保证数据不丢失的前提做到高可靠性、可用性。
RabbitMQ高性能的原因？
- Erlang语言在交换机的交互方面性能优秀的（Erlang语言最初在于交换机领域的架构模式，这样使得RabbitMQ在Broker之间进行数据交互的性能是非常优秀的）
- Erlang有着和原生Socket一样的延迟
什么是AMQP高级消息队列协议？
- AMQP定义：具有现代特征的二进制协议。是一个提供统一消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。
AMQP协议模型：

Publisher 推送消息前先与Server建立连接，找到Virtual host，然后将消息推送至Exchange交换机。而交换机与Message Queue有绑定关系(一个交换机相当于一个独立的虚拟机，而这个虚拟机内的各种独立的应用就相当于一个Queue，这个Queue与交换机绑定)，Consumer通过绑定的对队列，而交换机也绑定了队列。发送者将消息发送给交换机，这样就能完成消息的推送了。

2.3 AMQP核心概念

Server：又称Broker，接收客户端的连接，实现AMQP实体服务。
Connection:连接，应用程序与Broker的网络连接。
Channel：网络信道，几乎所有的操作都在Channel中进行，Channel是进行消息读写的通道。客户端可建立多个Channel，每个Channel代表一个会话任务。(它是一个核心的概念)
Message：消息，服务器和应用程序之间传送的数据，由Properties和Body组成。Properties可以对消息进行修饰，比如消息的优先级、延迟等高级特性；Body则就是消息体内容。
Virtual host: 虚拟地址，用于进行逻辑隔离，最上层的消息路由。一个Virtual Host里面可以有若干个Exchange和Queue，同一个Virtual Host里面不能有相同名称的Exchange或Queue
Exchange:交换机，接收消息，根据路由键转发消息到绑定的队列。
Binding：Exchange和Queue之间的虚拟连接，binding中可以包含routing key
Routing Key:一个路由规则，虚拟机可用它来确定如何路由一个特定消息。
Queue：也称为Message Queue，消息队列，保存消息并将它们转发给消费者。

2.4 RabbitMQ的整体架构？

在这里插入图片描述

2.5 消息流转图：

在这里插入图片描述

生产者将消息发送至交换机，交换机将消息发送至指定的队列，而消费者则通过绑定的队列拿到此消息。

三. RabbitMQ的安装与使用

3.1 安装

先下载Erlang,然后下载RabbitMQ
官网地址：http://www.rabbitmq.com/
提前准备：安装Linux必要依赖包
下载RabbitMQ必须安装包
配置文件修改

下载的版本不一定找最新的版本，要找最稳定的(版本的升级会伴随一定的风险)。同时要照顾到整体架构，比如其他的是否支持最新版本等，在升级上也要注意是否值得升级。

3.2 安装过程

先查看ErlangVersion: Erlang与RabbitMQ版本对照

rpm一键安装是最简单的安装方式，初学者可以使用此方式

启动Linux系统：

3.下载相关的包到相关的路径：

erlang要先安装；

输入命令安装Erlang:

rmp -ivh erlang-18.3-1.e17.centos.x86_64.rpm
1

输入命令安装RabbitMQ:

rpm -ivh rabbitmq-server-3.6.5-1.noarch.rpm
1

安装时出现如下问题，说明我们要先安装密钥：

输入密钥安装命令：

rpm -ivh socat-1.7.3.2-5.e17.lux.x86.rpm
1

重新输入安装RabbitMQ命令：

rpm -ivh rabbitmq-server-3.6.5-1.noarch.rpm
1

配置RabbitMQ：
1. 进入目录并编辑：
```
vim /usr/lib/rabbitmq/lib/rabbitmq_server-3.6.5/ebin/rabbit.app
1
```
2. 部分配置内容截图如下：

在配置文件中即可配置端口号，内存等操作。后面细讲，这里先做到能使用即可。

3.3 安装文档：

安装前环境准备：

yum install
build-essential openssl openssl-devel unixODBC unixODBC-devel
make gcc gcc-c++ kernel-devel m4 ncurses-devel tk tc xz
1
2
3

下载：

wget www.rabbitmq.com/releases/erlang/erlang-18.3-1.e17.centos.x86_64.rpm
wget http://repo.iotti.biz/CentOS/7/x86_64/socat-1.7.3.2-5.e17.lux.x86_64.rpm
wget www.rabbitmq.com/releases/rabbitmq-server/v3.6.5/rabbitmq-server-3.6.5-1.noarch.rpm
1
2
3

配置文件：

vim /usr/lib/rabbitmq/lib/rabbitmq_server-3.6.5/ebin/rabbit.app
1

比如修改密码、配置等等，例如：loopback_users中的<<“guest”>>，只保留guest

服务启动和停止：

启动：
```
rabbitmq-server start &
1
```
停止：
```
rabbitmqctl app_stop
1
```

管理插件：

rabbitmq-plugins enable rabbitmq_management
1

访问地址： ip:15672/

3.4 命令行与管控台–基础到高级操作

rabbitmqctl stop_app: 关闭应用
rabbitmqctl start_app: 启动应用
rabbitmqctl status: 节点状态

rabbitmqctl add_user username password: 添加用户
rabbitmqctl list_users:列出所有用户
rabbitmqctl delete_user username:删除用户
rabbitmqctl clear_permissions -p vhostpath username:清除用户权限
rabbitmqctl list_user_permissions username: 列出用户权限
rabbitmqctl change_password username newpassword:修改密码
rabbitmqctl set_permissions -p vhostpath username “."".”".*": 设置用户权限

rabbitmqctl add_vhost vhostpath:创建虚拟主机
rabbitmqctl list_vhosts: 列出所有虚拟主机
rabbitmqctl list_permissions -p vhostpath:列出虚拟主机上所有权限
rabbitmqctl list_permissions -p vhostpath: 列出虚拟主机上所有权限

rabbitmqctl list_queues:查看所有队列信息
rabbitmqctl -p vhostpath purge_queue blue:清除队列里的消息

rabbitmqctl reset: 移除所有数据，要在rabbitmqctl stop_app之后使用
rabbitmqctl join_cluster [–ram]: 组成集群命令
rabbitmqctl cluster_status:查看集群状态
rabbitmqctl change_cluster_node_type disc | ram 修改集群节点的存储形式
rabbitmqctl forget_cluster_node [–offline] 忘记节点(摘除节点)

集群配置失败，故障转移等情况下可以将启动失败的节点给移除掉。它可以在不启动的情况下对节点的摘除

rabbitmqctl rename_cluster_node oldnode1 newnode1 [oldnode2] [newnode2…] (修改节点名称)

命令行的操作能做的，可视化界面也可以做的。

3.5 RabbitMQ可视化界面

右上角：
可视化界面：
Type类型：
1. direct: 直连的方式
2. fanout：广播的方式
3. headers:请求头方式
4. topic: 主题模式
在Admin中可以配置账户以及账户的权限(如操作虚拟主机的权限)。(右边有选择框)
导入导出迁移备份：

3.6 急速入门-消息生产与消费

ConnectionFactory:获取连接工厂
Connection：一个连接
Channel:数据通信信道，可发送和接收消息。
Queue:具体的消息存储队列
Producer&Consumer生产和消费者

3.7 实战步骤：

创建SpringBoot项目，并引入相关的依赖使程序跑起来

引入RabbitMQ依赖：

<dependency>
	<groupId>com.rabbitmq</groupId>
	<artifactId>amqp-client</artifactId>
	<version>3.6.5</version>
</dependency>
1
2
3
4
5

创建一个生产者

public class Consumer{
	public static void main(String[] args) throws Exception{
	// 1. 创建一个ConnectionFactory， 并进行配置
	ConnectionFactory connectionFactory=new ConnectionFactory();
	connectionFactory.setHost("192.168.11.76");
	connectionFactory.setPort(5672);	
	connectionFactory.setVirtualHost("/");
	
	//2. 通过连接工厂创建连接
	Connection connection =connectionFactory.newConnection();
	
	//3. 通过connection创建一个Channel
	Channel channel=connection.craeteChannel();

	//4. 通过Channel发送数据
	for(int i=0;i<5;i++){
		String msg="Hello RabbitMQ!";
	    channel.basicPublish("","test001",null,msg.getBytes());	
	}
	
	//5. 记得要关闭相关的连接
	channel.close();
	connection.close();
	}	
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

创建一个消费者

public class Consumer{
//1. 创建一个ConnectionFactory,并进行配置
ConnectionFactory connectionFactory=new ConnectionFactory();
connectionFactory.setHost("192.168.11.76");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/");

//2. 通过连接工厂创建连接
Connection connection=connectionFactory.newConnection();

//3. 通过connection创建一个Channel
Channel channel = connection.createChannel();

//4. 声明（创建）一个队列
String queueName="test001";
channel.queueDeclare("test001",true,false,false,null);

//5. 创建消费者
QueueingConsumer queueingConsumer=new QueueingConsumer(channel);

//6. 设置Channel
channel.basicConsume(queueName,true,queueingConsumer);

//7. 获取消息
while(true){
	Delivery delivery=queueingConsumer.nextDelivery();
	String msg=new String(delivery.getBod());
	//Evelope envelope=delivery.getEnvelope();
	System.out.println("消费端:"+msg);
}}}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

我们在发送消息的过程中必须指定一个Exchange，如果指定的Exchange为空的话，它会使用默认的Exchange;

3.8 Exchange交换机

Exchange: 接收消息，并根据路由键转发消息所绑定的队列。
图示：

3.9 交换机属性

Name:交换机名称
Type:交换机类型：direct、topic、fanout、headers
Durability:是否需要持久化,true为持久化。
Auto Delete:当最后一个绑定到Exchange上的队列删除后，自动删除该Exchange。
Internal:当前Exchange是否用于RabbitMQ内部使用，默认为false；(一般为false，因为这个主要用于自己熟悉Erlang语言，并自己构建交换机扩展自定义插件等使用。)
Arguments:扩展参数，用于扩展AMQP协议自制定化使用。

3.10 Direct Exchange

所有发送到Direct Exchange的消息被转发到RouteKey中指定的Queue。

注意：Direct模式可以使用RabbitMQ自带的Exchange:default Exchange，所以不需要将Exchange进行任何绑定(binding)操作，消息传递时，RouteKey必须完全匹配才会被队列接收，否则该消息会被抛弃。

是否支持重连，如图：

3.11 Topic Exchange

所有发送到Topic Exchange的消息被转发到所有关心RouteKey中指定Topic的Queue上
Exchange将RouteKey和某Topic进行模糊匹配；此时队列需要绑定一个Topic;
如图所示：

符合条件的，被匹配到的都可以被接收到消息。

3.12 Fanout Exchange

不处理路由键，只需要简单的将队列绑定到交换机上。
发送到交换机的消息都会被转发到与该交换机绑定的所有队列上。
Fanout交换机转发消息是最快的。
如图所示：

还有一些其他的但不是很常用，这里不赘述了；

3.12 Binding - 绑定

Exchange 和Exchange、Queue之间的连接关系
Binding中可以包含RoutingKey或者参数

3.13 Queue-- 消息队列

消息队列：实际存储消息数据
Durability:是否持久化，Durable:是，Transient:否
Auto delete:如选Yes,代表当最后一个监听被移除之后，该Queue会自动被删除。

3.14 Message–消息

服务器和应用程序之间传送的数据；
本质上就是一段数据，由Properties和Payload(Body)组成
常用属性：delivery mode、headers（自定义属性）
其他属性：
1. content_type、content_encoding、priority
2. correlation_id、reply_to、expiration、message_id
3. timestamp、type、user_id、app_id、cluster_id

生产者自定义Header代码示例：

public class Procuder{
	public static void main(String[] args) throws Exception{
	// 1. 创建一个ConnectionFactory ,并进行配置
	ConnectionFactory connectionFactory=new ConnectionFactory();
	connectionFactory.setHost("192.168.11.76");
	connectionFactory.setPort(5672);
	connectionFactory.setVirtualHost("/");
	//2. 通过连接工厂连接
	Connection connection=connectionFactory.newConnection();
	//3. 通过connection创建一个Channel
	Channel channel=connection.createChannel();

	Map<String,Object> headers=new HashMap<>();
	headers.put("my1","111");
	headers.put("my2","222");
	AMQP.BasicProperties properties=new AMQP.BasicProperties.Builder()
	.deliveryMode(2)
	.contentEncoding("UTF-8")
	.expiration("10000")
	.headers(headers)
	.build();
	//4. 通过Channel发送数据
	channel.basicPublish("","test001",null,msg.getBytes());
	}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

消费者自定义获取header里信息代码示例：

public class Consumer{
//1. 创建一个ConnectionFactory,并进行配置
ConnectionFactory connectionFactory=new ConnectionFactory();
connectionFactory.setHost("192.168.11.76");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/");

//2. 通过连接工厂创建连接
Connection connection=connectionFactory.newConnection();

//3. 通过connection创建一个Channel
Channel channel = connection.createChannel();

//4. 声明（创建）一个队列
String queueName="test001";
channel.queueDeclare("test001",true,false,false,null);

//5. 创建消费者
QueueingConsumer queueingConsumer=new QueueingConsumer(channel);

//6. 设置Channel
channel.basicConsume(queueName,true,queueingConsumer);

//7. 获取消息
while(true){
	Delivery delivery=queueingConsumer.nextDelivery();
	String msg=new String(delivery.getBod());
	//Evelope envelope=delivery.getEnvelope();
	System.out.println("消费端:"+msg);
}}}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

3.15 Virtual host-虚拟主机

虚拟地址，用于进行逻辑隔离，最上层的消息路由。
一个Virtual Host里面可以有若干个Exchange和Queue
同一个Virtual Host里面不能有相同名称的Exchange或Queue;

本章小结： RabbitMQ的概念、安装与使用、管控台操作，结合RabbitMQ的特性、Exchange、Queue、Binding、RoutingKey、Message进行核心API的讲解，通过本章的学习，希望大家对RabbitMQ有一个初步的认知！

四. RabbitMQ的高级特性

4.1 本章导航

消息如何保障100%投递成功？
幂等性概念详解
在海量订单产生的业务高峰期，如何避免消息的重复消费问题？
Confirm确认消息、Return返回消息。

自定义消费者
消息的ACK域重回队列
消息的限流
TTL消息
死信队列

4.2 消息如何保障100%的投递成功？

什么是生产端的可靠性投递？
1. 保障消息的成功发出
2. 保障MQ节点的成功接收
3. 发送端收到MQ节点(Broker)确认应答
4. 完善的消息进行补偿机制
生产端:
1. 消息落库，对消息状态进行打标
  
  这里的打标，我们可以在消息将要发出的时候，将发出消息的状态修改，当确认收到了消息之后再修改状态。做一个定期轮询检查是否漏发，如果有则重新发送。
  简单来说，就是先将要发送消息的订单入库，然后再发送消息，如果消息未发送成功则进行补偿重发（延迟检查如五分钟后），最好是不做事务(影响性能)，少入DB。
2. 消息的延迟投递，做二次确认，回调检查

4.3 幂等性概念

幂等性是什么？
- 我们可以借鉴数据库的乐观锁机制：
- 比如我们执行一条更新库存的SQL语句：
- UPDATE T_REPS SET COUNT=COUNT-1,VERSION=VERSION+1 WHERE VERSION=1
消费端-幂等性保障
- 在海量订单产生的业务高峰期，如何避免消息的重复消费问题？
  - 消费端实现幂等性，就意味着，我们的消息永远不会消费多次，即使我们收到了多条哦一样的消息
- 业界主流的幂等性操作：
  - 唯一ID+指纹吗机制，利用数据库主键去重。
    1. 唯一ID+指纹码机制，利用数据库主键去重
    2. SELECT COUNT(1) FROM T_ORDER WHERE ID= 唯一ID+指纹码
    3. 好处：实现简单
    4. 坏处：高并发下有数据库写入的性能瓶颈
    5. 解决方案：跟进ID进行分库分表进行算法路由
  - 利用Redis的原子性去实现。
    1. 使用Redis进行幂等，需要考虑的问题
    2. 第一：我们是否要进行数据落库，如果落库的话，关键解决的问题是数据库和缓存如何做到原子性？
      
      在业务逻辑中，如果使用了数据库和Redis，在进行数据的流转中，Redis和数据库的事务不是一样的，要考虑到如何使事务一致性，同时成功同时失败等问题。
    3. 第二：如果不进行落库，那么都存储到缓存中，如何设置定时同步的策略？

4.4 Confirm 确认消息

理解Confirm消息确认机制：
- 消息的确认，是指生产者投递消息后，如果Broker收到消息，则会给我们生产者一个应答。
- 生产者接收应答，用来确定这条消息是否正常的发送到Broker，这种方式也是消息的可靠性投递的核心保障！
- 确认机制流程图：

如何实现Confirm确认消息？

第一步：在channel上开启确认模式：channel.confirmSelect()
第二步：在Channel上添加监听:addConfirmListener,监听成功和失败的返回结果，根据具体的结果对消息进行重新发送、或记录日志等后续处理！

代码实现如下：生产端

public class Producer{
public static void main(String[] args){
	//1. 创建ConnectionFactory
ConnectionFactory connectionFactory=new ConnectionFactory();
connectionFactory.setHost("192.168.11.76")
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/");

// 2. 获取Connection
Connection connection=ConnectionFactory.newConnection();

// 3. 通过Connection创建一个新的Channel
Channel channel=connection.createChannel();

// 4. 指定我们的消息投递模式：消息的确认模式
channel.confirmSelect();

String exchangeName="test_confirm_exchange";
String routingKey="confirm.save";

// 5. 发送一条消息
String msg="Hello RabbitMQ Send confirm message!";
channel.basicPublish(exchangeName,routingKey,null,msg.getBytes());

// 6. 添加一个确认监听
channel.addConfirmListener(new ConfirmListener(){
	@Override
	public void handleNack(long deliveryTag,boolean multiple) throws IOException{
		System.err.println("-------no ack!-------");
	}
	@Override
	public void handleAck(long deliveryTag,boolean multiple) throws IOException{
		System.err.println("---------ack!----------");
	}
})
}	}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

消费端：

public class Producer{
	public static void main(String[] args)throws Exception{
		// 1. 创建ConnectionFactory
	ConnectionFactory connectionFactory=new ConnectionFactory();
	connectionFactory.setHost("192.168.11.76");
	connectionFactory.setPort(5672);
	connectionFactory.setVirtualHost("/");
		//2. 获取connection
	Connection connection =connectionFactory.newConnection();
		
		//3. 通过Conneciton创建一个新的Channle
	String exchangeName="test_confirm_exchange";
	String routingKey="confirm.#";
	String queueName="test_confirm_queue";

	// 4. 声明交换机和队列  然后进行绑定设置，最后制定路由key
	channel.exchangeDeclare(exchangeName,"topic",true);
	channel.queueDeclare(queueName,true,false,false,null);
	channel.queueBind(queueName,exchangeName,routingKey);
	
	// 5.创建消费者
	QueueingConsumer queueingConsumer=new QueueingConsumer(channel);
	channel.basicConsume(queueName,true,queueingConsumer);
	while(true){
		Delivery delivery=queueingConsumer.nextDelivery();
		String msg=new String(delivery.getBody());
		System.err.println("消费端"+msg);
	}}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

4.5 Return消息机制

Return Lis tener 用于处理一些不可路由的消息！
我们的消息生产者，通过指定一个Exchange和RoutingKey，把消息送达到某一个队列中去，然后我们的消费者监听队列，进行消费处理操作！
但是在某些情况下，如果我们在发送消息的时候，当前的Exchange不存在或者指定的路由key路由不到，这个时候如果我们需要监听这种不可达的消息，就要使用Return Listener!
在基础API中有一个关键的配置项：
Mandatory:如果为true,则监听器会接收到路由不可达的消息，然后进行后续处理，如果为false，那么broker端自动删除该消息！

默认为false，当我们使用Return 消息机制的时候，我们需要将它设置为true;

接收端接收到了消息：

4.6 消费端自定义监听

自定义监听的原因：
- 我们一般就是在代码中编写while循环，进行consumer.nextDelivery方法进行获取下一条消息，然后进行消费处理！
- 但是我们使用自定义的Consumer更加的方便，解耦性更加的强，也是在实际工作中最常用的使用方式！
- 非常简单，消费者只需要继承DefaultConsumer类，然后重写handleDelivery方法即可；

继承DefaultConsumer的此类被写出后，需要进行绑定。(在交换机绑定时绑定自定义的Consumer)；

4.7 消费端限流

假设一个场景，首先，我们RabbitMQ服务器上有上万条未处理的消息，我们随便打开一个消费者客户端，会出现下面情况：
巨量的消息瞬间全部推送过来，但是我们单个客户端无法同时处理这么多数据；
RabbitMQ提供了一种qos(服务质量保证)功能，即在非自动确认消息的前提下，如果一定数目的消息(通过基于consume或者channel设置Qos的值)未被确认前，不进行消费新的消息。

void BasicQos(uint prefetchSize,ushort prefetchCount,bool global);
prefetchSize:0 #这里为0表示不限制
prefetchCount: 会告诉RabbitMQ不要同时给一个消费者推送多于N个消息，即一旦有N个消息还没有ack，则该consumer将block掉，直到有消息ack; (prefetchCount等于1即可)
global:true\false 是否将上面设置应用于channel
简单来说，就是上面限制是channel级别的还是consumer级别；

4.8 消息ACK与重回队列

消费端的手工ACK和NACK
- 消费端进行消费的时候，如果由于业务异常我们可以进行日志的记录，然后进行补偿！
- 如果由于服务器宕机等严重问题，那我们就需要手工进行ACK保障消费端消费成功！
消费端的重回队列
- 消费端重回队列是为了对没有处理成功的消息，把消息重新会递给Broker!
- 一般我们在实际应用中，都会关闭重回队列，也就是设置为False;

4.9 TTL队列/消息

TTL：
- TTL是Time To Live的缩写，也就是生存时间
- RabbitMQ支持消息的过期时间，在消息发送时可以进行指定。
- RabbitMQ支持队列的过期时间，从消息入队列开始计算，只要超过了队列的超时时间配置，那么消息会自动的删除。
图示：

这里可以配置队列的相关参数配置；
交换机与交换机之间也可以进行绑定

4.10 死信队列（这个消息没有被任何人消费，它就变成了死信队列）

利用DLX，当消息在一个队列中变成死信（dead message）之后，它能被重新publish到另一个Exchange，这个Exchange就是DLX;
消息变成死信有以下几种情况：
- 消息被拒绝(basic.reject/basic.nack) 并且requeue=false;
- 消息TTL过期
- 队列达到最大长度
详细解说：
- DLX也是一个正常的Exchange,和一般的Exchange没有区别，它能在任何的队列上被指定，实际上就是设置某个队列的属性。
- 当这个队列中有死信时，RabbitMQ就会自动的将这个消息重新发布到设置的Exchange上去，进而被路由到另一个队列。
- 可以监听这个队列中消息做相应的处理，这个特性可以弥补RabbitMQ3.0以前支持的immediate参数的功能；
- 然后我们进行正常声明交换机、队列、绑定，只不过我们需要在队列上加上一个参数即可：arguments.put(“x-dead-letter-exchange”,“dlx.exchange”);
- 这样消息在过期、requeue、队列在达到最大长度时，消息就可以直接路由到死信队列。

五. RabbitMQ高级整合应用

5.1 本章导航

RabbitMQ整合SpringAMQP实战
RabbitMQ整合SpringBoot实战
RabbitMQ整合SpringCloud实战

5.2 RabbitMQ整合Spring AMQP实战

RabbitAdmin
Spring AMQP声明
RabbitTemplate
SimpleMessageListenerContainer
MessageListenerAdapter
MessageConverter

5.3 RabbitAdmin

RabbitAdmin类可以很好的操作RabbitMQ，在Spring中直接进行注入即可

核心代码如下：

@Bean
public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory){
	RabbitAdmin rabbitAdmin=new RabbitAdmin(connectionFactory);
	rabbitAdmin.setAutoStartup(true);
	return rabbitAdmin;
}
1
2
3
4
5
6

然后使用RabbitTemplate的execute方法执行对应的声明、修改、删除等一系列RabbitMQ基础功能操作
例如：添加一个交换机、删除一个绑定、清空一个队列里的消息等等。

5.4 Spring的整合

假定已经创建好了项目，过程如下

引入依赖：

<dependency>
	<groupId>com.rabbitmq</groupId>
	<artifactId>amqp-client</artifactId</artifactId>
</dependency>
<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
1
2
3
4
5
6
7
8

添加配置类：RabbitMQConfig:

@Configuration
@ComponentScan("com.bfxy.spring.*")		//让此类被扫描到
public class RabbitMQConfig{
	
	@Bean
	public ConnectionFactory connectionFactory(){		//如果bean没有给name，就默认为方法的名称
		CachingConnectionFactory connectionFactory=new CachingConnectionFactory();
		connectionFactory.setAddress("192.168.11.76:5672");
		connectionFactory.setUsername("guest");
		connectionFactory.setPassword("guest");
		connectionFactory.setVirtualHost("/");
		return connectionFactory;
	}
	@Bean
	public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory){
		RabbitAdmin rabbitAdmin=new RabbitAdmin(connectionFactory);
		rabbitAdmin.setAutoStartup(true);
		return rabbitAdmin;
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

写一个测试类

@RunWith(SpringRunner.class)
@SpringBootTest
public class ApplicationTests{

@Test
public void contextLoads(){}

}

@Autowired
private RabbitAdmin rabbitAdmin;

@Test
public void testAdmin() throws Exception{
	rabbitAdmin.declareExchange(new DirectExchange("test.direct",false,false));
	rabbitAdmin.declareExchange(new TopicExchange("test.topic",false,false));
	rabbitAdmin.declareExchange(new FanoutExchange("test.fanout",false,false));
	rabbitAdmin.declareQueue(new Queue("test.direct.queue",false));
	rabbitAdmin.declareQueue(new Queue("test.topic.queue",false));
	rabbitAdmin.declareQueue(new Queue("test.fanout.queue",false));
	rabbitAdmin.declareBinding(new Binding("test.direct.queue",Binding.DestinationType.QUEUE,"test.direct","direct",new HashMap<>()));
	rabbitAdmin.declareBinding(
		BindingBuilder
		.bind(new Queue("test.topic.queue",false));		//直接创建队列
		.to(new TopicExchange("test.topic",false,false))		//直接创建交换机，建立关联关系
		.with("user.#"));
	rabbitAdmin.declarebinding(
		BindingBuilder
		.bind(new Queue("test.fanout.queue",false))
		.to(new FanoutExchange("test.fanout",false,false));
	)

	//清空队列数据
	rabbitAdmin.purgeQueue("test.topic.queue",false);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

常见的队列：1. FanoutExchange：将消息分发到所有的绑定队列，无RoutingKey的概念 2.HeadersExchange:通过添加属性key-value匹配 3. DirectExchange:按照routingKey分发到指定队列 4. TopicExchange:多关键词匹配

5.5 RabbitAdmin

注意: autoStartup 必须要设置为true,否则Spring容器不会加载RabbitAdmin类
RabbitAdmin底层实现就是从Spring容器中获取Exchange、Bingding、RoutingKey以及Queue的@Bean声明；
然后使用RabbitTemplate的execute方法执行对应的声明、修改、删除等一系列RabbitMQ基础功能操作；

5.6 SpringAMQP声明

使用SpringAMQP去声明，就需要使用SpringAMQP的如下模式，即声明@Bean方式

代码如下：

@Bean
public TopicExchange exchange(){
	return new TopicExchange("topic001",true,false);
}

@Bean
public Queue queue(){
	return new Queue("queue001",true);		//队列持久
}

@Bean
public Binding binding(){
	return BindingBuilder.bind(queue()).to(exchange()).with("spring.*");		//这里的with内容为routKey，可自定义
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

5.7 消息模板 RabbitTemplate

RabbitTemplate,即消息模板。我们在与SpringAMQP整合的时候进行发送消息的关键类
该类提供了丰富的发送消息的方法，包括可靠性投递消息方法、回调监听消息接口ConfirmCallback、返回值确认接口ReturnCallback等等。同样我们需要进入注入到Spring容器中，然后直接使用；（同样需要注入到Spring容器中进行使用）
在与Spring整合时需要实例化，但是在与SpringBoot整合时，在配置文件里添加配置即可；

RabbitMQConfig.java 核心代码如下：

@Configuration
@ComponentScan({com.bfxy.spring.*})			//这里的路径换成自己的路径
public class RabbitMQConfig{

@Bean
public ConnectionFactory connectionFactory(){
	CachingConnectionFactory connectionFactory=new CachingConnectionFactory();
	connectionFactory.setAddress("192.168.11.76:5672");
	connectionFactory.setUsername("guest");
	connectionFactory.setPassword("guest");
	connectionFactory.setVirtualHost("/");
	return connectionFactory;
}

@Bean
public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory){
	RabbitAdmin rabbitAdmin=new RabbitAdmin(connectionFactory);
	rabbitAdmin.setAutoStartup(true);
	return rabbitAdmin;
}

@Bean
public TopicExchange exchange(){
	return new TopicExchange("topic001",true,false);
}

@Bean
public Queue queue(){
	return new Queue("queue001",true);		//队列持久
}

@Bean
public Binding binding(){
	return BindingBuilder.bind(queue()).to(exchange()).with("spring.*");		//这里的with内容为routKey，可自定义
}

@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){
	RabbitTemplate rabbitTemplate =new RabbitTemplate(connectionFactory);
	return rabbitTemplate;		//在这里的上面可以对RabbitTemplate设置一些属性，最后再返回；
}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

测试类中测试

@Autowired
private RabbitTemplate rabbitTemplate;

@Test
public void testSendMessage() throws Exception{
	//1. 创建消息
	MessageProperties messageProperties =new MessageProperties();
	messageProperties.getHeaders().put("desc","信息描述...");
	messageProperties.getHeaders().put("type","自定义消息类型..");
	Message message=new Message("Hello RabbitMQ".getBytes(),messageProperteis);
	rabbitTemplate.converAndSend("topic001","spring.amqp",message,new MessagePostProcessor(){
		@Overide
		public Message postProcessMessage(Message message) throws AmqpException{
			System.err.println("----------添加额外的配置-------");
			message.getMessageProperties().getHeaders().put("desc","额外修改的信息描述");
			message.getMessageProperties().getHeaders().put("attr","额外新加的属性");		//新加的属性不一定设置在Header，我们也可以设置在别的地方
			return message;
		}
	});
}

// 简单版   最后面的方法是可以不要的 
@Test
public void testSendMessage2() throws Exception{
//创建消息
MessageProperties messageProperties=new MessageProperties();
messageProperteis.setContentType("text/plain");		//文本类型
Message message=new Message("mq 消息".getBytes(),messageproperties);
rabbitTemplate.converAndSend("topic001","spring.amqp",message);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

// 最简单版(消息体也可以不要，可直接传文本内容)
@Test
public void testSendMessage3() throws Exception{
rabbitTemplate.converAndSend(“topic001”,“spring.amqp”,“我是一段消息内容”);
}
}

5.8 简单消息监听容器：SimpleMessageListenerContainer

这个类非常的强大，我们可以对他进行很多设置，对于消费者的配置项，这个类都可以满足
监听队列（多个队列）、自动启动、自动声明功能
设置事务特性、事务管理器、事务属性、事务容量(并发)、是否开启事务、回滚消息等
设置消费者数量、最小最大数量、批量消费
设置消息确认和自动确认模式、是否重回队列、异常捕获handler函数
设置消费者标签生成策略、是否独占模式、消费者属性等
设置具体的监听器、消息转换器等等。

注意：SimpleMessageListenerContainer可以进行动态设置，比如在运行中的应用可以动态的修改其消费者数量的大小、接收消息的模式等。很多基于RabbitMQ的自制定化后端管控台在进行动态设置的时候，也是根据这一特性去实现的。所以可以看出SpringAMQP非常的强大；
思考一下：SimpleMessageListenerContainer为什么可以动态感知配置变更？

代码示例： (在RabbiMQConfig.java上面的这个配置类中继续添加Bean)

@Bean
public SimpleMessageListenerContainer messageContainer(ConnectioNFactory connectionFactory){
SimpleMessageListenerContainer container=new SimpleMessageListenerContainer(connectionFactory);
container.setQueues(queue001(),queue002());		//包含的队列
container.setConcurrentConsumers(1);		//当前消费者数量
container.setMaxConcurrentConsumers(5);		//最大数量
container.setDefaultRequeueRejected(false);
container.setAcknowledgeMode(AcknowledgMode.AUTO);
container.setConsumerTagStrategy(new ConsumerTagStrategy(){
	@Overide
	public String createConsumerTag(String queue){
		return queue+"_"+UUID.randomUUID().toString();
	}
});
container.setMessageListener(new ChannelAwareMessageListener(){
	@Override
	public void onMessage(Message message,Channel channel) throws Exception{
		String msg=new String(message.getBody());
		System.err.println("----消费者:---"+msg);
	}
})
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

部分图示：

我们在SimpleMessageListenerContainer 中给队列名设置了队列+UUID的形式，于是在可视化界面中就看到了；

5.9 消息监听适配器：MessageListenerAdapter

通过messageListenerAdapter的代码我们可以看出如下核心属性：
- defaultListenerMethod默认监听方法名称：用于设置监听方法名称
- Delegate委托对象：实际真实的委托对象，用于处理消息、
- queueOrTagToMethodName: 队列标识与方法名称组成的集合
- 可以一一进行队列与方法名称的匹配；
- 队列和方法名称绑定，即指定队列里的消息会被绑定的方法所接收处理；

比如我们上面使用SimpleMessageListenerContainer 定义了同一个队列的不同消费者(队列名+UUID)，我们可以使这每个消费者分别执行不同的方法，或者进行负载均衡等操作，可以使用此消息监听器实现；具体可百度；

5.10 MessageConverter消息转换器

我们在进行发送消息的时候，正常情况下消息体为二进制的数据方式进行传输，如果希望内部帮我们进行转换，或者指定自定义的转换器，就需要用到MessageConverter;
自定义常用转换器：MessageConverter,一般来讲都需要实现这个接口
重写下面两个方法：
- toMessage:java对象转换为Message
- fromMessage:Message对象转换为java对象
MessageConverter消息转换器：
- Json转换器：Jackson2JsonMessageConverter:可以进行java对象的转换功能；
- DefaultJackson2JavaTypeMapper映射器：可以进行java对象的映射关系；
- 自定义二进制转换器：比如图片类型、PDF、PPT、流媒体等

使用转换器的目的是当传入不同的类型的数据(如json，类，PDF，图片等)时，在消息的接收方接收到时也总是以传入的类型接收结果对象；我们通过写入不同的转换器以达到此种效果。具体可百度。

部分代码示例：

5.11 SpringBoot整合配置详解

publisher-confirms,实现一个监听器用于监听Broker端给我们返回的确认请求：RabbitTemplate.ConfirmCallback
publisher-returns,保证消息对Broker端是可达的，如果出现路由键不可达的情况，则使用监听器对不可达的消息进行后续的处理，保证消息的路由成功： RabbitTemplate.ReturnCallback

注意一点，在发送消息的时候对template进行配置mandatory=true保证监听有效；生产端还可以配置其他属性，比如发送重试，超时时间、次数、间隔等。

代码如下：

配置application.properties:

spring.rabbitmq.publisher-confirms=true
spring.rabbitmq.publisher-returns=true
spring.rabbitmq.template.mandatory=true
1
2
3

创建RabbitMQConfig.java（此步骤省略，参照以前代码）

创建生产者RabbitSender:

@Component
public class RabbitSender{

	@Autowired
	private RabbitTemplate rabbitTemplate;
	
	// ack确认
	final ConfirmCallback confirmCallback=new RabbitTemplate.ConfirmCallback(){
		@Override
		public void confirm(CorrelationData correlationData,boolean ack,String cause){
			System.err.println("correlationData:"+ correlationData);
			System.err.println("ack:"+ack);
			if(!ack){
				System.err.println("异常处理...");
			}
		}
	};

	// 消息发送后的返回
	final ReturnCallback returnCallback=new RabbitTemplate.ReturnCallback(){
		@Override
		public void returnedMessage(org.springframwork.amqp.core.Message message, int replyCode,String replyText, String exchange,String routingKey){
			System.err.println("return exchange: "+ exchange+",routingKey:"+routingKey);
		}
	};

		public void send(Object message,Map<String,Object> properties) throws Exception{
			MessageHeaders mhs=new MessageHeaders(properteis);
			Message msg=MessageBuilder.createMessage(message,mhs);
			rabbitTemplate.setConfirmCallback(confirmCallback);
			rabbitTemplate.setReturnCallback(returnCallback);
			CorrelationData cd=new CorrelationData("123456789");		//这里的id值一定要唯一。我们为了测试随便写的。用于生产环境可使用：id+时间戳等方式
			rabbitTemplate.converAndSend("exchange-1","springboot.hello",msg);
		}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

ack确认中的if(!ack){}里面，我们可以实现自己的方法。比如做事务的时候，当消息发送不成功，即ack=false时，我们可以将消息发送不成功的状态录入数据库。做定时任务对数据库中这些发送不成功的消息进行消息重试（重新发送），以保证消息最终都能成功发送，不会漏发；而returnCallback这里的方法，当消息返回失败的时候，它能告诉我们消息发送失败的原因等。 CorrelationData这里是为了确定此消息的唯一性。当我们确定消息发送失败等，我们可以对此指定的消息进行操作。所以里面的id要保证全局唯一。

测试类：

@RunWith(SpringRunner.class)
@SpringBootTest
public class ApplicationTests{
	@Test
	public void contextLoads(){}
	
	@Autowired
	private RabbitSender rabbitSender;
	
	private static SimpleDateFormat simpleDateFormat=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");

	@Test
	public void testSender1() throws Exception{
		Map<String,Object> properties =new HashMap<>();
		properties.put("number","12345");
		properties.put("send_time",simpleDateFormat.format(new Date()));
		rabbitSender.send("Hello RabbitMQ",properties);
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

这里的SimpleDateFormat是一个线程不安全的时间解析类，此处仅用于功能实现，生产环境使用请注意。

5.12 SpringBoot整合配置详解2

消费端核心配置：

spring.rabbitmq.listener.simple.acknowledge-mode=MANUAL
spring.rabbitmq.listener.simple.concurrency=1
spring.rabbitmq.listener.simple.max-concurrency=5;
1
2
3

其他要点：
- 首先配置手工确认模式，用于ACK的手工处理，这样我们可以保证消息的可靠性送达，或者在消费端消费失败的时候可以做到重回队列、根据业务记录日志等处理
- 可以设置消费端的监听个数和最大个数，用于控制消费端的并发情况。
@RabbitListener注解的使用
- 消费端监听@RabbitMQListener注解，这个对于在实际工作中非常的好用。
- @RabbitListener是一个组合注解，里面可以注解配置
- @QueueBinding、@Queue、@Exchange直接通过这个组合注解一次性搞定消费端交换机、队列、绑定、路由、并且配置监听功能等。

核心两个代码：

@RabbitListener(bindings=@QueueBinding(
	value=@Queue(value="queue-1",durable="true"),
	exchange=@Exchange(value="exchange-1",durable="true"),
	type="topic",
	ignoreDeclarationExceptions="true"),,key="springboot.*"))
1
2
3
4
5

@RabbitHandler
public void onMessage(Message message,Channel channel) throws Exception{]
1
2

由于类配置写在代码里非常不友好，所以强烈建议大家使用配置文件配置。

配置文件写法，代码如下：
1. 生产端代码配置：
2. 消费端代码配置：
3. 消息接收端：
4. 消息发送端：
  
  完整代码请参考上面一小节的发送端代码。
使用实体类接收和发送消息代码演示：
- 实体类：
- 消费监听的配置：
- 消费者代码编写：
- 生产者方法：
- 消息发送示例：
  
  这里是引入生产者的发送消息的方法，直接将此对象传输；

5.13 Spring Cloud Stream 整合

SpringCloud ,这个全家桶框架在整个中小型互联网公司异常的火爆，那么相对应着，Spring Cloud Stream就渐渐的被大家所重视起来，这一节课主要来介绍Spring Cloud Stream 如何与RabbitMQ进行集成。
Spring Cloud Stream 整体架构核心概念图
Spring Cloud Stream 整体架构核心概念图
Barista接口：
- Barista接口是定义来作为后面类的参数，这一接口定义来通道类型和通道名称，通道名称是作为配置用，通道类型则决定了app会使用这一通道进行发送消息还是从中接收消息
- @Output:输出注解，用于定义发送消息接口
- @Input: 输入注解，用于定义消息的消费者接口
- @StreamListener:用于定义监听方法的注解
使用Spring Cloud Stream 非常简单，只需要使用好这3个注解即可，在实现高性能消息的生产和消费的场景非常适合，但是使用Spring Cloud Stream 框架有一个非常大的问题就是不能实现可靠性的投递，也就是没法保证消息的100%可靠性，会存在少量消息丢失问题。

存在消息丢失的问题我们可以通过补偿机制进行解决；

为什么存在消息丢失？
- 这个原因是因为Spring Cloud Stream 框架为了和Kafka兼顾所以在实际工作中使用它的目的就是针对高性能的消息通信的！这点就是在当前版本Spring Cloud Stream 的定位。

5.14 Spring Cloud Stream 实战

xml中核心依赖：
Brista接口：
相关application.yml配置：
生产端代码：

消费端XML引入的核心依赖：
实体类(类名可以改的，不一定都叫这名，这是官方的名字)
消费端配置application.properties:

requeue-rejected 是否支持return； acknowledge-model=MANUAL (手动签收) ； recovery-interval 服务不稳定多少毫秒后进行重连 durable-subscription 是否启动持久化订阅； max-concurrenty=5 最大监听数
消费端监听消息代码：

这里如果换成Kafka，大部分代码也是不需要改变的。并且我们的发送端可以使用Kafka，接收端可以使用RabbitMQ这种非同种消息类型的消息。
发送消息测试：

Spring Cloud Stream 这里就相当于多了一个中间层，它将底层是Kafka或者RabbitMQ或者是其他的作为配置不侵入代码。消息的接收和发送都通过管道进行。在以后如果需要替换或者同时使用多种消息类型，都是可以的。

5.15 本章小结：

本章我们学习了Spring AMQP 的相关知识，通过学习，我们对RabbitMQ集成Spring有了一个深入的认识，这样为我们后续的学习、工作使用都打下了坚实的基础。最后我们整合了SpringBoot 与Spring Cloud Stream，更方便更高效的集成到我们的应用服务中去！

六. Rabbit的集群架构

6.1 本章导航

首先是了解RabbitMQ集群架构模式
接下来从零开始构建一个高可靠的RabbitMQ集群
集群的配置文件与集群运维故障、失败转移讲解
高级插件的使用

6.2 RabbitMQ集群架构模式

主备模式：实现RabbitMQ的高可用集群，一般在并发和数据量不高的情况下，这种模型非常的好用且简单。主备模式也称之为Warren模式；
- 所谓的rabbitMQ另外一种模式就是warren(兔子窝)，就是一个主/备方案（主节点如果挂了，从节点提供服务而已，和activeMQ利用zookeeper做主/备一样）
- 架构图示如下：
- HaProxy配置：（tcp级别的代理）
远程模式：远程模式
- 远距离通信和复制，所谓Shovel就是我们可以把消息进行不同数据中心的复制工作，我们可以跨地域的让两个mq集群互联。我们下面看一下Shovel架构模型；
- 图示：
  
  在使用了shovel插件后，模型变成了近端同步确认，远端异步确认的方式，大大提高了订单确认速度，并且还能保证可靠性。
- 细节图示：
  
  正常队列压力过大的时，会将订单复制到远端中心，在远端进行数据消费进行异步确认；用的不是特别多的原因是因为我们目前已经有了更好的远端模式，这个是比较早期的使用方式；
- Shovel集群的配置，首先启动rabbitMQ插件，命令如下：
```
rabbitmq-plugins enable amqp-client
1
```
```
rabbitmq-plugins enable rabbitmq_shovel
1
```
- 步骤：
  1. 创建rabbitmq.config:touch /etc/rabbitmq/rabbitmq.config
  2. 添加配置见rabbitmq.config
  3. 最后我们需要源服务器和目的服务器都使用相同的配置文件(rabbitmq.config)
  4. 配置文件如图：
    
    具体的可以百度。
镜像模式
1. 概述：
  - 集群模式非常经典的就是Mirror镜像模式，保证100%数据不丢失，在实际工作中也是用的最多的。并且实现集群非常的简单，一般互联网大厂都会构建这种镜像集群模式；
2. Mirror镜像队列，目的是为了保证rabbitmq数据的高可靠性解决方案，主要就是实现数据的同步，一般来讲是2-3个节点实现数据同步（对于100%数据可靠性解决方案一般是3节点）集群架构如下：
多活模式(比Shovel更好用)
1. 概述：
  - 这种模式也是实现异地数据复制的主流模式，因为Shovel模式配置比较复杂，所以一般来说实现异地集群都是使用这种双活或者多活模型来去实现的。这种模型需要依赖rabbitmq的federation插件，可以实现持续的可靠的AMQP数据通信，多活模式在实际配置与应用非常的简单。
2. RabbitMQ部署架构采用双中心（多中心），那么在两套（或多套）数据中心各部署一套RabbitMQ集群，各中心的RabbitMQ服务除了需要为业务提供正常的消息服务外，中心之间还需要实现部分队列消息共享。多活集群架构如下：
3. Federation插件进行互相复制；
4. Federation插件是一个不需要构建Cluster，而在Brokers之间传输消息的高性能插件，Federation插件可以在Brokers或者Cluster之间传输消息，连接的双方可以使用不同的users和virtual hosts,双方也可以使用版本不同的RabbitMQ和Erlang. Federation插件使用AMQP协议通讯，可以接受不连续的传输。
5. Federation Exchanges,可以看成Downstream从Upstream主动拉取消息，但并不是拉取所有消息，必须是在Downstream上已经明确定义Bindings关系的Exchange,也就是有实际的物理Queue来接收消息，才会从Upstream拉取消息到Downstream。使用AMQP协议实施代理间通信，Dwonstream会将绑定关系组合在一起，绑定/解除绑定命令将发送到Upstream交换机。因此，Federation Exchange只接收具有订阅的消息，本处贴出官方图来说明：

集群构建可参考百度

6.3 Haproxy

HAProxy是一款提供高可用性、负载均衡以及基于TCP(第四层)和HTTP(第七层)应用的代理软件。支持虚拟主机，它是免费、快速并且可靠的一种解决方案。HAProxy特别适用于那些负载特大的web站点，这些站点通常又需要会话保持或七层处理。HAProxy运行在时下的硬件上，完全可以支持数以万计的并发连接。并且它的运行模式使得它可以很简单安全的整合进您当前的架构中，同时可以保护你的web服务器不被暴露到网络上。
Haproxy性能最大化（一）
1. 如图所示：

实战安装：

下载依赖包：
```
yum install gcc vim wget
1
```

下载haproxy:

wget http://www.haproxy.org/download/1.6/src/haproxy-1.6.5.tar.gz
1

解压：

进入下载的软件位置：
```
cd /usr/local/software/
1
```

解压:

tar -zxvf haproxy-1.6.5.tar.gz -C /usr/local
1

进行操作：

进入目录
```
cd haproxy-1.6.5/
1
```

编译命令：

make TARGET=linux31 PREFIx=/usr/local/haproxy
1

安装命令：

make install PREFIX=/usr/local/haproxy
1

创建文件夹：
```
mkdir /etc/haproxy
1
```

赋权：(给予角色权限)

groupadd -r -g 149 haproxy
useradd -g haproxy -r -s /sbin/nologin -u 149 haproxy
1
2

创建haproxy配置文件
```
touch /etc/haproxy/haproxy.cfg
1
```
我们对代理等的配置，可以在这里面进行配置；通过 cd software/ 会看到一个haproxy.cfg ，我们可以通过命令:
mv haproxy.cfg /etc/haproxy/ 将配置移动然后 cd /etc/haproxy/

启动：

/usr/local/haproxy/sbin/haproxy -f /etc/haproxy/haproxy.cfg
1

查看haproxy进程状态：
```
ps -ef | grep haproxy
1
```
访问 haproxy
```
localhost:8100/rabbitmq-stas
1
```

6.4 KeepAlived

简介：
三个重要功能：
- 管理LVS负载均衡软件
- 实现LVS集群节点的健康检查中
- 作为系统网络服务的高可用性
KeepAlived高可用原理：
什么是VRRP?

具体的安装可以百度

集群配置文件：

6.5 集群恢复与故障转移

6.7 延迟插件的作用

延迟队列可以做什么事情？
比如消息的延迟推送、定时任务（消息）的执行。包括一些消息重试策略的配合使用，以及用于业务削峰限流、降级的异步延迟消息机制，都是延迟队列的实际应用场景。
安装：
1. 下载插件
2. 把下载好的文件放到指定目录
3. 启动插件
4. 访问地址：http://192.168.1.21:15672/#/exchanges

本章小结：本章我们掌握了RabbitMQ各种集群构建姿势，真正从零开始构建一个高可用的RabbitMQ集群，通过镜像队列+Haproxy+KeepAlived的架构进行构建！并且我们介绍了集群的节点宕机故障问题如何进行解决的5个解决方案！也学习了延迟插件的使用！

七. 互联网SET化架构

7.1 本章导航

了解SET化架构的进衍
互联网大长是如何进行SET化的
SET化架构的设计与解决方案
RabbitMQ SET化架构的搭建

7.2 BAT/TMD大厂单元化机构设计衍变之路（一）

1. 容灾问题：
  1. 核心服务（比如订单服务）会挂掉，会影响全网所有用户，导致整个业务不可用；
  2. 数据库主库集中在一个IDC，主机房挂掉，会影响全网所有用户，整个业务无法快速切换和恢复；
2. 资源扩展问题：
  1. 单IDC的资源（机器、网络带宽等）已经没法满足，扩展IDC时，存在跨机房访问时延问题（增加异地机房时，时延问题更加严重）；
  2. 数据库主库单点，连接数有限，不能支持应用程序的持续扩展；
3. 大集群拆分问题：
  1. 核心问题：分布式集群规模扩大后，会相应的带来资源扩展、大集群拆分以及容灾问题。
  2. 所以出于对业务扩展性以及容灾需求的考虑，我们需要一套从底层架构彻底解决问题的方案；

业界主流方案：单元化架构方案（阿里，支付宝，饿了么，微信等）

7.3 同城“双活”架构介绍

目前很多大型互联网公司的业务架构可以理解为同城“双活”架构，这里的“双活”是加引号的，具体可以这样理解：
1. 业务层面上已经做到真正的双活（或者多活），分别承担部分流量；
2. 存储层面比如定时任务、缓存、持久层、数据分析等都是主从架构，会有跨机房写；
3. 一个数据中心故障，可以手动切换流量，部分组件可以自动切换；

7.4 两地三中心架构介绍：

- 使用灾备的思想，在同城“双活”的基础上，在异地部署一套灾备数据中心，每个中心都具有完备的数据处理能力，只有当主节点故障需要容灾的时候才会紧急启动备用数据中心；
- 优缺点图示：
	![在这里插入图片描述](https://img-blog.csdnimg.cn/20191023214325480.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3MTI4MDQ5,size_16,color_FFFFFF,t_70)
1
2
3

7.5 SET化架构策略路

SET化架构设计：
- 流量路由：
  - 按照特殊的key（通常为userid）进行路由，判断某次请求该路由到中心集群还是单元化集群
- 中心集群：
  - 未进行单元化改造的服务（通常不在核心交易链路，比如供应链系统）被称为中心集群，跟当前架构保持一致；
- 单元化集群：
  - 每个单元化集群只负责本单元内的流量处理，以实现流量拆分以及故障隔离；
  - 每个单元化集群前期只存储本单元产生的交易数据，后续会做双向数据同步，实现容灾切换需求；
- 中间件（RPC、KV、MQ等）：
  - RPC：对于SET服务，调用封闭在SET内；对于非SET服务，沿用现有路由逻辑；
  - KV：支持分SET的数据生产和查询
  - MQ：支持分SET的消息生产和消费；
- 数据同步：
  - 全局数据（数据量小且变化不大，比如商家的菜品数据）部署在中心集群，其他单元化集群同步全局数据到本单元化内；
  - 未来演变为异地多活架构时，各单元化集群数据需要进行双向同步来实现容灾需要；
- SET化路由策略及其能力：
  - 异地容灾：
    - 通过SET化架构的流量调度能力，将SET分别部署在不同地区的数据中心，实现跨地区容灾支持；
- 高效的本地化服务：
  - 利用前端位置信息采集和域名解析策略，将流量路由到最近的SET，提供最高效的本地化服务；
  - 比如O2O场景天然具有本地生产，本地消费的特点，更加需要SET化支持；
SET化架构图：
SET化架构流转图：

7.6 SET化架构原则：

SET化重要的原则：
1. 对业务透明原则：
  - SET化架构的实现对业务代码透明，业务代码层面不需要关心SET化规则，SET部署等问题；
2. SET切分规则：
  1. 理论上，切分规则由业务层面按需定制
  2. 实现上，建议优先选最大的业务维度进行切分
  3. 比如海量用户的O2O业务，按用户位置信息进行切分。此外，接入层、逻辑层和数据层可以有独立的SET切分规则，有利于实现部署和运维成本的最优化；
3. 部署规范原则：
  1. 一个SET并不一定只限制在一个机房，也可以跨机房或者跨地区部署；为保证灵活性，单个SET内机器数不宜过多（如不超过1000台物理机）
SET消息中间件架构实现（RabbitMQ双活）：
- 集群同步插件：

使用此集群插件，发送一个消息到集群中，它可以进行转发复制到另外一个集群中；具体部署方式请百度；

SET配置规则：
- 第一：
- 第二：
- 第三：

本章小结：本章主要讲了互联网大长的SET化架构进衍，以及使用SET化架构能解决哪些问题，SET化架构的核心设计目标和重要原则，通过对RabbitMQ的SET化设计，也就是使用federation插件构建多活集群，实现多中心的数据同步！我们可以对大规模集群的部署有一个更可靠的解决方案！

七. 一线大厂的MQ组件实现思路和架构设计方案

7.1 本章导航

基础组件封装设计- 迅速消息发送
基础组件封装设计- 确认消息发送
基础组件封装设计- 批量消息发送
基础组件封装设计- 延迟消息发送
基础组件封装设计- 顺序消息发送
基础组件封装设计- 事务消息发送
消息幂等性保障- 消息路由规则架构设计

7.2

一线大厂的MQ组件实现思路和架构设计方案
MQ组件实现功能点（一）
1. 支持消息高性能的序列化转换、异步化发送消息
2. 支持消息生产实例与消费实例的连接池缓存化，提升性能
3. 支持可靠性投递消息，保障消息的100%不丢失
4. 支持消费端的幂等操作，避免消费端重复消息的问题；
MQ组件实现的功能点（二）
1. 支持迅速消息发送模式，在一些系统日志收集/统计分析等需求下可以保证高性能，超高吞吐量。
2. 支持延迟消息模式，消息可以延迟发送，指定延迟时间，用于某些延迟检查、服务限流场景
3. 支持事务消息，且100%保障可靠性投递，在金融行业单笔大金额操作时会有此类需求；
MQ组件实现功能点（三）
1. 支持顺序消息，保证消息送达消费端的前后顺序，例如下订单等复合性操作；
2. 支持消息补偿，重试，以及快速定位异常/失败消息
3. 支持集群消息负载均衡，保证消息落到具体SET集群的负载均衡
4. 支持消息路由策略，指定某些消息路由到指定的SET集群

7.3 消息发送模式- 迅速消息发送

迅速消息：
- 迅速消息是指消息不进行落库存储，不做可靠性的保障
- 在一些非核心消息、日志数据、或者统计分析等场景下比较合适
- 迅速消息的you的优点就是性能最高，吞吐量很大
图示：

不落库存储，不做消息的其他处理，直接拿到就消费；损失了可靠性，提高了性能

7.4 消息发送模式- 确认消息发送

图示：

7.5 消息发送模式- 批量消息发送

概述：
- 批量消息就是指我们把消息放到一个集合里进行统一提交
- 这种方案设计思路是期望消息在一个会话里，比如放到threadlocal里的集合，然后拥有相同的会话ID，并且带有这次提交消息的SIZE等相关属性，最重要的一点是要把这一批消息进行合并。
- 对于Channel而言，就是发送一次消息。这种方式也是希望消费端在消费的时候，可以进行批量化的消费，针对于某一个原子业务的操作去处理，但是不保障可靠性，需要进行补偿机制；
图示：

7.6 消息发送模式- 延迟消息发送

概述：延迟消息相对比较简单，就是我们在Message封装的时候加delayTime属性即可，使得我们的消息可以进行延迟发送，根据具体的业务场景也可以很好的使用到！
场景举例：
- 比如你在电商平台买到的商品签收后，不点击确认支付，那么系统自动会在7天（一定时间）去进行支付操作。
- 还有一些自动超时作废的场景，你的优惠券/红包有使用时间限制，也可以用延迟消息机制；

7.7 消息发送模式- 顺序消息（一）

顺序消息，比较类似于批量消息的实现机制，但是也有些不同。
我们要保障以下几点：
1. 发送的顺序消息，必须保障消息投递到同一个队列，且这个消费者只能有一个（独占模式）
2. 然后需要统一提交（可能是合并成一个大消息，也可能是拆分为多个消息），并且所有消息的会话ID一致
3. 添加消息属性：顺序标记的序号、和本次顺序消息的SIZE属性，进行落库操作
4. 并行进行发送给自身的延迟消息（注意带上关键属性：会话ID、SIZE）进行后续处理消费
5. 当收到延迟消息后，根据会话ID、SIZE抽取数据库数据进行处理即可。
6. 定时轮询补偿机制，对于异常情况：
  
  备注：比如生产端消息没有完全投递成功、或者消费端落库异常，导致消费端落库后缺少消息条目的情况

建议还是拆分成小消息，这样不用考虑顺序问题，多线程处理，然后接收到后再进行手动排序；

可靠性投递，就需要数据库；

7.8 消息发送模式- 事务消息发送(一)

事务消息，相对使用比较少见，但是也可能会有此方面的需求；如：单笔转账超过一个上限的时候，我们就希望这个消息优先级最高，并且可靠性要求达到100%，当然我们的系统和银行端系统都要兼顾才行，所以也会有一些补偿机制，主动发起银行端查询指令机制等。

7.9 消息发送模式- 事务消息发送（二）

为了保障性能的同时，也支持事务。我们并没有选择传统的RabbitMQ事务和Spring集成的机制，因为在性能测试的过程中，效果并不理想，非常消耗系统资源且会出现阻塞等情况，在高峰期也是一定程度上影响MQ集群的性能；
解决方案：
- 我们采用类似可靠性投递的机制，也就是补偿机制
- 但是我们的数据源必须是同一个，也就是业务操作DB1数据库和消息记录DB2数据库使用同一个数据源。
- 然后利用重写Spring DataSourceTransactionManager，在本地事务提交的时候进行发送消息，但是也有可能事务提交陈宫但是消息发送失败，这个时候就需要进行补偿了。

7.10 消息发送模式- 事务消息发送（五）

图示：

7.11 消息的幂等性的必要性

保障消息的幂等性，这也是我们在使用MQ中至关重要的环节。
可能导致消息出现非幂等的原因：
- 1. 可靠性消息投递机制
- 1. MQ Broker服务与消费端传输消息的过程中的网络抖动
- 1. 消费端故障或异常

7.12 消息的幂等性的设计

图示：

7.13 本章小结

本章主要和大家一起来探讨和交流互联网大厂的MQ基础组件服务的核心解决方案和架构设计思路，包含我们要实现哪些功能，比如各种消息的投递方式，以及消费的幂等性保障，通过本章的学习希望能够帮助大家开阔思路，更深入的理解MQ中间件在实际工作中的使用！

八. RabbitMQ课程总结

掌握MQ在各种场景下的应用
熟悉了解/深入探究主流消息中间件特点、架构、原理、底层机制
学习Rabbit基础原理和API级别使用
巧妙运用Rabbit高级特性服务于你的应用服务、架构设计
学会MQ与Spring体系整合，更高效的实现与应用服务的集成
构建高可靠性的MQ服务、搭建完备的平台化体系
紧密结合互联网大厂的架构设计思路，构建自己的架构设计思想
解决各种场景、需求下的MQ中间件设计与落地
封装基础组件，让别人站在你的肩膀上编码；

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/笔触狂放9/article/detail/126796

RabbitMQ消息中间件技术精讲全集

RabbitMQ消息中间件技术精讲

导航：

一. 主流消息中间件介绍

1.1 ActiveMQ

1.2 Kafka

1.3 RocketMQ

1.4 RabbitMQ

二. RabbitMQ核心概念及AMQP协议

2.1 本章导航

2.2 初始RabbitMQ

2.3 AMQP核心概念

2.4 RabbitMQ的整体架构？

2.5 消息流转图：

三. RabbitMQ的安装与使用

3.1 安装

3.2 安装过程

3.3 安装文档：

3.4 命令行与管控台–基础到高级操作

3.5 RabbitMQ可视化界面

3.6 急速入门-消息生产与消费

3.7 实战步骤：

3.8 Exchange交换机

3.9 交换机属性

3.10 Direct Exchange

3.11 Topic Exchange

3.12 Fanout Exchange

3.12 Binding - 绑定

3.13 Queue-- 消息队列

3.14 Message–消息

3.15 Virtual host-虚拟主机

四. RabbitMQ的高级特性

4.1 本章导航

4.2 消息如何保障100%的投递成功？

4.3 幂等性概念

4.4 Confirm 确认消息

4.5 Return消息机制

4.6 消费端自定义监听

4.7 消费端限流

4.8 消息ACK与重回队列

4.9 TTL队列/消息

4.10 死信队列（这个消息没有被任何人消费，它就变成了死信队列）

五. RabbitMQ高级整合应用

5.1 本章导航

5.2 RabbitMQ整合Spring AMQP实战

5.3 RabbitAdmin

5.4 Spring的整合

5.5 RabbitAdmin

5.6 SpringAMQP声明

5.7 消息模板 RabbitTemplate

5.8 简单消息监听容器：SimpleMessageListenerContainer

5.9 消息监听适配器：MessageListenerAdapter

5.10 MessageConverter消息转换器

5.11 SpringBoot整合配置详解

5.12 SpringBoot整合配置详解2

5.13 Spring Cloud Stream 整合

5.14 Spring Cloud Stream 实战

5.15 本章小结：

六. Rabbit的集群架构

6.1 本章导航

6.2 RabbitMQ集群架构模式

6.3 Haproxy

6.4 KeepAlived

6.5 集群恢复与故障转移

6.7 延迟插件的作用

七. 互联网SET化架构

7.1 本章导航

7.2 BAT/TMD大厂单元化机构设计衍变之路（一）

7.3 同城“双活”架构介绍

7.4 两地三中心架构介绍：

7.5 SET化架构策略路

7.6 SET化架构原则：

七. 一线大厂的MQ组件实现思路和架构设计方案

7.1 本章导航

7.2

7.3 消息发送模式- 迅速消息发送

7.4 消息发送模式- 确认消息发送

7.5 消息发送模式- 批量消息发送

7.6 消息发送模式- 延迟消息发送

7.7 消息发送模式- 顺序消息（一）