如何用 RocketMQ 打造金融级消息服务平台？微众银行这么做...

阿里妹导读：近年来，随着微服务架构的流行，分布式消息引擎在物联网、分布式事务、实时计算和大规模缓存同步等场景中的应用日益增多。

本文将分享微众银行基于RocketMQ构建消息服务平台的实践，通过添加诸多高级特性来解决消息收发过程中遇到的各种问题。你将了解到：金融行业服务架构的演进历程、微众银行的消息服务架构以及基于RocketMQ定制的消息高级特性。

银行应用架构的演进历史

不管是银行的系统还是其他一些传统企业的系统，他们在最早的时候都使用到了服务总线，即ESB或者某种形式存在于SOA架构中，目的是把所有的服务都串起来，让服务之间能够形成一个调用。但这类服务架构其实是比较重的，所有的服务架构都要经过总线，总线成为了架构上的瓶颈。很多商业化的ESB总线大家可能都用过。从服务调用的维度来看，银行的应用架构的演进经历了以下3个阶段。

第一阶段：90年代中后期分布式架构

这个阶段的架构具有以下3个特点：

1.将总行的集中式系统在各个省分行分别都部署一套，每天晚上再以批量处理的方式将各省数据进行集中。

2.这种架构的方式能够最快的解决联机性能问题, 但存在跨省转账交易无法实时到账的问题。

3.系统发布的实时性是硬伤。

第二阶段：2000-2010年集中式总线架构

到了2012年以后，随着各个海外开放平台获得的巨大成功，一线互联网公司都逐步将自己的接口开放出来，并实施了开放平台生态圈战略，从而推动了SOA服务化的快速发展。

左边是之前的传统银行集中式总线架构，右边是互联网服务化架构，包含了开放平台、服务注册和发现，以及服务化产品系统。

通过开放平台对外提供接口暴露，可以发现这种架构在保障传统银行系统稳定性的同时也可以满足互联网金融需求的快速迭代实施，并且也使用了新兴的互联网分布式技术，来降低开发和运维的成本。

####微众银行的消息服务架构

微众银行基于Apache RocketMQ构建了自己的分布式消息服务架构，我们以RMB（Reliable Message Bus）为接入层，以基于Apache RocketMQ定制开发的WeMQ（WeBank Message Queue）为消息服务核心，通过GSL（Global Service Location）进行服务定位，通过SGS（Service Governance System）进行服务请求和服务响应的服务治理，整个分布式链路的追踪日志会上报到Log中。

接下来，我们来看看我们基于RocketMQ改造使用到的常见的消息服务模式：

单播/多播pub-sub模式

Consumer可以是一个或者多个，但是一个消息会被多个不同系统的其中一个consumer收到。

广播pub-sub模式

多个在线的Consumer会同时收到广播消息。

Active/Standby消费模式

生产者只有一个，消费者有多个，但是作为HA，只有一个Active，其他都是StandBy。当Active挂掉一个，Standby会迅速接管。

request-reply模式

发送请求-等待响应结果。在发送方做了一个线程的等待，要等待结果的notify。

在分布式消息系统的构建过程中，基于业务的需求，我们在RocketMQ的消息系统中添加了多项高级特性，包括多中心多活、灰度发布、熔断机制、消息存活期、流量的权重、消息去重、惊群效应问题的解决、背压模式、消息服务治理、MQTT消息服务等。

基于RocketMQ添加的一些消息高级特性

同城多活

DC级别的多活希望解决的问题是，不仅消息不能丢，还要保证服务不能中断。这里有两个层面的故障，一个是应用全部宕机，那么希望被其他IDC的应用能够迅速来接管消息，另外一个是消息中间件宕机，那么希望生产者能够切换到其他IDC的中间件进行发送，并且这个中间件的消息在其他IDC有备份，能够进行消费。微众已经通过IDC断网演练检验同城多活能力。

灰度发布

灰度发布希望解决的问题是，同一个消费组内不同的实例有监听不一样的topic时，能保证不同topic的消息被正确的实例消费。

熔断机制

当希望消息的堆积到一定程度时，可能是消费者出现了故障，我们希望能够提醒生产者。

流量权重(自动伸缩Q)

说到流量的权重，有一个问题是，Topic的Q值是在使用过程中手动设置的，当实例的数量超过Q的数量，那么超过部分的实例是收不到消息的。但是，如果你的实例数量小于Q的话，它们之间会由于负载均衡分Q。根据负载均衡算法，分到的Q可能是不一致的。比如有的分到2个，有的分到3个。在这种集群消费的情况下，就会出现处理的不对等。比如当大流量到来的时候，分到3个Q的那个实例可能会出现一些问题，比如挂掉了。

所以我们希望，不同的实例拿到的消息量应该是对等的。所以，流量权重希望解决的问题是，随着实例数的动态增加和减少，能够动态调整consumeQueue的数量，不至于出现流量不均匀的情况。因此，我们做了一个自动伸缩Q的功能。默认Topic建成时，Q的数量是1，当启动一个新的实例的时候，会自动扩展一个，停掉一个实例的时候会自动缩一个。从而达到Q个数量和实例的数量是一一对等的。这解决了实例和消息量不对等的问题。

消息去重

在负载均衡的一个很短时间内，当新上一个实例的时候，由于大家分到的Q都是相同的，当前一个分到Q的还在继续拉消息，下一个实例由于负载均衡很快做完，也分到Q，就会去拿这个Q的消息，这个时候就会出现消息的重复。此时，通常会通过Redis等缓存方式进行去重，也可以在Broker上做一个简单的处理，例如用互斥锁，在竞争消费的短时间内，对其进行加锁，抢到锁的才能进行消费，同时占有锁的时间有限制，从而解决消息去重的问题。

消息服务去重原理图

消息的背压消费模式

背压模式示意图

在一些特殊场景下，需要对消息引擎做一些加强，例如背压模式。当消息拉到本地的消费线程池时，会出现一个问题。当要做一些例如DB的写的操作导致出现线程卡死，处理能力会下降，服务出现降级，但是消息还在不停地往本地拉。

这个时候，我们希望达到一种效果，能够根据后续服务的治理能力决定拉的消息数量。当然RocketMQ的ProcessQ也能达到这个效果，但是还不够精细化。因为在金融场景下，交易一旦出现不一致或者超时，会很麻烦。所以我们希望在实时的交易链路上去解决这个问题。于是我们做了一个类似Reactor框架的背压处理，能够根据处理能力实时拉取消息。

消息存活期
当对消息的有效期有要求时，可以在消费消息时对存活时间进行判断，超时则丢弃。

内存模式

对于存活期非常短和对延时要求比较低的消息，我们通过内存模式（不落盘）进行加速，降低延时。

惊群效应问题

因为负载均衡算法在客户端，客户端的连接和断开都会触发消费组内的所有实例会收到notification做负载均衡。比较理想的情况是，一个实例的掉线不能影响到其他实例，当监听的topic比较多时，会出现负载均衡慢的问题，因此我们希望负载均衡收敛到服务端来做，客户端只需要关注topic，不需要关注consumeQueue。

目前，我们团队已经参与到Apache RocketMQ的社区建设中，并对自用的消息服务以社区分支的形式在维护，希望各行业更多的开发者可以一起参与进来，以打造适用范围更广、更好用的分布式消息引擎。

原文发布时间为：2018-01-09
本文作者：陈广胜
本文来自云栖社区合作伙伴“阿里技术”，了解相关信息可以关注“阿里技术”。