携程异步消息系统实践

jack

顾庆：今天会跟大家分享一下我们在携程，现在应该是正在推广的一个新的消息系统，主要会偏重于讲一些架构和实现方面的内容。目前我在携程大概一年多都在做新的消息系统Hermes。

消息队列的优势
我们先回顾一下消息系统有哪些优势，为什么在企业里面会用这样的消息系统。MQ大家都会用到，像KAFKA等。企业里有很多的应用场景，比如实时的索引更新，或者是异步化的事情，使用MQ会发现这系统之间的耦合度降低了。MQ也是，它有什么特点？一个可能是可以做到异步的处理，这个处理时间可以很长，比如收完这个消息以后一小时再处理，非常灵活。还有就是可以抵御流量波峰。就算量很大，可以把这些量缓存在MQ里面，从而抵挡住这样的一些波峰，导致业务不会宕掉，按照我的吞吐量慢慢的做处理。另外它支持比较大的fan-out，我可以同时call很多人。目前我们面临着越来越多的个性化的实时的需求，一对多的场景会越来越多，这是MQ得以应用的地方。

MQ基本的模型如图1所示。最简单的它是一个Queue的模式，生产者的消息可以发给一个消费者，一个消息只能发送一次。Topic的话消息可以给多个人，每个人都可以拿到一个。这个名字可能不是标准的，很多MQ都会支持这样一种模式，就是Consumer可以组成一个组，你可以有好多个组，组之间是topic的语义，就是一个消息可以到达每一个组，但是组内只有一个人会消费，这种场景其实是最多的。比如说Group1是第一个业务，Group2是第二个业务，Group1之内这些机器其实是互备，或者说是并行处理，之间有一个吞吐量，它们之间不会互相影响，但是一个消息在一个组内只处理一次，最多的是这样的场景。

携程MQ系统架构演进
我们也曾做过很久的MQ的话，也犯过很多错踩过很多坑。1.0版本是在三四年前，那时需要的消息队列很简单，只要有一个地方可以存入和读取就可以，架构如图2所示。客户端直接写消息到mongo里面去，消费者直接从里面拖出来。它没有broker，就没有server，它支持queue和topic，不支持consumer gorup。1.0版本的特点在于，开发成本极低，但是一旦要升级，就会涉及大量的客户端，这个在企业里面基本上推不下去。还有它没有broker，但是客户端之间，主要是消费者之间，要做一定的协调。因为没有broker，只能通过DB做协调，可能就会涉及到DB里面有find and modify。大量的用DB去做协调，这时DB被用得很重，性能会很差。还有很多feature没有办法做，因为你没有broker，或者server帮你的协调，feature就没有办法去做。此外，没有弹性，或者说没有架构，基本上没法做任何调整。

基于这样一些东西，我们做了第2版，有broker，有中间的server，依然写到mongo DB里，如图3所示。差别在于，这个broke也不是一个纯粹意义上的集群，它其实是一个Master-Slave，通过mongo DB去做心跳。但是它是支持consum broker，因为它有broker，它可以做协调，broker来决定这个消息是要给谁。

还有一个比较大的变化在于，所有的客户端是不直接连向DB而是连向broker的，这在架构上也是一个很大的改善。此外，客户端很瘦，只是跟broker通讯，我告诉你我要发，consumer告诉broker我要收，然后broker把消息存进去或者推给consumer就可以了。这时候很多时候不需要做升级客户端这个操作，只要升级broker就可以了，而broker升级也很方便。

但是它的缺点在于，当初没有考虑到它的绑定关系。一个topic绑定到一个mongo DB的表上去，那么就是说，这个存储的粒度是一个topic。还有就是，broker之间不做协调不做管理，大家是不认识的，只是做心跳，所以还是有很多人工管理的成本在。还有mongo这边有一些改进，我刚才说这个消息怎么样去清。这可能跟Metric一样，要么去做分表，你写了这张表以后写下一张表，上一张表就可以清到了，drop掉就可以了，是很轻量的。如果放在一张表里面，你又要做insert，又要做delete，delete也是非常费的。所以当初我们选的时候，mongo有一个capped collection，这个是它是固定长度固定大小，比如就是2个G，超过2个G的存储它自己会滚回来。这样你不需要管数据清理这样的事情，只要设置好初始的大小就可以了，比如说10个G，满了自己会滚掉。一开始想还蛮好的，10个G，滚来滚去还不用做数据清理。后面你会发现，应用那边会说，一开始估计说可能2个G够了，后来你会发现其实长到20个G、200个G，然后mongo又不支持动态把capped-collection扩大，也蛮坑的。这个对架构上相对比较灵活，broker这个东西在，但是整个东西还是需要很多人工的管理，它是一个很笨的东西。

这个是现在的架构，其实还是三块，一块是producer，这边是消费者，中间是服务端，下面有存储。可以看到不一样的地方，一个是引入了metaserver它主要是做一些集群的协调。broker还在，底层的存储有两个，一个是用MySQL，一个是KAFKA。

两种消息类型
我们的消息存储有两种，一种是基于KAFKA，优点是吞吐和性能非常高，因为它写内存，OS刷磁盘，复制，很多事情让OS去做，非常高效。但是很多的特性它不支持，比如消息重发、优先级的消息、消息的过滤等。KAFKA的设计理念就是追求简单，所以吞吐和性能会负责好，但是很多feature你要自己去解决。还有就是，它的broker大量采用ZeroCopy，broker就像透明的东西一样透出来，它其实什么都不管的。在broker里面其实没有办法做很多的监控，很多事情没法做监控。这个在企业里面其实是非常关键的。KAFKA在外围生态方面比较欠缺，它有一些比较初级的监控系统，我们在用KAFKA的时候也遇到蛮多的坑。所以我们现在的策略就是说，核心的关键的，去走MySQL存储，整个系统都是我们自己搭建的，一旦出什么问题都可以找到，有什么问题也可以fix它。但是像一些日志的，或者用户行为监控的等等，这样的量非常大，但是其实不是关键的应用的数据，你可以去做KAFKA，一旦有什么问题影响也不是太大。

其实这种也是正好是比较，比如你这个消息是非常重要的，那么可能它的量就不是那么大，比如每天有好几个T，交易数据，可能淘宝有，携程其没有那么大，一天交易数据有上T，那公司的钱会非常多，我们可以投入更多的资源在这个上面。所以就现在来说这个模式可以做得非常好，非常大量的数据的，都不是特别重要的，可以KAFKA，很重要的都是MySQL这边，整套系统都是自己的。MySQL其实性能足以支撑绝大多数业务，我们可以做很多丰富的消息队列的特性，也可以做一些定制。整个的监控治理是非常深入和全面的。

如何构建高效的MQ
怎么构建MQ，这个事情比较大。所以我按这样一种思路来讲，先想一下这个系统如果是单机的系统，你怎么样做优化。然后你怎么把这样一个系统扩展到集群上去，因为单机肯定会有上限，你怎么扩展到集群。第三个扩展到集群以后你会发现涉及到集群的管理功能，怎么样做集群的管理？

单机优化
单机优化，对一个消息系统来说怎么评价这个消息系统是好的？从性能上来说，可能最重要的是两个指标，一个是我这个消息写入和读取速度很快，然后它的整个消息的通道非常宽，即使吞吐量非常大，也都可以发到这个消息系统里面去。还有一个是消息的延迟，消息写进去以后，消费者需要等待多少时间才能收到，这是消息的延迟。以MySQL为例，首先是讲消息的存储。一个是这些表的设计，只有ID的索引，即只有主键是有索引的，其他都没有索引。然后它的updata的操作也是根据主键去做，然后（sliver）的操作也是基本上一种，等于ID，大于ID，limit多少，就是这样一些非常轻量的一些操作。

原来MySQL有一个很大的问题是说，数据清理很烦。目前的版本可以支持一个表里面可以按ID或者按date等等去做一些partition，然后这个数据不要了，就drop掉一个partition就可以了。相对来说drop掉一个partition是比较轻量的操作，会比较快，数据的清理会非常方便。

那么它只做insert，当然会select。所以只是在ID上，所以它的insert是非常快的。还有消息队列一般都会有一个消息重发，那么就需要有一个地方来记录我这个消息什么时候需要重发出去。那么，重发时间对用户来说很重要吗？他自己也不知道。我们索性不让用户指定，直接隔十秒钟告知。这个schedule date就是当前时间加十秒，永远是递增的。那么索引就不需要了，这张表也可以变成只有ID的索引。这样你做一些优化，一种是从技术上，一种是从业务上考虑，这是消息系统需要解决的核心问题吗？如果不是但是又影响我的性能，那么你可以从业务的需求上想一些办法。

批量写入比较简单，大家都会想到，从broker端都会做一些批量化，很多producer写过来，我们在broker这边做一些批量。批量的效果其实蛮明显的，有的场景可以达到5倍以上的性能提升。当然对于JDBC来说要打开这样的开关：rewriteBatchedStatements=true。

刚讲的是我能够比较快的把数据写进去，还有一个是写进去以后，broker需要把消息读出来发给消费者。正常做法就是broker会轮询这张消息表，这边就要考虑一个折中，这个消息的延迟和你对这个DB的开销。这里我们会把这个事件捕获到，我什么时候有一个新的消息写进来，这样我不需要去轮询这个数据库就可以知道。但是这个前提是你所有的消息都会经过同一个broker，所以这个待会儿会讲到是怎么做到。

还有当消费者非常多的时候，比如说一百个消费者，写一次就要select一百次出来，这时候消息的缓存效果是非常好，一般的消费者都是追着这个producer，就是它消费的位置其实差不多的。你做一个消息是比较临近的，做缓存效果会非常好。DB可以开销降低很多。另外消息延迟也可以降低，原来从DB去取出来，现在从本地的memory里面取出来就可以。这样有一个问题，最快的消费者和最慢的消费者差很多，内存里放了太多的东西，所以会涉及到这个缓存要分裂跟合并，当中间这一块太多了要裂开，中间这一块就可以释放掉。当你后面这一块追上来以后又可以合并成这样一条。这样的话，缓存清理是非常快，就是一个列表，找到它后面扔掉就可以了，填充也是非常快的，往前填就可以了。

还有一个是消费者这边，broker可以很快的取到这个，它怎么把这个消息推给消费者？两种做法，一种是可以做push，broker收到以后push给消费者，另外一种是消费者来pull这个消息。大家可能会觉得push会比较好，pull可能就是我盲目的轮询，会比较慢。我们一开始确实是选择push，我broker有个消息以后，push给消费者那边去做处理。push有一个问题，所有的消费者状态都需要在broker那边做存储做管理，因为是push，要知道这个消费者消费到哪里了，有什么消息需要做重发的，很多状态hold在broker这边，因为要做push。这样一旦broker做重启，很多都丢掉了。丢掉当然不影响整个消息的接收，但是会有很多消息的重发，比如这条消息我收到过的，但是因为broker还没有刷到DB里面去，然后它重启了，就会重发。然后对push的话，整条链在broker里面的代码很复杂，整个push过来。一旦消费者那边断连了，broker那边要做很多调整，代码上写起来也比较恶心。所以后来换成pull的方式，比较简单，就像一个web一样，发一个请求，pull一下过来，对server是很简单，你是被动的，你有一个pull请求过来，处理好这个请求，把东西写给它就ok了。这边的状态保存非常少，而且保存时间非常短，代码也很简单。但是有一个问题，跟刚才轮询DB是一样的，客户端这边要pull得很快的话，就是实时性好，但是broker这边压力很大。所以后来用到LongPolling，就是我发一个pull请求，客户端。如果没有消息，你不要马上回给客户端说没有消息你再来pull。你就hold这个请求，然后一旦broker端发现有新的消息进来，把这个消息准备好再发给他。这样客户端这边也不需要频繁的轮询，所以是一个long的pulling，pull以后就hold在那里，broker有消息了再推给它。这样这个代码比较简单架构比较好，又可以兼顾实时性。

单机到集群
单机的优化，主要是focus在两点，可以很快的写进去、很快的传递到客户端。但单机肯定有极限，怎么扩展到集群上？集群关注扩展性，怎么样通过加一个broker加一个DB，整个的容量就上去了？

因为我们的重点是broker，最简单的方式，就是像web一样加一个负载均衡，堆上三个broker，随便发producer。这样有一个问题，消息顺序没有办法做保证。比如我发了一二三，三个消息，我希望收到的时候也是一二三，如果随便放到某一个broker的话，可能最终存储到DB的时候变成二一三，顺序会不一样，会面临这样的问题。还有一个问题，你会发现单机的优化就不再有效。包括消息轮询和消息缓存。
因此，一个topic，我希望它由一台broker做处理，否则很多优化没有办法做。那么能不能我把一个topic连到某一台broker上比如我有两个topic，有两个broker，分配一下，这样前面的一些消息顺序、单机优化是继续有效的，这种有一个问题，它的topic的吞吐会始终小于单个broker的吞吐。所以这边会引入一个topic partition。如果你希望这几个消息是有序，你就发到同一个partition，不同的partition之间是没有关系。如果你希望一二三是有序的，就把一二三放到同一个partition，四五六可以发到另外一个partition，这样可以兼顾你的吞吐又高顺序又可以保证。它还会跟消费者牵涉到，消费的时候一个partition也只能被一个group里面的一个消费消费到，这样才能保证partition里面是顺序处理的。

刚才我们讲消息表的存储，那个表是针对一个partition，不是针对一个topic，一个topic可能有十个partition，就会有十张消息表，每张有一个partition。所以我们最终可以看到，我希望不是把topic粘到broker上，我不需要这么强的约束，我只需要把topic点partition，某一个partition连到broker就可以了，这样我前面的优化都可以做到，消息顺序这些语义也可以保证到。拆了以后还有一个好处是粒度更细了，更容易做负载均衡。比如这个broker撑不住了，把这个partition迁走更方便。

基于Lease的集群管理
图5是我们希望达到的效果，怎么样做到？我可以把这个partition粘到某一个broker上去，这是集群的管理要做一些事情，就是我怎么样通过metaServer，把broker、topic管理好，这些partition是你管理的，你宕了之后可以balance到另外的broker上去。consumer也是，这是metaserver需要做的事情。

最基本的要做这样一些事情，broker加入和退出时要做什么，原来partition是你管理的，你宕了我要做什么事情。另外consumer加入、退出，你要做什么事情，因为消费者也会争抢消息，需要一定的管理。另外是partition怎么样做动态的分配，基本的一些事情。

这一块我们是基于lease去做，lease简单说就是一个有时限的lock，你lock了以后不用unlock，比如说给你一个10秒的lease，10秒之后自动的unlock，如果你说我需要这个lease，你可以去续，由metaserver决定要不要让你续。用lease的目的是说生成一个消息的路由表，就是说我这个消息从Producer出来以后先发给哪个broker，这个broker写到哪个DB，然后这个消息发给哪个consumer，因为这个是根据状态不断的做调整的。有的时候你是写到这个broker，不够了写到另外一个broker。这个时候是这个消费者在消费，如果这时候又起来一个消费者，可能一半的partition要分给另外一个消费者。通过这个lease来生成这个消息的路由表，就是producer到broker到consumer。

比如说看一个简单的例子，这个broker加入或者退出时候，metaserver要做什么事情。我们的broker跟metaserver之间是通过ZK做协调。然后比如左边是只有一台broker的时候，这个时候我有两个topic，T1、T2，各有两个partition。这个时候metaserver会感知到有一个broker在，会分配好这样的路由表。如果这个时候broker起来了，通过ZK，让metaserver感知到你之后，metaServer会把这个topic的partition重新做一个balance。原来一个broker管四个partition，现在只让你管两个，另外两个给另外一个。

因为broker，比如说这个broker是管理这个partition的，所有的消息都会写向你，你会拿到metaserver给你的lease，就是告诉你这个partition是你管理的，但是你要不断的续。metaserver如果这个时候想说这个partition不给你了，只要是broker来续的时候告诉你不续给你了就OK了。整个集群里面的交互是比较简单的，根据这个lease根据这个时间到达这个时间点之后，大家认为的集群状态肯定是一致的。比如说metaserver告诉broker说，这个lock只给你10秒，10秒之后不管这个broker能不能跟metaserver做通讯，跟ZK做通讯，它都会释放掉这个lease，也就是它不再接受写给它的这个partition的消息。这样整个集群协调很简单。

这样的话，你消息发送的时候做什么？producer需要不断的定时去刷这个路由，就是它会从metaserver这边刷一个路由过来，它就知道这个partition怎么去找broker，然后broker拿到这个partition的lease之后，要定时去续这个lease，只要说broker能够续上这个lease，那么producer发给它的消息它都可以处理。一旦metaserver要做一些调整，这个partition不让你管理了，只要这个lease不给就可以了。然后会发生两个事情，producer会拿到一个新的路由表，它知道这个消息要写到另外一个broker。broker那边会发现lease过了，这个时候你写给我我会告诉你现在不是我，你要写给另外一个broker。然后另外一个broker会拿到这个lease，通过这样一种机制，broker加入以后就可以提升整个消息的吞吐量。

consumer的加入不太一样的地方是说，它不是通过ZK去发现这个consumer起来了，那是通过什么？因为consumer，正好是lease，你要续lease也是通过metaserver去续的，那么只要把续lease这个操作做一个类似心跳一样，你就可以发现这个consumer起来了，consumer一起来它就问metaserver说，我想消费这个partition可不可以？这样通过获取lease或者续lease这样一个操作，metaserver就可以感知到这样一个consumer，就不需要借助ZK。这个时候lease的操作，metaserver发现以后，原来是有一个消费者消费两个partition，这个时候又起来一个消费者，就把一个partition分给另外一个消费者去做处理就可以了。

所以说，消息接收的时候，消费者也要刷这个消息的路由，知道我这个partition应该问哪个broker去要。然后consumer会连上broker说我需要消费这个partition，然后broker那边也会去通过这个metaserver续这个lease，这个是跟发送消息是一样的。然后如果说新的consumer起来以后，consumer这边也要续lease，它会续不到，这时候它就不会消费这个partition，会把这个partition让出来，让给其他的消费者去处理。

所以集群管理的话，跟KAFKA不太一样的地方是，consumer是不连接ZK的，可能也是在朝这个方向去走。然后通过metaserver去竞争lease，metaserver在这里面其实就是承担了一个大脑的角色，broker是很笨的，producer、consumer也是很笨的，这也是我们在做架构设计里面的一个经验，你在这个分布式系统里面只有一个人是聪明的做决策的，其他的角色都是服从于这个控制者，这样整个系统不会乱。只要控制好这个metaserver，整个系统都不会出乱子。broker是很笨的，你告诉它要处理就处理，不处理就不处理。如果每个人都很聪明就很容易出现一些不一致的情况，整个会变得不可控。这个也是我们希望metaserver对这个集群有非常灵活的控制能力，它是整个Hermes的核心。

把ZooKeeper，metaserver和（story）这边加在里面。还有一个没讲的点就是说，刚才讲broker是通过ZK去跟metaserver做通讯，实现HA。那么metaserver本身是怎么做HA的？metaserver的话就是说，它首先因为客户端要不断要刷路由，不断要访问这个metaserver，那肯定它本身是HA的，它怎么做？比较简单，跟metaserver的很多的都是走HTTP协议的，跟broker都是TCP，自定义的协议，跟metaserver是走HP，因为操作很简单，比如刷一个路由，要续一些Lease。所以它是放在一个域名下，通过这个域名来做HA，就比较简单。比如我这个环境里面的metaserver就是有一个固定域名，客户端起来以后就是通过这个固定域名扫出一堆IP来，然后任意找到一台就可以了。metaserver本身也是会分配，有一些topic是这个metaserver管理，有些是另外的管理。但是最终，只有一个人能够生成最终的消息，就是master，但是它会通过ZK去竞争说哪个metaserver是master，取路由的时候可以找到任意一台metaserver，如果它不是master，它会把你转到master的metaserver上去。

总结
所以说，消息队列的话，当然其实有很多的其他的一些事情，从我们想优化来说的话，我们需要关注几个事情，一个是消息的写入，我希望消息写入是很快，通道是很大，所以我们做了很多批量。然后在表的设计上做很多insertOnly这样一些事情。另外我希望消息的投递是非常及时的，延迟很短，你一直发出消息都可以收到。比如说我们做了一些partition Stick，stick到broker上去，很多优化都可以去做。写入事件，可以截获你，你不需要做轮询，可以做Long Pulling，很多事情是可以缩短整个因为写到DB再拿出来的时间的延迟。另外一个是在集群管理方面，我们是借助这个lease来做。整个lease实现起来会比较简单，当然lease会有一些它的问题，比如我拿到这个lease是二十秒，如果你不做额外的处理，你这台机器宕机了，20秒之内是没有办法给其他人的，因为你不知道是宕机了还是网络断了。所以我们会做一些主动释放lease的事情，把lease带来的时间延迟降到最低。归出来另外一种partition的顺序。我想可能还是只能做双选或者三选的方案。