27 Topic 分区 订阅如何实现

动态配置

在消息队列里面，因为需要保持架构的简洁度，基于本地文件也是一种常用的方案。比如 Kafka 和 Pulsar 就是基于 ZooKeeper 来实现的动态配置，因为架构中已经集成了 ZooKeeper。RocketMQ 的 Nameserver 是一个缓存组件，没有实际的存储和 Watch 机制，无法实现类似 ZooKeeper 的效果，所以用的是热加载本地文件的方案。

集群和节点元数据的格式和存储

因为 Broker 数量不会很多，一个集群大概是百或千的量级，所以如果从 Broker 数量来看，这两个方案的区别不大。目前业界主要使用的是第二个思路，主要的原因是：Broker 元数据分开存储方便管理，避免节点间相互影响，也可以避免单个 ZooKeeper Node 的数据量过大存不下。

在集群中支持 Topic 和分区

消费进度的保存机制

28 顺序消息和幂等 如何实现

基于顺序存储结构的设计

消息队列实现顺序消息的前提是： 一个生产者同步发送消息到一个分区才能保证消息的有序。

我们就只要保证局部有序的数据写入同一个分区即可，即根据某个标识将需要有序的数据发送到同一个分区中

主流消息队列的实现机制

Kafka 和 Pulsar 是通过生产端按 Key Hash 的方案将数据写入到同一个分区。RocketMQ 是通过消息组（功能上类似消息 Key）将同一个消息组的数据写入到不同的 MessageQueue。RabbitMQ 是通过 Exchange 和 Route Key 的机制，将数据写入到不同的 Queue 里面。接下来我们来看一下实现的细节。

Kafka 和 Puslar 的生产端支持按 Key Hash 的生产分区分配策略。我们只需要给每条消息赋予一个消息 Key，比如将属于 AppID 1001 客户的消息的 Key 都设置为 1001，此时 Key 为 1001 的消息会被固定发送到同一个分区。配合生产端的单个生产者和同步发送机制，就可以保证属于 AppID 为 1001 的数据被有序存储

RocketMQ 支持消息组（MessageGroup）的概念。在生产端指定消息组，则同一个消息组的消息就会被发送到同一个分区中。此时这个消息组起到的作用和 Kakfa 的消息的 Key 是一样的

总结来看，Kafka 和 Pulsar 中的消息 Key，RocketMQ 的消息组，RabbitMQ 的 Route Key 就是我们提到的标识。只要标识一样数据，就会被路由到同一个分区进行存储，从而保持消息有序

幂等机制的定义和实现

生产幂等的设计实现

29 延时消息

延时消息的场景和定义

假设生产端发送定时 30 分钟后或者明天早上 8 点可见的消息给 Broker，Broker 在接收到延时消息后，会先持久化存储消息，然后标记这个消息不可见。再通过内部实现的定时机制，延时到期后将不可见消息变为可见消息，从而让客户端可以正常消费到这条数据。

定时检测写入

如下图所示，是指 Broker 收到数据后先将数据存储到某一个存储中（比如某个内置 Topic），同时有独立的线程去判断数据是否到期。如果数据到期，则将数据拉出来写入到实际的 Topic，从而让消费端可以正常消费数据

消费时判断数据是否可见

是指每次消费时判断是否有到期的延时消息，是的话就从第三方存储拉取延时消息返回给消费者，从而实现消息从不可见到可见。

生产端在写入数据的时候也会将数据写入到第三方存储。但是和前一种方案不同的是，每次消费时会主动去判断第三方存储中是否有消息到期，有的话就把到期数据返回给客户端。

定时机制的实现

延时消息定位处理指的是随着定时器推进，在每个时间刻度可以高效定位，获得需要处理的延时消息列表。即需要重点关注添加、获取的时间复杂度

延时消息的技术方案

基于轮询检测机制的实现

通过分治的思想来缓解性能并提高精度

基于时间轮机制的实现

这是包含 Seconds、Minutes、Hours 三个级别的时间轮，每一个时间轮的最大刻度为 8，上一级时间轮最小刻度等于下一级时间轮刻度的总和。当我们设定好时间精度和时间轮的维度后，如果是添加延时消息，则在多级时间轮上找到对应时间的延时消息列表，把消息插入到列表中。如果是获取到期的延时消息，也是根据时间轮找到当前时间的延时消息列表，然后把整个列表拿出来处理即可

时间轮算法的核心思路比较好理解，难的是在工程实现方面。它的核心是：对于内存使用量的控制和状态持久化两个方面。即在实现多级时间轮的功能的基础上，要尽量减少这个时间轮对内存资源的占用

主流消息队列的延时机制实现

生产者把延迟消息发送到 Broker 之后，Broker 会根据生产者定义的延迟级别放到对应的队列中。而消息原本应该去的 Topic 和队列，会暂时存放在消息的属性（property）中。

在 RocketMQ 中，会有专门的线程池去处理延迟消息。比如 18 个延迟级别，就会生成 18 个定时任务，每个任务对应一个队列。这个任务每隔 100 毫秒就会去查看对应队列中的消息，判断消息的执行时间。如果到了执行时间，那么就会把消息发送到其本该投递的 Topic 中，这样消费者就能消费到消息了。

30 事务消息

主流消息队列的事务功能

RabbitMQ 的事务是在 Channel 维度实现的。将通道（Channel）设置为事务模式后，所有发送到该通道的消息都将被缓存下来。事务提交后，这些消息才会被投递到具体的 Exchange 中。如果事务提交失败，可以进行事务回滚，使消息不会被发送到目标 Exchange。

生产者发送事务消息到 Broker，Broker 会在 Commitlog 持久化存储这条消息并标记为不可见。当客户端本地操作执行完成后，再提交二次确认结果，将消息标记为可见，让消费端可以消费到消息

RocketMQ 事务消息仅支持在 MessageType 为 Transaction 的 Topic 内使用。也就是说事务是在 Topic 维度生效的，事务消息也只能发送到类型为事务消息的 Topic 中。

客户端会先设置一个事务 ID，多个生产者中设置的事务 ID 可以是一样的。用这个事务 ID 开启事务后，可以实现对多个 Topic 、多个 Partition 的原子性的写入。Broker 收到消息后，会按正常流程保存事务消息，只是将这些消息标记为不可见。当提交事务后，才将这些消息标记为可见，让消费端可以消费到

从底层实现来看，Pulsar 的事务处理流程与 Kafka 的事务处理思路大致保持一致。都是通过事务 ID 来标记事务，开启事务投递消息，都会将消息标记为不可见，同时往一个内部的 Topic 记录事务的状态数据。等全流程处理都成功后，才会提交事务。此时在生产端标记消息可见，在消费端提交消费位点，从而完成整个流程。Pulsar 的事务可以看作是 Kafka 事务的升级版，它保证的是流处理操作的原子性。

消息队列的事务方案设计

31 死信队列和优先级队列

死信队列

如果是消息队列内核 SDK 实现的死信队列，一般只支持同一个集群内的另外一个 Topic 作为目标存储，最多支持跨集群 Topic 的投递。这是因为如果客户端 SDK 集成其他的引擎，客户端就需要耦合其他引擎的写入逻辑，这会让消息队列的 SDK 变得很臃肿，不够单一，长期来看维护成本很高，所以社区的 SDK 一般不会支持跨存储引擎的投递

生产死信队列

消费死信队列

RocketMQ 和 RabbitMQ 的投递目标都是本集群内的资源，比如 Topic、Exchange、Queue

优先级队列

为每个优先级分配一个分区，写入时将不同的优先级数据写入到不同的分区。消费时指定分区消费，优先消费优先级高的分区

缺陷比较明显。主要缺点是没法支持灵活的优先级设置，在优先级级别很多的情况下，会额外冗余很多的分区和 Topic。另外在生产端和消费端都需要感知到分区 / Topic 和优先级的关系，控制写入和消费，这会导致客户端的逻辑很复杂。

因为消息队列堆积的数据可能会很大，所以我们需要选择数据量大时性能仍然优秀且稳定的算法。从具体业务使用场景分析，消息队列优先级一般是相对固定的、有阶梯的，比如固定的 5 个、10 个优先级这样子。基于这两个信息，结合上面的表格，我会建议你选择归并排序。

32 消息查询

什么时候会用到消息查询

消息数据存储结构

根据offset查询数据

假设我们有 10 亿条数据，按照上面的设计应该也有 10 亿个索引节点。以此类推，如果数据更大，索引数据会占用大量的存储空间，所以我们在顺序链表的基础上，可以引入跳跃表来节省空间。

引入跳跃表的主要思路是按照一定的间隔跳跃着保留中间元素。当检索数据时，通过二分算法找到离目标最近的前一个跳跃表元素。如果恰好是需要寻找的元素，就直接返回，否则就往后遍历数据找到数据

根据时间戳查询数据

根据消息 ID 查询数据

复杂方案则是是基于哈希表、B+ 树、红黑树等数据结构来构建消息 ID 和 Offset 的索引，同时保证在索引元素添加和获取的时候的时间复杂度较低，从而满足查询需求。因为根据消息 ID 查询消息的需求是很固定的，所以我会建议你使用哈希表来构建索引，因为哈希索引结构能够实现高效的消息查询。

借助第三方工具实现复杂查询

33 消息队列中的schema模块

Schema 就是用来解决全流程中的数据格式的规范定义问题，即保证上下游数据在传递过程中，消息可以根据指定的格式和定义进行传递。

34 消息队列支持WebSocket

WebSocket 是什么

WebSocket 是一种基于 TCP 传输协议的应用层协议，它设计的初衷是解决 Web 应用程序中的实时双向通信问题。它跟 HTTP 协议一样，也是一种标准的公有协议。所以它也有协议头、协议体、数据帧格式、建立连接、维持连接、数据交换等等各个细节。

在需要高度实时性和低延迟的场景中，WebSocket 是一种理想的解决方案。所以 WebSocket 在 Web 即时通讯、在线游戏、实时股票和金融信息流、物联网通信和在线协作工具等场景中会被大量使用。

长连接和双向通信是 WebSocket 实现高度实时性和低延迟的技术核心，也是消息队列需要支持 WebSocket 协议的重要原因。

这里想要展现的的场景是：物联网传感器（例如气体传感器）收集数据后，将数据通过 MQTT 协议发送到 MQTT Broker。收到数据的同时，Broker 可以通过 WebSocket 将数据推送到浏览器或客户端，从而及时更新监控面板或者触发告警

内核中支持 WebSocket 协议

主动消息推送

当数据写入到 Broker 中的 Topic tp1 时，因为 WebSocket 是双向通信的，所以 Broker 收到消息后，可以直接将消息推送给客户端。这个推送的逻辑一般是在内核中维护异步线程去回写数据到客户端实现的

35 主流MQ中间件的高级功能