Java|525、Java工程师的进阶之路 -【 RocketMQ （二）】 2022.01.06 java|大数据|Java框架

- 1. RocketMQ 设计目的
- - 1.1. 发布/订阅
  - 1.2. 消息优先级
  - 1.3. 消息顺序
  - 1.4. 消息过滤
  - 1.5. 消息持久化
  - 1.6. 消息可靠性
  - 1.7. 消息实时性
  - 1.8. 保证至少一次
  - 1.9. 保证只有一次
  - 1.10. Broker的Buffer溢出
  - 1.11. 回溯消费
  - 1.12. 消息堆积
  - 1.13. 分布式事务
  - 1.14. 定时消息
  - 1.15. 消息重试
- 2. RocketMQ 消费模型
- 3. RocketMQ 网络模型
- 4. RocketMQ 存储模型
- 5、RocketMQ 高可用性
- 6. RocketMQ 定时/延时消息
- 7. RocketMQ 事务消息
- 8、参考链接

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

Java|525、Java工程师的进阶之路 -【 RocketMQ （二）】 2022.01.06

文章图片

1.2. 消息优先级
规范中描述的优先级是指在一个消息队列中，每条消息都有不同的优先级，一般用整数来描述，优先级高的消息先投递，如果消息完全在一个内存队列中，那么在投递前可以按照优先级排序，令优先级高的先投递。
由于RocketMQ所有消息都是持久化的，所以如果按照优先级来排序，开销会非常大，因此RocketMQ没有特意支持消息优先级，但是可以通过变通的方式实现类似功能，即单独配置一个优先级高的队列，和一个普通优先级的队列，将不同优先级发送到不同队列即可。
对于优先级问题，可以归纳为2类：

只要达到优先级目的即可，不是严格意义上的优先级，通常将优先级划分为高、中、低，或者再多几个级别。每个优先级可以用不同的topic表示，发消息时，指定不同的topic来表示优先级，这种方式可以解决绝大部分的优先级问题，但是对业务的优先级精确性做了妥协。
严格的优先级，优先级用整数表示，例如0 ~ 65535，这种优先级问题一般使用不同topic解决就非常不合适。如果要让MQ解决此问题，会对MQ的性能造成非常大的影响。这里要确保一点，业务上是否确实需要这种严格的优先级，如果将优先级压缩成几个，对业务的影响有多大？

1.3. 消息顺序
消息有序指的是一类消息消费时，能按照发送的顺序来消费。例如：一个订单产生了3条消息，分别是订单创建，订单付款，订单完成。消费时，要按照这个顺序消费才能有意义。但是同时订单之间是可以并行消费的。
RocketMQ可以严格的保证消息有序。
1.4. 消息过滤
Broker端消息过滤
在Broker中，按照Consumer的要求做过滤，优点是减少了对于Consumer无用消息的网络传输。
缺点是增加了Broker的负担，实现相对复杂。

淘宝Notify支持多种过滤方式，包含直接按照消息类型过滤，灵活的语法表达式过滤，几乎可以满足最苛刻的过滤需求。
淘宝RocketMQ支持按照简单的Message Tag过滤，也支持按照Message Header、body进行过滤。
CORBA Notification规范中也支持灵活的语法表达式过滤。

Consumer端消息过滤
这种过滤方式可由应用完全自定义实现，但是缺点是很多无用的消息要传输到Consumer端。

文章图片

1.5. 消息持久化
消息中间件通常采用的几种持久化方式：

持久化到数据库，例如Mysql。
持久化到KV存储，例如levelDB、伯克利DB等KV存储系统。
文件记录形式持久化，例如Kafka，RocketMQ
对内存数据做一个持久化镜像，例如beanstalkd，VisiNotify
(1)、(2)、(3)三种持久化方式都具有将内存队列Buffer进行扩展的能力，(4)只是一个内存的镜像，作用是当Broker挂掉重启后仍然能将之前内存的数据恢复出来。

JMS与CORBA Notification规范没有明确说明如何持久化，但是持久化部分的性能直接决定了整个消息中间件的性能。
RocketMQ充分利用Linux文件系统内存cache来提高性能。

文章图片

1.6. 消息可靠性
影响消息可靠性的几种情况：

Broker正常关闭
Broker异常Crash
OS Crash
机器掉电，但是能立即恢复供电情况。
机器无法开机（可能是cpu、主板、内存等关键设备损坏）
磁盘设备损坏。

(1)、(2)、(3)、(4)四种情况都属于硬件资源可立即恢复情况，RocketMQ在这四种情况下能保证消息不丢，或者丢失少量数据（依赖刷盘方式是同步还是异步）。
(5)、(6)属于单点故障，且无法恢复，一旦发生，在此单点上的消息全部丢失。RocketMQ在这两种情况下，通过异步复制，可保证99%的消息不丢，但是仍然会有极少量的消息可能丢失。通过同步双写技术可以完全避免单点，同步双写势必会影响性能，适合对消息可靠性要求极高的场合，例如与Money相关的应用。
RocketMQ从3.0版本开始支持同步双写。
1.7. 消息实时性
在消息不堆积情况下，消息到达Broker后，能立刻到达Consumer。
RocketMQ使用长轮询Pull方式，可保证消息非常实时，消息实时性不低于Push。
1.8. 保证至少一次
是指每个消息必须投递一次。
RocketMQ Consumer先pull消息到本地，消费完成后，才向服务器返回ack，如果没有消费一定不会ack消息，所以RocketMQ可以很好的支持此特性。
1.9. 保证只有一次

发送消息阶段，不允许发送重复的消息。
消费消息阶段，不允许消费重复的消息。

只有以上两个条件都满足情况下，才能认为消息是“Exactly Only Once”，而要实现以上两点，在分布式系统环境下，不可避免要产生巨大的开销。所以RocketMQ为了追求高性能，并不保证此特性，要求在业务上进行去重，也就是说消费消息要做到幂等性。
RocketMQ虽然不能严格保证不重复，但是正常情况下很少会出现重复发送、消费情况，只有网络异常，Consumer启停等异常情况下会出现消息重复。
1.10. Broker的Buffer溢出
Broker的Buffer通常指的是Broker中一个队列的内存Buffer大小，这类Buffer通常大小有限，如果Buffer满了以后怎么办？
下面是CORBA Notification规范中处理方式：

拒绝新来的消息，向Producer返回RejectNewEvents错误码。
按照特定策略丢弃已有消息:

AnyOrder - Any event may be discarded on overflow. This is the default setting for this property.
FifoOrder - The first event received will be the first discarded.
LifoOrder - The last event received will be the first discarded.
PriorityOrder - Events should be discarded in priority order, such that lower priority events will be discarded before higher priority events.
DeadlineOrder - Events should be discarded in the order of shortest expiry deadline first.

RocketMQ没有内存Buffer概念，RocketMQ的队列都是持久化磁盘，数据定期清除。
对于此问题的解决思路，RocketMQ同其他MQ有非常显著的区别，RocketMQ的内存Buffer抽象成一个无限长度的队列，不管有多少数据进来都能装得下，这个无限是有前提的，Broker会定期删除过期的数据，例如Broker只保存3天的消息，那么这个Buffer虽然长度无限，但是3天前的数据会被从队尾删除。
此问题的本质原因是网络调用存在不确定性，即既不成功也不失败的第三种状态，所以才产生了消息重复性问题。
1.11. 回溯消费
回溯消费是指Consumer已经消费成功的消息，由于业务上需求需要重新消费，要支持此功能，Broker在向Consumer投递成功消息后，消息仍然需要保留。并且重新消费一般是按照时间维度，例如由于Consumer系统故障，恢复后需要重新消费1小时前的数据，那么Broker要提供一种机制，可以按照时间维度来回退消费进度。
RocketMQ支持按照时间回溯消费，时间维度精确到毫秒，可以向前回溯，也可以向后回溯。
1.12. 消息堆积
消息中间件的主要功能是异步解耦，还有个重要功能是挡住前端的数据洪峰，保证后端系统的稳定性，这就要求消息中间件具有一定的消息堆积能力，消息堆积分以下两种情况：

消息堆积在内存Buffer，一旦超过内存Buffer，可以根据一定的丢弃策略来丢弃消息，如CORBA Notification规范中描述。适合能容忍丢弃消息的业务，这种情况消息的堆积能力主要在于内存Buffer大小，而且消息堆积后，性能下降不会太大，因为内存中数据多少对于对外提供的访问能力影响有限。
消息堆积到持久化存储系统中，例如DB，KV存储，文件记录形式。当消息不能在内存Cache命中时，要不可避免的访问磁盘，会产生大量读IO，读IO的吞吐量直接决定了消息堆积后的访问能力。

评估消息堆积能力主要有以下四点：

消息能堆积多少条，多少字节？即消息的堆积容量。
消息堆积后，发消息的吞吐量大小，是否会受堆积影响？
消息堆积后，正常消费的Consumer是否会受影响？
消息堆积后，访问堆积在磁盘的消息时，吞吐量有多大？

1.13. 分布式事务
已知的几个分布式事务规范，如XA，JTA等。其中XA规范被各大数据库厂商广泛支持，如Oracle，Mysql等。其中XA的TM实现佼佼者如Oracle Tuxedo，在金融、电信等领域被广泛应用。
分布式事务涉及到两阶段提交问题，在数据存储方面的方面必然需要KV存储的支持，因为第二阶段的提交回滚需要修改消息状态，一定涉及到根据Key去查找Message的动作。RocketMQ在第二阶段绕过了根据Key去查找Message的问题，采用第一阶段发送Prepared消息时，拿到了消息的Offset，第二阶段通过Offset去访问消息，并修改状态，Offset就是数据的地址。
RocketMQ这种实现事务方式，没有通过KV存储做，而是通过Offset方式，存在一个显著缺陷，即通过Offset更改数据，会令系统的脏页过多，需要特别关注。
1.14. 定时消息
定时消息是指消息发到Broker后，不能立刻被Consumer消费，要到特定的时间点或者等待特定的时间后才能被消费。
如果要支持任意的时间精度，在Broker层面，必须要做消息排序，如果再涉及到持久化，那么消息排序要不可避免的产生巨大性能开销。
RocketMQ支持定时消息，但是不支持任意时间精度，支持特定的level，例如定时5s，10s，1m等。
1.15. 消息重试
【Java|525、Java工程师的进阶之路 -【 RocketMQ （二）】 2022.01.06】Consumer消费消息失败后，要提供一种重试机制，令消息再消费一次。
Consumer消费消息失败通常可以认为有以下几种情况：

由于消息本身的原因，例如反序列化失败，消息数据本身无法处理（例如话费充值，当前消息的手机号被注销，无法充值）等。这种错误通常需要跳过这条消息，再消费其他消息，而这条失败的消息即使立刻重试消费，99%也不成功，所以最好提供一种定时重试机制，即过10s秒后再重试。
由于依赖的下游应用服务不可用，例如db连接不可用，外系统网络不可达等。遇到这种错误，即使跳过当前失败的消息，消费其他消息同样也会报错。这种情况建议应用sleep 30s，再消费下一条消息，这样可以减轻Broker重试消息的压力。

2. RocketMQ 消费模型一般来说消息队列的消费模型分为两种，基于推送的消息 (push) 模型和基于拉取 (poll) 的消息模型。
基于推送模型的消息系统，由消息代理记录消费状态。消息代理将消息推送到消费者后，标记这条消息为已经被消费，但是这种方式无法很好地保证消费的处理语义。比如当我们把已经把消息发送给消费者之后，由于消费进程挂掉或者由于网络原因没有收到这条消息，如果我们在消费代理将其标记为已消费，这个消息就永久丢失了。如果我们利用生产者收到消息后回复这种方法，消息代理需要记录消费状态，这种不可取。
用过RocketMQ的同学肯定不禁会想到，在RocketMQ中不是提供了两种消费者吗？
2.1 MQPullConsumer 和 MQPushConsumer
其中 MQPushConsumer 不就是我们的推模型吗？其实这两种模型都是客户端主动去拉消息，其中的实现区别如下：
MQPullConsumer：每次拉取消息需要传入拉取消息的 offset 和每次拉取多少消息量，具体拉取哪里的消息，拉取多少是由客户端控制。
MQPushConsumer：同样也是客户端主动拉取消息，但是消息进度是由服务端保存，Consumer 会定时上报自己消费到哪里，所以 Consumer 下次消费的时候是可以找到上次消费的点，一般来说使用PushConsumer 我们不需要关心 offset 和拉取多少数据，直接使用即可。
2.2 集群消费和广播消费
消费模式我们分为两种，集群消费，广播消费:
集群消费: 同一个 GroupId 都属于一个集群，一般来说一条消息只会被任意一个消费者处理。
广播消费：广播消费的消息会被集群中所有消费者进行消息，但是要注意一下因为广播消费的 offset 在服务端保存成本太高，所以客户端每一次重启都会从最新消息消费，而不是上次保存的 offset。
3. RocketMQ 网络模型在 Kafka 中使用的原生的 socket 实现网络通信，而RocketMQ 使用的是 Netty 网络框架，现在越来越多的中间件都不会直接选择原生的 socket，而是使用的 Netty 框架，主要得益于下面几个原因:
3.1 API 使用简单，不需要关心过多的网络细节，更专注于中间件逻辑。
3.2 性能高。成熟稳定，jdk nio 的 bug 都被修复了。
选择框架是一方面，而想要保证网络通信的高效，网络线程模型也是一方面，我们常见的有 1+N (1 个 Acceptor 线程，N 个 IO 线程)，1+N+M (1个 acceptor 线程，N 个 IO 线程，M 个 worker线程)等模型，RocketMQ 使用的是 1+N1+N2+M 的模型，如下图所示：

文章图片

1个 acceptor 线程，N1 个 IO 线程，N2 个线程用来做 Shake-hand, SSL 验证, 编解码; M 个线程用来做业务处理。这样的好处将编解码，和 SSL 验证等一些可能耗时的操作放在了一个单独的线程池，不会占据我们业务线程和 IO 线程。
4. RocketMQ 存储模型做为一个好的消息系统，高性能的存储，高可用都不可少。
RocketMQ 和 Kafka 的存储核心设计有很大的不同，所以其在写入性能方面也有很大的差别，这是16年阿里中间件团队对 RocketMQ 和 Kafka 不同 Topic 下做的性能测试:

文章图片

从图上可以看出：

Kafka 在 Topic 数量由 64 增长到 256 时，吞吐量下降了 98.37%。
RocketMQ 在 Topic 数量由 64 增长到 256 时，吞吐量只下降了 16%。

这是为什么呢？kafka 一个 topic 下面的所有消息都是以 partition 的方式分布式的存储在多个节点上。同时在 kafka 的机器上，每个 Partition 其实都会对应一个日志目录，在目录下面会对应多个日志分段。所以如果 Topic 很多的时候 Kafka 虽然写文件是顺序写，但实际上文件过多，会造成磁盘 IO 竞争非常激烈。
那 RocketMQ 为什么在多 Topic 的情况下，依然还能很好的保持较多的吞吐量呢？我们首先来看一下RocketMQ 中比较关键的文件:
commitLog：消息主体以及元数据的存储主体，存储 Producer 端写入的消息主体内容, 消息内容不是定长的。单个文件大小默认 1G ，文件名长度为 20 位，左边补零，剩余为起始偏移量，比如00000000000000000000 代表了第一个文件，起始偏移量为 0，文件大小为 1G=1073741824；当第一个文件写满了，第二个文件为 00000000001073741824，起始偏移量为 1073741824，以此类推。消息主要是顺序写入日志文件，当文件满了，写入下一个文件；
config：保存一些配置信息，包括一些 Group，Topic 以及 Consumer 消费 offset 等信息。
consumeQueue：消息消费队列，引入的目的主要是提高消息消费的性能，由于 RocketMQ 是基于主题topic 的订阅模式，消息消费是针对主题进行的，如果要遍历 commitlog 文件中根据 topic 检索消息是非常低效的。

文章图片

Consumer即可根据ConsumeQueue来查找待消费的消息。其中，ConsumeQueue（逻辑消费队列）作为消费消息的索引，保存了指定Topic下的队列消息在 commitLog 中的起始物理偏移量 offset，消息大小 size 和消息 Tag 的 HashCode 值。
consumequeue 文件可以看成是基于 topic 的 commitlog 索引文件，故 consumequeue 文件夹的组织方式如下：topic/queue/file 三层组织结构，具体存储路径为：HOME \store\index${fileName}，文件名fileName 是以创建时的时间戳命名的，固定的单个 IndexFile 文件大小约为 400M，一个 IndexFile 可以保存 2000W 个索引，IndexFile 的底层存储设计为在文件系统中实现 HashMap 结构，故 rocketmq 的索引文件其底层实现为 hash 索引。
我们发现我们的消息主体数据并没有像 Kafka 一样写入多个文件，而是写入一个文件，这样我们的写入 IO 竞争就非常小，可以在很多 Topic 的时候依然保持很高的吞吐量。有同学说这里的 ConsumeQueue 写是在不停的写入呢，并且 ConsumeQueue 是以 Queue 维度来创建文件，那么文件数量依然很多，在这里 ConsumeQueue 的写入的数据量很小，每条消息只有 20 个字节，30W 条数据也才 6M 左右，所以其实对我们的影响相对 Kafka 的 Topic 之间影响是要小很多的。我们整个的逻辑可以如下：
Producer 不断的再往 CommitLog 添加新的消息，有一个定时任务 ReputService 会不断的扫描新添加进来的 CommitLog，然后不断的去构建 ConsumerQueue 和 Index。

文章图片

注意：这里指的都是普通的硬盘，在 SSD 上面多个文件并发写入和单个文件写入影响不大。
Kafka 中每个 Partition 都会是一个单独的文件，所以当消费某个消息的时候，会很好的出现顺序读，我们知道 OS 从物理磁盘上访问读取文件的同时，会顺序对其他相邻块的数据文件进行预读取，将数据放入PageCache，所以 Kafka 的读取消息性能比较好。
RocketMQ读取流程如下：

先读取 ConsumerQueue 中的 offset 对应 CommitLog 物理的 offset
根据 offset 读取 CommitLog

ConsumerQueue 也是每个 Queue 一个单独的文件，并且其文件体积小，所以很容易利用 PageCache 提高性能。而 CommitLog，由于同一个 Queue 的连续消息在 CommitLog 其实是不连续的，所以会造成随机读， RocketMQ 对此做了几个优化：

Mmap 映射读取，Mmap 的方式减少了传统IO将磁盘文件数据在操作系统内核地址空间的缓冲区和用户应用程序地址空间的缓冲区之间来回进行拷贝的性能开销
使用 DeadLine 调度算法 + SSD 存储盘

由于 Mmap 映射受到内存限制，当不在 Mmmap 映射这部分数据的时候(也就是消息堆积过多)，默认是内存的40%，会将请求发送到 SLAVE, 减缓 Master 的压力。
5、RocketMQ 高可用性集群模式
我们首先需要选择一种集群模式，来适应我们可忍耐的可用程度，一般来说分为三种：

单 Master: 这种模式，可用性最低，但是成本也是最低，一旦宕机，所有都不可用。这种一般只适用于本地测试。
单 Master 多 Slave: 这种模式，可用性一般，如果主宕机，那么所有写入都不可用，读取依然可用，如果 master 磁盘损坏，可以依赖 slave 的数据。
多 Master: 这种模式，可用性一般，如果出现部分 master 宕机，那么这部分 master 上的消息都不可消费，也不可写数据，如果一个 Topic 的队列在多个 master 上都有，那么可以保证没有宕机的那部分可以正常消费，写入。如果 master 的磁盘损坏会导致消息丢失。
多 Master 多 Slave：这种模式，可用性最高，但是维护成本也最高，当 master 宕机了之后，只会出现在这部分 master 上的队列不可写入，但是读取依然是可以的，并且如果 master 磁盘损坏，可以依赖 slave 的数据。

一般来说投入生产环境的话都会选择第四种，来保证最高的可用性。
消息的可用性
当我们选择好了集群模式之后，那么我们需要关心的就是怎么去存储和复制这个数据，RocketMQ 对消息的刷盘提供了同步和异步的策略来满足我们的，当我们选择同步刷盘之后，如果刷盘超时会给返回 FLUSH_DISK_TIMEOUT，如果是异步刷盘不会返回刷盘相关信息，选择同步刷盘可以尽最大程度满足我们的消息不会丢失。
除了存储有选择之后，我们的主从同步提供了同步和异步两种模式来进行复制，当然选择同步可以提升可用性，但是消息的发送 RT 时间会下降 10% 左右。
Dleger-RocketMQ
我们上面对于 master-slave 部署模式已经做了很多分析，我们发现，当 master 出现问题的时候，我们的写入怎么都会不可用，除非恢复 master，或者手动将我们的 slave 切换成 master，导致了我们的 slave 在多数情况下只有读取的作用。RocketMQ 在最近的几个版本中推出了 Dleger-RocketMQ，使用 Raft 协议复制 CommitLog，并且自动进行选主，这样 master 宕机的时候，写入依然保持可用。
6. RocketMQ 定时/延时消息定时消息和延时消息在实际业务场景中使用的比较多，比如下面的一些场景：

订单超时未支付自动关闭，因为在很多场景中下单之后库存就被锁定了，这里需要将其进行超时关闭。
需要一些延时的操作，比如一些兜底的逻辑，当做完某个逻辑之后，可以发送延时消息比如延时半个小时，进行兜底检查补偿。
在某个时间给用户发送消息，同样也可以使用延时消息。

在开源版本的 RocketMQ 中延时消息并不支持任意时间的延时，需要设置几个固定的延时等级，目前默认设置为：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h，从 1s 到 2h 分别对应着等级 1到 18，而阿里云中的版本(收费)是可以支持40天内的任何时刻（毫秒级别）。我们先看下在RocketMQ中定时任务原理图：

文章图片

Step1：Producer 在自己发送的消息上设置好需要延时的级别。
Step2: Broker 发现此消息是延时消息，将 Topic 进行替换成延时 Topic，每个延时级别都会作为一个单独的 queue，将自己的 Topic 作为额外信息存储。
Step3: 构建 ConsumerQueue
Step4: 定时任务定时扫描每个延时级别的 ConsumerQueue。
Step5: 拿到 ConsumerQueue 中的 CommitLog 的 Offset，获取消息，判断是否已经达到执行时间
Step6: 如果达到，那么将消息的 Topic 恢复，进行重新投递。如果没有达到则延迟没有达到的这段时间执行任务。
可以看见延时消息是利用新建单独的 Topic 和 Queue 来实现的，如果我们要实现 40 天之内的任意时间度，基于这种方案，那么需要 402460601000 个 queue，这样的成本是非常之高的，那阿里云上面的支持任意时间是怎么实现的呢？这里猜测是持久化二级 TimeWheel 时间轮，二级时间轮用于替代我们的 ConsumeQueue，保存 Commitlog-Offset，然后通过时间轮不断的取出当前已经到了的时间，然后再次投递消息。
7. RocketMQ 事务消息事务消息同样的也是 RocketMQ 中的一大特色，其可以帮助我们完成分布式事务的最终一致性:
加粗样式

文章图片

具体使用事务消息步骤如下：
Step1: 调用 sendMessageInTransaction 发送事务消息。
Step2: 如果发送成功，则执行本地事务。
Step3: 如果执行本地事务成功则发送 commit，如果失败则发送 rollback。
Step4: 如果其中某个阶段比如 commit 发送失败，rocketMQ 会进行定时从 Broker 回查，本地事务的状态。
事务消息的使用整个流程相对之前几种消息使用比较复杂，下面是事务消息实现的原理图：

文章图片

Step1: 发送事务消息,这里也叫做 halfMessage，会将 Topic 替换为 HalfMessage 的 Topic。
Step2: 发送 commit 或者 rollback，如果是 commit 这里会查询出之前的消息，然后将消息复原成原 Topic，并且发送一个 OpMessage 用于记录当前消息可以删除。如果是 rollback 这里会直接发送一个 OpMessage 删除。
Step3: 在 Broker 有个处理事务消息的定时任务，定时对比 halfMessage 和 OpMessage, 如果有OpMessage 且状态为删除，那么该条消息必定 commit 或者 rollback，所以就可以删除这条消息。
Step4: 如果事务超时（默认是6s），还没有 opMessage，那么很有可能 commit 信息丢了，这里会去反查我们的 Producer 本地事务状态。
Step5: 根据查询出来的信息做 Step2。
我们发现 RocketMQ 实现事务消息也是通过修改原 Topic 信息，和延迟消息一样，然后模拟成消费者进行消费，做一些特殊的业务逻辑。当然我们还可以利用这种方式去做 RocketMQ 更多的扩展。
8、参考链接 [01] Java工程师的进阶之路 -【 RocketMQ （二）】