QMQ 顺序消息设计与实现（下）

在上一篇文章中，我们首先讨论了目前已有的MQ提供的顺序消息的缺陷，并且探讨了借鉴数据库分库分表的思想解决这些缺陷的可能性。

首先我们简单回顾一下上一篇文章中对扩容缩容移动的描述：

• 扩容即对physical partition按照logic partition的范围进行分裂的过程

• 缩容即按照logic partition的范围对physical partition进行合并的过程

• 移动即改变logic partition与physical partition的映射的过程
  

虽然我们从Database的分库分表思想中学习到了logic partition，但是Message Queue和Database究竟是两种不同的模型。在DB里，reader是无状态的，也就是每次读取传入的查询条件都是独立的。而MQ的reader(consumer)当前的读取位置(offset)是依赖上次的读取位置，一旦partition发生改变，则这个offset将无法继续保持，那消费就会错乱了，顺序也无从谈起。另外因为数据量太大，我们在执行扩容缩容移动的时候并不想对数据进行移动。

接下来以实际的例子来进行说明，下面是一个扩容的实例。order.changed这个主题，原来分配了P1, P2两个分区，现在因为容量不够，需要对P2进行扩容(分裂)。也就是将physical partition P2进行分裂，分裂成P3, P4两个分区。分裂的原则是按照logic partition的范围进行，logic partition [500, 1000)原来映射到P1，现在logic partition [500, 750)映射到P3, [750, 1000)映射到P4。也就是分裂以后producer发送新的消息就会按照新的映射关系将消息append到P1, P3或P4，P2不再接收新的消息了。

接下来具体描述一下实现步骤。在QMQ里有个metaserver的组件，它管理所有元数据信息，比如某topic分配到哪些partition上(我们将其称之为路由)：

metaserver还管理partition分配在哪些server上，以及logic partition与physical partition的映射关系。

在需要对P2进行分裂的时候，metaserver会发送一条消息给P2所在的server，这条消息会被append到P2上，该消息称之为 指令消息 (command message)，对客户端不可见，也就是业务代码不会消费到这条消息。P2收到这条指令消息后将不再接收新的消息了，所有业务消息均被拒绝，那么这条指令消息就是P2上的最后一条消息，相当将P2关闭了。

metaserver发送完指令消息后会变更对应topic的路由信息：

注意看上面的表格的特点，这个路由信息表与众不同的地方在于它有一个version字段。对于producer而言它总是获取最新版本的路由信息，也就是路由发生变更后，producer就会获得更高版本的路由信息，然后向这些分区上发送消息。

但是对于consumer来讲，它必须将前面的消息消费完成才能消费后面的，否则顺序就乱了。比如前面分裂的示例，P2分裂为P3, P4了，这个时候P3, P4并不是立即对consumer可见的(只要对consumer不可见，就没有consumer来消费它)。只有当consumer消费到指令消息时，才会触发consumer的路由变更。并且指令消息里携带了路由的版本信息，假设路由已经发生了多次变更，consumer消费到某个指令消息的时候，只会将consumer的路由变更到该指令的下一个版本，而不会跳到其他版本，这里触发路由变更的时候会使用乐观锁去更新版本(伪代码)：

update routes set version = version + 1 where topic = @topic and version = @current version

总结起来就是producer总是使用最新版本的路由，而consumer使用指定版本的路由，路由的版本由指令消息进行同步。

其实这个流程中最有趣的不是扩容(分裂)和缩容(合并)，而是移动。比如我们现在发现P4分区所在机器负载比较高或磁盘就要满了，现在给集群加了几台机器，怎么做能在继续保持顺序的基础上又能将负载分散过去呢？那么只需要发送一个移动的指令消息给P4，然后P4就会关闭，然后变更路由，order.changed的路由现在是P1, P3, P5，这次路由变更分区的个数没有发生改变，改变的只是logic partition和physical partition的映射关系:

因为P5是新分区，所以他可以分配在新机器上了。而且这个特性可以用在提高顺序消息的可用性上，比如需要对某台server停机，那么我们只需要对其上面所有分区发送移动指令即可。

另外，在实现的时候我们还增加了如下约束条件：

• 版本必须是连续递增的

• 每次只能执行一项变更，比如只能对一个partition分裂，不能对多个partition进行分裂

• 对logic partition范围的每次操作必须是连续的，比如合并的时候只能将[0, 100) 与[100, 200)合并，而不能将[0, 100)与[200, 300)合并

• 路由变更必须是本次变更分区所有的消费者都确认执行到指令消息才能触发。比如将多个分区合并的时候，必须是这几个分区都消费到了指令消息的时候触发。

总结

上面以示例的方式描述了QMQ如何进行扩容(分裂)，那么只需要按照这个步骤进行，consumer在没有将更早的消息消费完成的情况下就不会拿到更新的路由。

至于如何确保顺序的消费这些分区的消息那就跟其他MQ一样了，只需要将分区分配给指定的consumer实例，只允许指定的实例独占消费该分区即可。

QMQ是去哪儿网开源的消息中间件，支持事务消息，延时消息，tag过滤等特性。点击原文链接将跳转到github开源地址上，欢迎给我们提交PR，Star。