教程揭秘 | 动力节点内部Java零基础教学文档第十九篇：RocketMQ
接上期后续
本期分享第十九章节
RocketMQ
教学文档马上就分享完了，你们都跟上了吗？
每天都在学习嘛？
有什么不会的嘛？
今日教学文档分享来了
今日新篇章
【RocketMQ】
1. RocketMQ简介
官网： http://rocketmq.apache.org/
RocketMQ是阿里巴巴2016年MQ中间件，使用Java语言开发，RocketMQ 是一款开源的分布式消息系统，基于高可用分布式集群技术，提供低延时的、高可靠的消息发布与订阅服务。同时，广泛应用于多个领域，包括异步通信解耦、企业解决方案、金融支付、电信、电子商务、快递物流、广告营销、社交、即时通信、移动应用、手游、视频、物联网、车联网等。
具有以下特点：
1. l能够保证严格的消息顺序
2. l提供丰富的消息拉取模式
3. l高效的订阅者水平扩展能力
4. l实时的消息订阅机制
5. l亿级消息堆积能力
2. 为什么要使用MQ
1，要做到系统解耦，当新的模块进来时，可以做到代码改动最小; 能够解耦
2，设置流程缓冲池，可以让后端系统按自身吞吐能力进行消费，不被冲垮; 能够削峰，限流
3，强弱依赖梳理能把非关键调用链路的操作异步化并提升整体系统的吞吐能力;能够异步
Mq的作用 削峰限流 异步 解耦合
2.1 定义
中间件（缓存中间件 redis memcache 数据库中间件 mycat canal 消息中间件mq ）
面向消息的中间件(message-oriented middleware0) MOM能够很好的解决以上的问题。
是指利用高效可靠的消息传递机制进行与平台无关（跨平台）的数据交流，并基于数据通信来进行分布式系统的集成。
通过提供消息传递和消息排队模型在分布式环境下提供应用解耦，弹性伸缩，冗余存储，流量削峰，异步通信，数据同步等
大致流程
发送者把消息发给消息服务器，消息服务器把消息存放在若干队列/主题中，在合适的时候，消息服务器会把消息转发给接受者。在这个过程中，发送和接受是异步的,也就是发送无需等待，发送者和接受者的生命周期也没有必然关系在发布pub/订阅sub模式下，也可以完成一对多的通信，可以让一个消息有多个接受者[微信订阅号就是这样的]
2.2 特点
2.2.1 异步处理模式
消息发送者可以发送一个消息而无需等待响应。消息发送者把消息发送到一条虚拟的通道(主题或队列)上;
消息接收者则订阅或监听该通道。一条信息可能最终转发给一个或多个消息接收者，这些接收者都无需对消息发送者做出回应。整个过程都是异步的。
案例：
也就是说，一个系统和另一个系统间进行通信的时候，假如系统A希望发送一个消息给系统B，让它去处理，但是系统A不关注系统B到底怎么处理或者有没有处理好，所以系统A把消息发送给MQ，然后就不管这条消息的“死活” 了，接着系统B从MQ里面消费出来处理即可。至于怎么处理，是否处理完毕，什么时候处理，都是系统B的事，与系统A无关。
这样的一种通信方式，就是所谓的“异步”通信方式，对于系统A来说，只要把消息发给MQ,然后系统B就会异步处去进行处理了，系统A不能“同步”的等待系统B处理完。这样的好处是什么呢？解耦
2.2.2 应用系统的解耦
发送者和接收者不必了解对方，只需要确认消息
发送者和接收者不必同时在线
2.2.3 现实中的业务
3. 各个MQ产品的比较
4. RocketMQ重要概念【重点】
Producer：消息的发送者，生产者；举例：发件人
Consumer：消息接收者，消费者；举例：收件人
Broker：暂存和传输消息的通道；举例：快递
NameServer：管理Broker；举例：各个快递公司的管理机构 相当于broker的注册中心，保留了broker的信息
Queue：队列，消息存放的位置，一个Broker中可以有多个队列
Topic：主题，消息的分类
ProducerGroup：生产者组
ConsumerGroup：消费者组，多个消费者组可以同时消费一个主题的消息
消息发送的流程是，Producer询问NameServer，NameServer分配一个broker 然后Consumer也要询问NameServer，得到一个具体的broker，然后消费消息
5. 生产和消费理解【重点】
6. RocketMQ安装
了解了mq的基本概念和角色以后，我们开始安装rocketmq，建议在linux上
6.1 下载RocketMQ
下载地址：https://rocketmq.apache.org/dowloading/releases/
注意选择版本，这里我们选择4.9.2的版本，后面使用alibaba时对应
下载地址：
https://archive.apache.org/dist/rocketmq/4.9.2/rocketmq-all-4.9.2-bin-release.zip
6.2 上传服务器
在root目录下创建文件夹
mkdir rocketmq
将下载后的压缩包上传到阿里云服务器或者虚拟机中去
6.3 解压
unzip rocketmq-all-4.9.2-bin-release.zip
如果你的服务器没有unzip命令，则下载安装一个
yum install unzip
目录分析
Benchmark：包含一些性能测试的脚本；
Bin：可执行文件目录；
Conf：配置文件目录；
Lib：第三方依赖；
LICENSE：授权信息;
NOTICE：版本公告；
6.4 配置环境变量
vi /etc/profile
在文件末尾添加
export NAMESRV_ADDR=阿里云公网IP:9876
刷新环境变量
source /etc/profile
6.5 修改nameServer的运行脚本
进入bin目录下，修改runserver.sh文件,将71行和76行的Xms和Xmx等改小一点
vi runserver.sh
保存退出
6.6 修改broker的运行脚本
进入bin目录下，修改runbroker.sh文件,修改67行
保存退出
6.7 修改broker的配置文件
进入conf目录下，修改broker.conf文件
brokerClusterName = DefaultCluster
brokerName = broker-a
brokerId = 0
deleteWhen = 04
fileReservedTime = 48
brokerRole = ASYNC_MASTER
flushDiskType = ASYNC_FLUSH
namesrvAddr=localhost:9876
autoCreateTopicEnable=true
brokerIP1=阿里云公网IP
添加参数解释
namesrvAddr：nameSrv地址 可以写localhost因为nameSrv和broker在一个服务器
autoCreateTopicEnable：自动创建主题，不然需要手动创建出来
brokerIP1：broker也需要一个公网ip，如果不指定，那么是阿里云的内网地址，我们再本地无法连接使用
6.8 启动
首先在安装目录下创建一个logs文件夹，用于存放日志
mkdir logs
一次运行两条命令
启动nameSrv
nohup sh bin/mqnamesrv > ./logs/namesrv.log &
启动broker 这里的-c是指定使用的配置文件
nohup sh bin/mqbroker -c conf/broker.conf > ./logs/broker.log &
查看启动结果
6.9 RocketMQ控制台的安装RocketMQ-Console
Rocketmq 控制台可以可视化MQ的消息发送！
旧版本源码是在rocketmq-external里的rocketmq-console，新版本已经单独拆分成dashboard
网址： https://github.com/apache/rocketmq-dashboard
下载地址：
https://github.com/apache/rocketmq-dashboard/archive/refs/tags/rocketmq-dashboard-1.0.0.zip
下载后解压出来，在跟目录下执行
mvn clean package -Dmaven.test.skip=true
将jar包上传到服务器上去
然后运行
nohup java -jar ./rocketmq-dashboard-1.0.0.jar > ./rocketmq-4.9.2/logs/dashboard.log &
命令拓展:--server.port指定运行的端口
--rocketmq.config.namesrvAddr=127.0.0.1:9876 指定namesrv地址
访问： http://localhost:8001
运行访问端口是8001，如果从官网拉下来打包的话，默认端口是8080
nohup java -jar rocketmq-dashboard-1.0.0.jar --server.port=8081 --rocketmq.config.namesrvAddr=47.100.238.122:9876 > rocketmq-dashboard.log &
7. RocketMQ安装之docker
7.1 下载RockerMQ需要的镜像
docker pull rocketmqinc/rocketmq
docker pull styletang/rocketmq-console-ng
7.2 启动NameServer服务
7.2.1 创建NameServer数据存储路径
mkdir -p /home/rocketmq/data/namesrv/logs /home/rocketmq/data/namesrv/store
7.2.2 启动NameServer容器
docker run -d --name rmqnamesrv -p 9876:9876 -v /home/rocketmq/data/namesrv/logs:/root/logs -v /home/rocketmq/data/namesrv/store:/root/store -e "MAX_POSSIBLE_HEAP=100000000" rocketmqinc/rocketmq sh mqnamesrv
7.3 启动Broker服务
7.3.1 创建Broker数据存储路径
mkdir -p /home/rocketmq/data/broker/logs /home/rocketmq/data/broker/store
7.3.2 创建conf配置文件目录
mkdir /home/rocketmq/conf
7.3.3 在配置文件目录下创建broker.conf配置文件
7.3.4 启动Broker容器
7.4 启动控制台
7.5 正常启动后的docker ps
7.6 访问控制台
http://你的服务器外网ip:9999/
8. RocketMQ快速入门
RocketMQ提供了发送多种发送消息的模式，例如同步消息，异步消息，顺序消息，延迟消息，事务消息等，我们一一学习
8.1 消息发送和监听的流程
我们先搞清楚消息发送和监听的流程，然后我们在开始敲代码
8.1.1 消息生产者
1.创建消息生产者producer，并制定生产者组名
2.指定Nameserver地址
3.启动producer
4.创建消息对象，指定主题Topic、Tag和消息体等
5.发送消息
6.关闭生产者producer
8.1.2 消息消费者
1.创建消费者consumer，制定消费者组名
2.指定Nameserver地址
3.创建监听订阅主题Topic和Tag等
4.处理消息
5.启动消费者consumer
8.2 搭建Rocketmq-demo
8.2.1 加入依赖
8.2.2 编写生产者
8.2.3 编写消费者
8.2.4 测试
启动生产者和消费者进行测试
9. RocketMQ发送同步消息*
上面的快速入门就是发送同步消息，发送过后会有一个返回值，也就是mq服务器接收到消息后返回的一个确认，这种方式非常安全，但是性能上并没有这么高，而且在mq集群中，也是要等到所有的从机都复制了消息以后才会返回，所以针对重要的消息可以选择这种方式
10. RocketMQ发送异步消息*
异步消息通常用在对响应时间敏感的业务场景，即发送端不能容忍长时间地等待Broker的响应。发送完以后会有一个异步消息通知
10.1 异步消息生产者
10.2 异步消息消费者
11. RocketMQ发送单向消息*
这种方式主要用在不关心发送结果的场景，这种方式吞吐量很大，但是存在消息丢失的风险，例如日志信息的发送
11.1 单向消息生产者
11.2 单向消息消费者
消费者和上面一样
12. RocketMQ发送延迟消息*
消息放入mq后，过一段时间，才会被监听到，然后消费
比如下订单业务，提交了一个订单就可以发送一个延时消息，30min后去检查这个订单的状态，如果还是未付款就取消订单释放库存。
12.1 延迟消息生产者
12.2 延迟消息消费者
消费者和上面一样
这里注意的是RocketMQ不支持任意时间的延时
只支持以下几个固定的延时等级，等级1就对应1s，以此类推，最高支持2h延迟
private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
13. RocketMQ发送顺序消息**
消息有序指的是可以按照消息的发送顺序来消费(FIFO)。RocketMQ可以严格的保证消息有序，可以分为：分区有序或者全局有序。
可能大家会有疑问，mq不就是FIFO吗？
rocketMq的broker的机制，导致了rocketMq会有这个问题
因为一个broker中对应了四个queue
顺序消费的原理解析，在默认的情况下消息发送会采取Round Robin轮询方式把消息发送到不同的queue(分区队列)；而消费消息的时候从多个queue上拉取消息，这种情况发送和消费是不能保证顺序。但是如果控制发送的顺序消息只依次发送到同一个queue中，消费的时候只从这个queue上依次拉取，则就保证了顺序。当发送和消费参与的queue只有一个，则是全局有序；如果多个queue参与，则为分区有序，即相对每个queue，消息都是有序的。
下面用订单进行分区有序的示例。一个订单的顺序流程是：下订单、发短信通知、物流、签收。订单顺序号相同的消息会被先后发送到同一个队列中，消费时，同一个顺序获取到的肯定是同一个队列。
13.1 场景分析
模拟一个订单的发送流程，创建两个订单，发送的消息分别是
订单号111 消息流程 下订单->物流->签收
订单号112 消息流程 下订单->物流->拒收
13.2 创建一个订单对象
13.3 顺序消息生产者
13.4 顺序消息消费者，测试时等一会即可有延迟
14. RocketMQ发送批量消息
Rocketmq可以一次性发送一组消息，那么这一组消息会被当做一个消息消费
14.1 批量消息生产者
14.2 批量消息消费者
15. RocketMQ发送事务消息****
15.1 事务消息的发送流程
它可以被认为是一个两阶段的提交消息实现，以确保分布式系统的最终一致性。事务性消息确保本地事务的执行和消息的发送可以原子地执行。
上图说明了事务消息的大致方案，其中分为两个流程：正常事务消息的发送及提交、事务消息的补偿流程。
事务消息发送及提交
1. 发送消息（half消息）。
2. 服务端响应消息写入结果。
3. 根据发送结果执行本地事务（如果写入失败，此时half消息对业务不可见，本地逻辑不执行）。
4. 根据本地事务状态执行Commit或Rollback（Commit操作生成消息索引，消息对消费者可见）
事务补偿
1. 对没有Commit/Rollback的事务消息（pending状态的消息），从服务端发起一次“回查”
2. Producer收到回查消息，检查回查消息对应的本地事务的状态
3. 根据本地事务状态，重新Commit或者Rollback
其中，补偿阶段用于解决消息UNKNOW或者Rollback发生超时或者失败的情况。
事务消息状态
事务消息共有三种状态，提交状态、回滚状态、中间状态：
l TransactionStatus.CommitTransaction: 提交事务，它允许消费者消费此消息。
l TransactionStatus.RollbackTransaction: 回滚事务，它代表该消息将被删除，不允许被消费。
l TransactionStatus.Unknown: 中间状态，它代表需要检查消息队列来确定状态。
15.2 事务消息生产者
15.3 事务消息消费者
15.4 测试结果
16. RocketMQ发送带标签的消息，消息过滤
Rocketmq提供消息过滤功能，通过tag或者key进行区分
我们往一个主题里面发送消息的时候，根据业务逻辑，可能需要区分，比如带有tagA标签的被A消费，带有tagB标签的被B消费，还有在事务监听的类里面，只要是事务消息都要走同一个监听，我们也需要通过过滤才区别对待
16.1 标签消息生产者
16.2 标签消息消费者
17. RocketMQ中消息的Key(业务相关)
在rocketmq中的消息，默认会有一个messageId当做消息的唯一标识，我们也可以给消息携带一个key，用作唯一标识或者业务标识，包括在控制面板查询的时候也可以使用messageId或者key来进行查询
17.1 带key消息生产者
17.2 带key消息消费者
18. RocketMQ重试机制
18.1 生产者重试
// 失败的情况重发3次
producer.setRetryTimesWhenSendFailed(3);
// 消息在1S内没有发送成功，就会重试
producer.send(msg, 1000);
18.2 消费者重试
在消费者放return ConsumeConcurrentlyStatus.RECONSUME_LATER;后就会执行重试
上图代码中说明了，我们再实际生产过程中，一般重试5-7次，如果还没有消费成功，则可以把消息签收了，通知人工等处理
messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
19. RocketMQ死信消息
当消费重试到达阈值以后，消息不会被投递给消费者了，而是进入了死信队列
19.1 消息生产者
19.2 消息消费者
19.3 信消费者
注意权限问题 perm 2读 4写 6读写
19.4 控制台显示
由于字数限制本文只分享了一半哦
完整版获取可私信小动~
更多干货我们下期再说！
下期会分享
第二十章节
秒杀
相关知识~
下期见！
