rocket 阿里RocketMQ如何解决消息的顺序&重复两大硬伤？

分布式消息系统作为实现分布式系统可扩展性的关键组件，需要高吞吐量和高可用性。谈到消息系统的设计，我们不能避免两个问题:

消息顺序

信息的重复

RocketMQ作为阿里开源的高性能、高吞吐量的消息中间件，是如何解决这两个问题的？RocketMQ的关键特性是什么？它的实现原理是什么？

关键特征及其实施原则

一.顺序信息

消息排序意味着消息可以按照发送的顺序进行消费。例如，订单生成三条消息，即订单创建、订单支付和订单完成。消费的时候，按顺序消费才有意义。同时可以并行消费多个订单。让我们从下面的例子开始:

如果制作人产生两条消息:M1和M2，应该怎么做才能保证这两条消息的顺序？你可能会这样想:

消耗队列文件组织，如图:

用户文件存储单元的格式

提交日志偏移量是指提交日志文件中该消息的实际偏移量

大小存储中消息的大小

消息Tag HashCode存储消息的Tag的哈希值:主要用于订阅时的消息过滤(如果订阅时指定了Tag，订阅的消息会根据HashCode快速找到)

2、提交日志

CommitLog:存储消息的物理文件。每个代理上commitlog由该机器的所有队列共享，没有任何区别。

文件的默认位置如下，仍然可以通过配置文件进行修改:

${user.home}商店${commitlog}${fileName}

CommitLog的消息存储单元长度不固定，文件按顺序写入，随机读取。报文的存储结构如下表所示，按照编号顺序和编号对应的内容依次存储。

提交日志存储单元结构图

3.消息存储实现

消息存储的实现比较复杂，值得深入了解。以后再单独分析(目前正在收集资料)。本节仅用代码解释具体过程。

4.消息的索引文件

如果消息包含键值，索引文件将用于存储消息索引。文件的内容结构如下:

消息索引

索引文件主要用于按关键字查询消息，流程主要包括:

根据被查询键的hashcode % slotNum(slotNum是一个索引文件中包含的最大槽数，例如，如图所示slotNum=5000000)

根据槽值(对应于槽位置的值)，找到索引项列表中的最后一项(以相反的顺序，槽值总是指向最新的索引项)

遍历索引项列表以返回查询时间范围内的结果集(默认情况下，一次返回的最大记录数为32)

不及物动词消息订阅

RocketMQ消息订阅有两种模式，一种是Push模式，即MQServer主动推送给消费者；另一种是Pull模式，即消费者在需要时主动从MQServer中拉取。但是在具体实现中，Push和Pull两种模式都是由消费者主动拉动的。

首先看看消费者端的负载平衡:

消费者负载平衡

消费者将通过重新平衡服务线程每10秒加载一次基于主题的所有队列:

遍历消费者下的所有主题，然后根据主题订阅所有消息

将所有消费者归入同一个主题和消费者组

然后根据具体的分配策略分配消费队列，具体包括:平均分配、消费者配置等

如上图所示，如果有5个队列和2个消费者，第一个消费者消费3个队列，第二个消费者消费2个队列。这里采用平均分配策略，类似于分页过程。TOPIC下的所有队列都是记录，Consumers的数量相当于总页数，所以每页的记录数类似于一个consumer将要消耗的队列。

通过这样的策略来实现一般的平均消费，这样的设计也可以扩大消费者的消费能力来提高消费能力。

用户的推送模式通过长轮询模式实现，如下图所示:

推送模式示意图

每隔一段时间，消费者主动向代理发送一个消息拉请求。收到请求后，如果有消息，代理会立即返回数据。接收到返回的消息后，消费者回调由消费者设置的侦听器方法。如果当代理接收拉请求时消息队列中没有数据，代理将阻止该请求，直到有数据传递或超时。

当然，消费者端发送阻塞队列链接阻塞队列

七.火箭MQ的其他特性

前六个特性基本都是点对点的。如果想了解更多，需要查看源代码，并在实践中使用。当然，除了已经提到的特性，RocketMQ还支持:

定时信息

刷消息策略

主动同步策略:同步双写和异步复制

海量信息积累能力

高效沟通

.......

其中涉及的很多设计思路和解决方案都值得我们深入研究:

消息存储设计:既要满足海量消息积累的容量，又要满足极快的查询效率，保证写入效率。

高效通信组件设计:高吞吐量和毫秒级消息传递能力与高效通信密不可分。

.......

火箭MQ最佳实践

首先，生产商最佳实践

应用程序尽可能多地使用主题，消息子类型由标签标识，标签可以由应用程序自由设置。只有在为发送消息设置了标记时，消费者才能在订阅消息时使用标记来过滤代理中的消息。

业务级的每条报文的唯一标识码都要设置在key字段，方便以后定位报文丢失的问题。由于是哈希索引，请确保密钥尽可能唯一，以避免潜在的哈希冲突。

消息发送成功或失败。要打印消息日志，请确保打印sendresult和key字段。

对于消息不能丢失的应用，必须有消息重传机制。例如，如果消息发送失败并存储在数据库中，定时程序可以尝试重新发送它或手动触发重新发送。

在某些应用中，如果不注意消息是否发送成功，请直接通过sendOneWay发送消息。

第二，消费者最佳实践

消费过程应该是幂等的(也就是说，消费端应该去重复)

尽量使用批量消费模式，可以大大提高消费吞吐量。

优化每条消息的消费流程

第三，其他配置

AutoCreateTopicEnable应该联机关闭，也就是说，应该在配置文件中设置为false。

发送消息时，RocketMQ将首先获取路由信息。如果是新消息，由于MQServer上还没有创建对应的TOPIC，此时如果打开上面的配置，会返回默认TOPIC的路由信息(RocketMQ会在每个broker上创建一个名为TBW102的Topic)，然后生产者会选择一个Broker发送消息。当选定的代理存储消息时，如果发现消息的主题尚未创建，它将自动创建一个主题。结果就是这个TOPIC的所有消息以后都会发送到这个代理，达不到负载均衡的目的。

所以基于RocketMQ目前的设计，建议关闭自动创建TOPIC的功能，然后根据消息大小手动创建TOPIC。

火箭MQ设计相关

火箭MQ的设计假设:

每台电脑都可能停机或停止服务

任何群集都可能没有足够的处理能力

最糟糕的事情一定会发生

内部网环境需要低延迟来提供最佳的用户体验

火箭MQ的关键设计:

分布式集群

强大的数据安全性

海量数据积累

毫秒交付延迟(推挽模式)