设计消息中间件时我关心什么?(解密电商数据一致性与完整性实现,(2)

1. 开启事务.

2. 业务操作,比如说订单进行持久化等等动作.

3. 生成消息,并存储,这和业务操作是在同一个事务里.

4. 事务提交.

5. 消息真正发到出去.消息如果发送成功了,会将消息表里的消息删除,而此时如果消息发送失败了,后台有一个任务会把消息表里面发送失败的消息重新进行发送,这样最终达到一致性,保证业务操作成功了,消息一定能发出去.

(小编：更多了解分布式事务的实现,可参看文末推荐阅读的文章)

现在看看这种模型的优缺点.这种模型 API 非常简单,业务开发只需要使用 sendMessage 这个简单的 API,不需要关心事务等.同时运维也非常简单,我们的做法是公司的 DBA 给所有 DB 实例上预初始化一个消息库,业务根本不用关心,对业务完全是透明,API 把这些封装在底下,使用起来还是非常简单. 但是这样有另外一个问题,就是存储成本.本来只有业务操作访问 DB,然后还要持久化消息.原来承受一个 QPS 现在可能只能承受一半了,所以对数据库操作还是略重一些

另外,有的场景中,可能不仅做数据库操作,还调用了 RPC.这样的动作是不能放在一个数据库事务里的,所以对于这种场景就不能满足了.现在遇到这种情况只有把 RPC 这种操作拆出去了.所以这种模型的优点就是使用方便,但是有些限制.

还有另外一种实现一致性的方法.

1. 发新的消息,直接发给 broker,这个消息发给 broker 并不立即将消息投递出去.

2. 做本地的操作

3. 再调 broker 的接口,这一步真正把消息发送出去.如果这个时候,即使第二步操作成功,第三步发送失败了,第一步发送给 broker 的消息就是一个未决状态.

4. broker 反过来询问 producer,那条消息是发还是不发出去呢?这种模型就不需要一个将一个消息库放在业务库同实例了,比较灵活,成本也更低些.但是业务使用的复杂度可能要高一些,需要提供一个接口供 broker 反查.

容错

讲完了一致性,再来看看容错.broker 会有不同的集群,producer 发消息有一个优先级,默认消息优先发到本机房集群,本机房宕掉或者其他什么原因不可用再向别的机房进行发送.本机房出故障自动不向本机房发送,自动熔断故障机房.后台系统里面可以按照 subject 指定路由到特定的 broker 集群.

Broker 消息中间件设计

消息中间件要支持非常多的 subject,全公司都在使用消息中间件,各业务开发水平也参差不齐,如果有的系统弄了一个死循环,疯狂的发消息就会给系统带来不可控的压力,所以中间层需要做好隔离.其次,业务使用消息中间件可能会遇到各种各样的问题,需要辅助工具进行诊断.最后还需要全面的监控能力.?

隔离

隔离包括配额和调度.Producer 给 broker 发消息的时候,每一个 subject 需要给它多少配额,QPS 一旦高于这个配额,做什么处理?

这个地方我们也踩了一个小坑.假设给每个 subject 3000 QPS 限制,最初 producer 端的设计没有考虑配额这种情况,配额生效之后,达到 3000 QPS broker 开始拒绝消息,也就是返回异常. Producer 一般的设计遇到这种异常时候就不断地重试,这种一拒绝 producer 就不断地重试,雪上加霜,带宽都要打满了.

公司业务部门比较多,因为不能要求所有 producer 都立即配合升级,于是我们做了改进,producer 达到 QPS 不是立即拒绝发送过来的消息,而是拖一会儿,这样来避免将中间层拖垮.当然最好的方式是 producer 进行配合,当 broker 超配额,producer 降低发送速率.

还有一种情况,我们有很多个 subject,有 180 个 consumer group 进行订阅,如果 QPS 达到 100,就是 1.8 万,如果 QPS 达到 1000,就是 18 万,180 倍的增长,所以怎么与其它 subject 进行隔离很重要.?

我们第一版做的很简单,用线程池隔离,每个 subject 分配一个线程池,这种做法隔离效果是很好,但是 subject 不断地增长,资源就不够用了.这个问题抽象来看就像操作系统的 Scheduler 线程调度器,每一个队列可以想象成 OS 里的线程,然后系统用一些来发送这些队列,这些线程就对应 CPU 的 core.我们就模仿 Linux 的 Scheduler 实现了个调度,可能实现水平关系,但是最后测试发现效果很差,量一大起来,队列就堵住了,完全发不出去消息.

最后我们看 actor 这种模型,系统里可以跑成千上万的 actor,它肯定也有一个调度器,最后就模仿 akka 的调度器,它叫 dispatcher,实现了调度的策略.

可治理

像刚才配额都是可以动态调整,不能消息量突然上来了,重启消息系统来调整.还有消息可靠级别,当消息上线时,我们要关心消息 QPS 能达到,能容忍多少丢失?如果这个时候消息中间件出问题了,我们就可以根据上线时可靠级别给有的消息降级.

降级也有很多种策略,比如仅仅给投递一次,如果中间件出问题,这种消息就投递一次算了,不管你是否消费成功,都不给你重发.还有重发次数,比如有的消费者那边有问题,消费不成功,比如消费格式变了,重发一天也消费不了,就可以实时的去调整消息的重发次数.还有可以按多少比例给它发消息,比如说 50%,那么 50% 的消息就给抛弃.

还有日志,为了好查问题,每条消息都有轨迹日志,出问题的时候就可以有选择的是否保存这些日志了.一个公司不可能所有消息都要求 100% 可靠.比如订单支付的消息级别是最高的,但是搜索,比如说现在有报价,代理商帮旗下所有的酒店价格变了一下,关心它的系统就要受到价格的更新,这是通过消息广播出去的.这个消息,它的变动是比较频繁,QPS 也很高,丢了一两条消息,可能又被后面的消息覆盖了,它的可靠级别就比订单的级别要低,这个时候遇到问题,为了保护订单消息,肯定首先对它进行降级.

（编辑：ASP站长网）