消息中间件学习

引子

随着2017双十一落下帷幕，notify和metaq这两个集团最大规模的消息中间件依然保持着超跑般顺畅的一贯作风，在0点峰值的秒级消息收发量达1亿，当天的消息收发总量更是突破了2万亿。随着notify和metaq的功能特性越来越相似，并且交易消息也同时存在于notify和metaq，很多人不禁会问notify和metaq现在是什么关系，都有哪些区别，未来会如何发展？要了解未来，必先要了解过去，我们来纵览下两个消息产品的发展历程。

notify前传

先说集团最早的消息中间件notify，notify诞生于2007年，是伴随着淘宝交易系统进入分布式时代产生的。主要用于核心交易链路的异步解耦，以缩短了下单流程的RT，提高了电商业务的扩展能力。目前电商核心业务领域对象包括收藏夹、购物车、订单、用户、商品、物流单、资金的状态变更都会产生notify消息，很多电商下游和周边系统都可以通过订阅这些消息来透明的构建围绕自身的特色业务。以下是notify支撑的核心交易业务图，可以认为notify百分之90以上的消息都是来自交易相关业务。<br />![](https://img-blog.csdnimg.cn/img_convert/4da83a0e31557e0c706a50bd15195568.png#crop=0&crop=0&crop=1&crop=1&id=WgDrG&originHeight=348&originWidth=692&originalType=binary&ratio=1&rotation=0&showTitle=false&status=done&style=none&title=)<br />notify在设计之时，消息中间件规范JMS及AMQP已经发布，notify的很多领域对象都是参考这两大规范。

• Producer：消息的上游，消息生产者
• Consumer：消息的下游，消息消费者
• Message: 消息，作为异步通信的数据载体。
• Broker：消息中间件服务器
• Topic：消息主题，用于发布订阅模式
• Queue: 消息队列，用于点对点模式
• Subscription：订阅关系，用于发布订阅模式，表示消费者的感兴趣消息类型
• Exchange:交换器。用于消息订阅计算，找到目标Consumer。
  除了实现基本的消息收发功能之外，相比于其他消息中间件，notify有着自身的特点，主要是事务消息、header过滤订阅和实时推送。这两个功能是电商链路的刚需，解决了异步架构的最终一致性问题、也解决了电商下游细分业务领域的消息过滤需求。
  先说事务消息，它实现了消息生产者本地事务和消息发送的原子性。以订单创建为例，tp先创建订单（本地事务），再发订单创建的消息，如果订单创建成功，消息没发出去，那么下游所有系统都无法感知这个事件，会出现脏数据；如果先发订单创建消息，再创建订单，也有可能出现订单创建失败了，但是下游系统都认为此订单已经创建，也会出现脏数据。而事务消息的作用就是用于保障订单创建成功和消息发送成功的最终一致，交易上下游的状态也才能达到最终一致。
  再说header过滤订阅，它解决的是消息精细过滤问题。今年双十一峰值时刻TRADE（这是交易订单对应的notify topic）的秒级消息量是140w，而订阅TRADE的下游应用有600个，如果这600个应用照单全收所有的TRADE消息，对notify server来说需要每秒推送8.5亿消息，TRADE消息体大小为6k，notify的出口带宽会达到4894GByte，这会对notify服务器和机房带宽产生巨大的压力。对于订阅者应用而言，要处理每秒140w的6k大消息，每个应用也都要准备几百台的机器才能扛得住，会造成极大的机器浪费。有了header过滤订阅，notifyserver会按照订阅者设定的规则，只给每个应用推送它所需要的消息，例如某个类目、某个卖家、某个垂直市场的，甚至可以做到多种消息属性的与或非组合条件。
  一个系统的架构倾向往往和他所要满足的业务场景有关，为了方便实现事务消息和属性过滤，notify采用了kv的存储范式来维护消息状态、以及堆内即时过滤推送的消费模式。
  notify消息领域对象的设计如下图所示，每条消息记录包含messageId（这个就是key）、topic、body(二进制)、header（用户自定义kv集合）、commited(事务状态，是否提交)、Fail target(记录失败的消费者id，消息重试要用到)。
  
  普通消息的处理流程如下，当notify收到消息后，先进行持久化，在kv存储增加一条记录，然后ack给生产者；持久化后执行exchange，这一步会对消费者Subscription的属性过滤规则进行计算，得到目标消费者id集合，找到消费者channel直接push。push完后，内存中维护callback等待消费者ack（成功或者失败），如果没有ack，callback默认等待5秒，作为消费超时处理。最多等待5秒后，callback收集到所有ack，如果全部成功，那么直接删除消息记录；如果部分失败，对消息记录进行修改，主要是修改fail target字段，记录失败消费者id。notify会有后台任务扫描未删除记录，重新投递给失败的消费者id，再次执行4、5、6步骤。 对于事务消息，区别之处在于消息生产环节。生产者先发送half消息，当notify收到half消息后，先进行持久化，在kv存储增加一条half（即commited字段为false）记录，然后ack给生产者，此时并不会触发推送消费者事件。生产者half发送成功后，执行本地事务，事务执行成功后，发送commit请求。notify收到commit请求后，讲消息的commited字段更新为true，然后就立马执行推送消费者的流程，消费相关逻辑和普通消息无异。
  
  通过以上的分析我们可以看到，notify的消息涉及到更新操作，所以选择了支持更新语义的kv存储；另外属性过滤需要拿到消息的header，在接收时刻消息header已经反序列化完成，此时直接进行属性过滤计算得到消费者id集合后立马推送，开销最低。
  从07年到现在，notify采用的kv存储模块经历过多次改造。最早notify的消息存储是采用本地文件存储的，参考了ActiveMQ的kaha实现了单机kv存储引擎，在07年12月上线，支撑直冲业务，日均500w消息。然而考虑到交易消息重要性，消息要保证不能丢失，第一个版本的单机存储引擎没有经过多年生产环境的验证，不够成熟，另外也没有多副本机制，无法满足交易对消息可靠性的严苛需求。到了第二个版本，notify开始采用当时最稳定的关系型数据库oracle来做消息存储，在08年3月份发布，支撑核心交易链路，日均1000w消息。随后集团开始启动去O战略，数据库存储需要全面从oracle迁到mysql，为了弥补mysql和oracle的性能和稳定性差异，notify实现了多点高可用mysql存储集群，全部迁移到mysql存储，10年4月完成。由于多点mysql在存储层具备了无状态横向扩容的能力，在稳定性和性能方面许久未遇到瓶颈，所以一直使用了很多年。

metaq前传

再来看看metaq的发展史。在2011年，Linkedin开源了一款全新的消息队列kafka，是面向高吞吐量设计的系统，主要用于日志和大数据领域。同年，据说是出于兴趣的原因，当时的notify负责人伯岩花了两个礼拜的时间，开发了java版的kafka，取名为metamorphosis（变形记是奥地利作家卡夫卡的名作，算是对kafka的致敬），主要用于集团内部的日志传输业务。
metamorphosis的第一个版本是基本上是参考kafka的架构，每个topic都会对应broker的多个分区文件，一旦topic数量增多，broker的分区文件数也会随着增大，本来高性能的顺序写文件会变成随机写，吞吐量会有较大的下降。2012年9月，誓嘉解决了这个难题，发布metaq 2.0对存储层进行重新设计以满足大型互联网复杂业务的需求，性能不再随着分区数增大而下降。核心存储的设计如下图

metaq存储的核心是实现一个持久化的分布式队列，重新设计后的metaq抽象出了CommitLog和Consume queue。其中CommitLog属于物理队列，存储完整的消息数据，不定长记录，也起到了类似redo log的功能，一旦CommitLog落盘成功，消息就不会丢失；所有topic的消息都会写入到同一个CommitLog，哪怕单机一万个topic，还是能保持顺序写，保障吞吐量。Consume queue可以认为是逻辑队列、索引队列，每个topic的消息在写完commitlog之后，都会写到独立的consume queue；队列里的每个元素为定长记录，元素内容包含该消息在对应commitlog的offset和size。基于这样的存储结构，metaq对客户端暴露的主要是consume queue逻辑视图，提供队列访问接口。消费者通过指定consume queue的位点来读取消息，通过提交consume queue的位点来维护消费进度。
去O战略实施后，集团大部分应用都采用分库分表的mysql架构，出现了很多binlog数据同步需求，例如买卖家库的数据同步。这类场景对消费顺序有强需求，而metaq的队列存储结构的结构天然就是FIFO（相比而言notify的kv模型实现起来代价极大），很适合实现顺序消费的场景。于是metaq就开发了一个最具特色的功能，局部顺序消息，广泛使用于binlog分发订阅的业务场景。随后2013年7月，metaq 3.0发布了，新的版本提供更加丰富的功能，支持消息属性、无序消息、延迟消息、广播消息、长轮询消费、高可用特性，这些功能基本上覆盖了大部分应用对消息中间件的需求。除了功能丰富之外，metaq基于顺序写，大概率顺序读的队列存储结构和拉模式的消费方式，使得metaq具备了最快的消息写入速度和百亿级的堆积能力，特别适合用来削峰填谷。

双十一的挑战

从功能层面上来说，近几年notify新增的几个重大功能特性都无需都改动notify的底层架构，例如单元化、管控体系建立、消息轨迹、全链路压测、灰度环境、消息路由等。采用kv的存储模式和SEDA的事件驱动架构，很容易扩展出新的功能特性。真正驱动notify进行架构级改造是来自于双十一的挑战，从13年到17年，每年双十一峰值消息tps基本上是按照翻番的速度增长，对notify的性能、横向扩展能力、成本、高可用、稳定性、削峰填谷能力等方面都提出了巨大的挑战。
第一个瓶颈点是出现在14年，随着notify客户端机器数增多，notifyserver的单机连接数突破2w，原来使用的gecko网络框架没有共享内存池机制，每个链接独自占用固定大小的内存缓冲区，2w的长连接会直接导致server内存不足，无法服务。这一年notify对网络层进行了重构，从gecko升级到netty。基于netty的共享内存池和堆外内存机制，链接消耗内存的问题得到解决，到今年支撑单机4w的长连接无压力。

第二个瓶颈点是消息堆积给存储层带来的稳定性压力，以及每年翻倍的峰值tps给存储层带来的成本压力。从notify对kv的存储范式来看，notify的一条普通消息会对mysql进行一次add和update操作；如果是事务消息那么会对mysql进行一次add和两次update操作；如果消息是采用HA同步双写的机制，那么写操作会翻倍。在2015年的双十一容量准备中（交易技术目标12-12-8），mysql使用的机型为D13，按照单机3.2w（消息记录大小4k）的写入tps来评估，最后共使用了500多台D13，这部分机器成本约4000w。另外一方面随着交易上游和下游链路在双十一的容量准备上的悬殊，峰值时刻消息生产速度和消费速度差距越来越大，每年的消息堆积量都几倍上升，现在已经到百亿级。如果对应到数据库层面，也就是单表数据量会特别庞大。在消息重试阶段，所有的notifyserver不断从mysql读取大量堆积数据，进行再次更新或者删除操作。这些操作对于以B+树结构存储数据的mysql而言并不友好，会产生大量的随机IO，影响性能。所以从成本和稳定性来考虑，notify的底层存储的架构改造刻不容缓。
既然不用关系型数据库，那么notify又是按照kv的范式设计的，那么自然而然想到的是采用nosql。比较成熟的持久化nosql产品有levelDB、hbase等，这类产品都是基于LSM树的存储结构，完全顺序写，写性能很高，但是随机读性能不好。而notify在消息重试阶段会产生大量随机读，读性能不满足需求，百亿级堆积的场景没有太大把握，消息组内也没有运维这类产品的经验。就当时想到的存储模型来看，也就只有像metaq、kafka这类的文件队列存储结构比较适合支撑大规模堆积的场景，并且比kv存储具有更高的吞吐量。但是这类存储结构暴露的接口是队列和位点，和notify的kv范式有着天然的阻抗，没法直接使用。
通过对notify的kv访问特点进行分析，同时结合sql、nosql等存储模型的设计思想，终于找到了一种方式，冲破队列和KV的阻抗。让notify能够以kv的接口来读写metaq队列，维护每条消息的处理状态。存储结构的设计如下：

如上图，新的存储设计以metaq的队列存储为基础，抽象出了Data Queue（DQ）和Operate Queue(OQ)。其中DQ存储的是消息完整数据，notify的消息对象通过一定的映射规则转换为metaq的消息对象，写入DQ。而OQ是和DQ成双成对的，每个DQ都会有对应的OQ，OQ里面的记录存储的是对应DQ消息的操作，每条DQ的消息默认会在5秒（消费ack 默认5秒超时）之内有对应的OQ数据写入。如果是删除，那么OQ里的内容是DQ位点打上删除标记；如果是更新，则会包含DQ位点需要更新的字段值，更新的字段一般是header、commited、failtarget等字段。采用kv over queue的设计，所有原来的对消息记录的增删改都转化为对metaq队列的顺序写、append-only，如下图。消息处理的上层业务逻辑没有变化，所有状态的变更都通过写metaq消息来持久化。在重试阶段，通过合并读取DQ和OQ的数据就可以得到更新后的消息数据，找到fail target失败消费者id，重新推送消息，再次执行4、5、6步骤。

采用基于metaq存储的架构，notify将具备和metaq同等级别的抗堆积能力，并能节约大量的机器成本。notify的机器成本可以分为两部分，notifyserver（4c8g虚拟机）的机器成本+消息存储成本（一般是物理机或者高配虚拟机），采用了新架构，notifyserver的事务消息、同步双写、批量删除和修改都得到优化，性能在部分场景提高了50%以上；而消息存储方面，metaq的队列存储是为消息中间件的业务场景深度定制的，面向大吞吐量和堆积能力设计的（写消息已经突破50w tps，读写全开可以打满万兆网卡，），而mysql则是通用的关系型存储，数据的读写逻辑要比metaq的队列模型复杂、开销大，所以在消息存储的场景下，metaq的机器成本是远低于mysql的。
2015年双十一，我们在一个共享集群和交易镜像集群深圳单元上线这个新架构notify3.0，基本成功的支撑了双十一部分流量的验证，虽然中间还是出了些小插曲。这个小插曲是metaq的毛刺问题，在2016年双十一之前得到优化，有另外一篇文章论述metaq毛刺优化。
在2016年我们又针对notify3.0的架构做了很多稳定性优化，在2016年双十一全量上线，秒级消息收发量在2000w，机器成本和原来的架构相比整体节约了3000w以上，完美支撑当年双十一。16年的双十一对集团消息中间件产品是具有历史突破性的一年，这是集团第一次用自研定制存储全量支撑核心交易链路消息流量，性能和稳定性得到最有力的验证。

消息统一

在维护notify的过程中，经常会有用户问notify和metaq怎么选型，从功能特性的角度来看，两个产品太相似了。大部分的功能一样，但是又有少部分情况下需要做选择，例如需要事务消息、属性过滤用notify；需要广播消息、顺序消息、延迟消息、拉模式消费用metaq，有些用户没有考虑全面就会选型错误。在我们看来，最理想的方式就是一个消息中间件满足所有产品特性，用户无需选型。这个想法在几年前是很难实现的，对于notify来说，kv模型难以实现顺序消费和拉模式；对于metaq而言，append-only的模式难以实现事务消息（事务消息的提交和回滚需要更新消息的状态），topic粒度级别的文件队列、拉模式、零拷贝等特点都增加实现属性过滤的难度和性能代价。而现在notify3.0的架构给我们带来新的思路，采用notify3.0的模式能够让append-only队列模拟更新操作，也就是说我们可以把notify3.0的kv over queue的模块移植到metaq内部，实现事务消息的更新操作；而metaq的消息属性过滤也有了一个新方案，通过引入一个扩展的文件队列，采用预计算+布隆过滤器的技术，可以极大的减少消息过滤的内存开销和计算开销，低成本实现属性过滤订阅（服务端SQL表达式过滤）。基于这些进步，之前无法做到的集团消息中间件统一，现在具备了技术可行性。一旦metaq实现功能全集后，用户就无需投入太多精力做消息选型，新业务直接选用metaq；还有一个好处是集团消息中间件积累十年的所有产品特性都可以由一个产品来承载，统一通过metaq（云上MQ）为阿里云的客户提供服务。
metaq实现功能全集后，新业务选型的问题是解决了。但是还有一个问题没有解决，就是核心交易链路大部分的消息都是走notify的，并在这部分消息量在双十一峰值很高，有些用户希望能够采用metaq拉的模式来慢慢消费消息。目前只有交易的消息能够通过metaq订阅，因为3.0的设计只能支持一个集群共用一个metaq topic，像notify tradesub集群刚好只服务一个topic，所以直接把DQ的topic设置为TRADE，来实现notify写入，metaq订阅。而大部分的notify集群都是服务多个topic的，没法按照单个topic的粒度来写到metaq。如果要在3.0的架构上实现单机拆分多个DQ，虽然功能能够实现，但是队列数增多的话（17年双十一，metaq扩容后，单个DQ的队列数已经到接近1000，N个DQ则会是N*1000的队列数，明年会更多），删除和更新的批量效果会很差、消息重试的读会比较分散，整体性能会下降。另一方面，3.0是通过metaq的客户端来写消息，notify消息压缩标记无法透传到metaq的压缩标记，对于压缩的业务消息，metaq客户端无法识别解压（交易消息比较特别，没有使用notify客户端内部封装的压缩机制，本身无压缩标记）。
综合考虑功能和性能因素，我们打算采用notify和metaq深度融合的架构，notify内嵌rocketmq内核，同进程部署，直接调用rocketmq的store模块，同时暴露notify和metaq服务。一来队列数可以减少几百倍，提高批量因素，二来可以透传压缩标记到store模块。除了提高更新和删除的批量因素外，深度融合后的架构还能使性能获得额外的更大收益，原来消息的读写操作走要走一次RPC，现在变成本地方法调用；metaq是IO型应用，notify是计算型应用，同机部署notify刚好可以利用metaq富余的cpu资源。4.0和3.0的架构区别如下

在今年双十一，消息统一项目已经有了阶段性成果，metaq已经完成功能全集。事务消息已经移植到metaq，在txc集群上线；metaq的属性过滤也在metaq的交易集群上线，有近十个用户已经在使用，其中菜鸟的蓄洪项目也采用了新的属性过滤功能，进一步提高蓄洪能力；而notify4.0作为两个消息中间件融合的初版，也在notify共享集群上线，承担部分峰值流量。今年双十一的情况详见总结 《2017metaq双十一总结》](https://www.atatech.org/articles/94762))、[《2017notify双十一总结》]([https://www.atatech.org/articles/94672)](https://www.atatech.org/articles/94672))
接下来我们还会对notify 4.0进行持续的优化，通过简单配置，可以实现notify发送的任何topic消息可以通过metaq客户端来拉取；推模型实现首次投递的动态限流，真正削峰填谷；推模型消息重试实现优先级队列和独立队列，以topic为维度或者消费者id为维度，保障关键业务消息重试延迟不受集群整体堆积量的影响。

消息统一项目的终局就是Notify专注于为核心交易链路服务，把推模式做到极致，同时提供Metaq拉模式消费；而Metaq则继续丰富完善功能特性，满足集团内所有的消息领域场景，作为新业务接入的首选消息产品。
最后贴下最新的notify和metaq功能对比图，供用户参考。