欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

分布式事务消息中心TMC

程序员文章站 2023-12-29 14:59:34
...

系统原理

     贷款和理财是51信用卡目前最主要的业务。金融相关的应用,往往对数据的一致性有着较高的要求,通常对DB的操作都是用事务来保障。但是在分布式的环境下,要保持事务的一致性从来都不是一件容易的事,传统通过两阶段、三阶段提交方式实现的XA事务由于代价太高,性能损失太大,在互联网公司中并不常用,而更常用的是实现最终一致性,即系统允许短暂的不一致,但是最终能达到一致的状态。

 

     51信用卡金融研发团队之前对跨微服务的调用就有一些方法,在执行业务逻辑前先插一张状态日志表,在执行成功后更新日志表里的状态,如果有中途失败的交易,后续也可以通过扫描日志表进行更正。TMC做的事就是业务里做的逻辑抽象独立出来,让业务专注于业务开发,无需再自己维护状态。下面简述其原理:

     两个事务运行在不同的进程上,为了平衡上下游系统压力差、削峰填谷都会使用MQ。但是直接使用MQ并不能实现分布式事务,下面举个例子说明:

     假设两个独立账务系统ABA要向B100块钱,直接使用MQ的如下图所示:

 

分布式事务消息中心TMC
            
    
    博客分类: 分布式事务  

 

     操作分四个步骤

1.    系统A更新DB,扣减100

2.    publish一条消息进MQ

3.    MQ将消息发给系统B

4.    系统B更新DB,加100

 

     分布式系统,任何一个环节都有可能失败,下面一一分析:

   步骤1A更新DB前挂了,无影响,因为还没扣钱

   更新DB后投进MQ前挂掉,消息丢了,100块不知道去哪了

   A成功写进MQ,但是MQ这时候挂了,同样消息丢了

   MQ成功投递给B,但是B写进DB之前B挂掉,同样100块不知道去哪了

   B成功写进DBB系统加100块,这种情况正常

 

     因此要实现分布式事务,就是要考虑消息在各个环节中都不丢,任一系统在挂掉的情况下,都能保证能够把消息找回来,并且流程还能够继续下去。

 

     下面开始介绍我们的实现方案,为了记录状态,我们引入了一个中间人,即TMCTransactionMessage Center),示意如下图:

 

分布式事务消息中心TMC
            
    
    博客分类: 分布式事务  

 

 

     现在的步骤如下:

1.    系统A publish一条消息给TMC

2.    TMCDB中记录下这条消息

3.    系统A执行本地事务更新DB

4.    系统A如果本地事务执行成功,告诉TMC这条消息submit,如果本地事务执行失败,告诉TMC这条消息rollback

5.    TMC对于成功submit的消息,开始往MQ进行投递,等待ack,如果一段时间没有收到ack,会继续投递该消息

6.    MQ将消息push给消费者系统B

7.    系统B执行事务更新DB

8.    系统BTMCack,告诉这条消息消费成功

 

     正常的流程就是这样,下面分析下各个环节可能出现的异常,以及异常流程。

   系统A publish消息给TMC之前挂了,无影响,因为啥都还没做

   系统A执行本地事务之前或者之后挂了,这时候会涉及到9. check过程,因为TMC已经记录下这条消息,但是后面没有收到submit或者rollback,不知道这条消息的状态,只好过了一段时间去问系统A,这条消息到底是什么状态,如果A是执行事务之前挂的,会答复rollback,这时候TMC删除这条消息,如果是执行事务之后挂的,答复commit,这时候TMC将这条消息转变成可消费状态,开始向MQ投递

   TMC将消息投递进MQ没有成功,或者MQ挂了,消息丢了。这时候消费者肯定收不到消息,也就不会进行ackTMC会在一段时间后继续将该消息投递进MQ,这个过程会持续好几次,超过一定次数标记消费失败,给出报警

   消费者系统B受到消息,更新DB后挂了,ack没有给到TMC,这就像上一条所说,TMC会继续投递该消息,因此消费者一定要实现幂等应对重复消息

 

 

系统设计

     TMC的设计和51信用卡公司内部在用的系统紧密结合,使用了MySqlBase框架(51信用卡内部的一个微服务框架,最主要功能是RPC和服务发现)、ConsulRabbitMQ,这里做一个详细一点的介绍,帮助大家理解系统,在测试环境能够更好的调试。系统结构图如下:

 

分布式事务消息中心TMC
            
    
    博客分类: 分布式事务  

 

 

   TMC Server是一个标准的基于Base的微服务

   TMC Client是一个基于BaseClient,通过maven分发,包含ProducerConsumer功能

 

     系统之间的交互可以描述为以下几点:

-     Producer通过微服务名tmc以及client端配置的tmcTag,从Consul上获得到TMC Server的地址,然后通过Base RPC调用TMC Server发送或者submit消息,记住:ProducerRabbitMQ没有关系,只和TMC Server打交道,所有操作都是通过Base RPC进行

-     对于状态未知的消息,TMC Server会通过发送者的微服务名以及Topic里配置的Tag51信用卡内部根据tag区分环境),从Consul上获得业务应用的地址,通过Base RPC去向业务应用check消息的状态

-     TMC Server会将成功submit的消息投递进RabbitMQ

-     对于消费者,TMC Client会在启动的时候去监听配置要消费的RabbitMQ的队列,当然,消费者的配置只需要配置TMC里的topic,而TMC Client代码会自动转换成RabbitMQ的队列名,所以:消费者是直接从RabbitMQ获取数据的,需要配置TMC所使用的RabbitMQ集群地址

-     Consumer通过微服务名tmc以及client端配置的tmcTag,从Consul上获得到TMC Server的地址,然后将消费成功的ack发送给TMC

 

 

数据库表结构

       表结构很简单,主要包括应用表(T_Tmc_PubSub)、Topic表(T_Tmc_Topic)、订阅表(T_Tmc_Topic_Sub)、消息表(T_Tmc_Msg)、消费表(T_Tmc_Msg_Sub),关系如下图

 

分布式事务消息中心TMC
            
    
    博客分类: 分布式事务  

 

   应用表,AppId即对应客户端clientIdAppKey对应clientKeyAppName即是微服务名,另外还有报警人邮箱和手机号信息

   Topic表,每个topic都需要指定tag,这个表里的tag应该和客户端的clientTag对应,Mode表示topic模式,0是单队列,1是多队列,另外还有最大重试次数和重试间隔配置,这两个是针对check生产者的重试

   订阅表,即表示哪个消费者订阅了哪个topic,也有最大重试次数和重试间隔配置,这两个是针对消费者ack的重试

   消息表,消息分表存放,目前32张,将来不够很容易扩容,消息存放在哪张分表由topic表的TableIndex指定,MsgId是在消息落库时候生成,里面包含里topicId和生产者IdMsgKey是生产者指定,推荐存放一个具有一定业务意义的,用于做幂等控制,消息内容通过base64编码后存放在body字段,由于mysql限制,消息内容不能超过10KDelay是延迟消息时候需要设置延迟的时间,RetryNextRetryTimecheck生产者的次数和下次的时间,大多数消息都是生产者主动submit上来,所以这两个字段不常用

   消费表,在生产者submit消息之后,有几个消费者订阅了该topic,就会在该表中生成几条数据,同时消费表也是分表,分表规则和消息表一样。消费表并不复制消息的内容,只是记录每条消息每个消费者消费的状态,该表的RetryNextRetryTime是已经向RabbitMQ投递的次数,和下次再投递的时间

分布式事务消息中心TMC
            
    
    博客分类: 分布式事务  

 

 

单队列 or 多队列

       TMC刚上线的时候只有单队列模式,后来有业务提出问题,单队列可能造成低优先级消息阻塞高优先级消息的情况,因此又加了多队列模式,如上面的图。

     单队列是指每个消费者,会订阅不同的TMC topic,他所的所有TMC消息,会放在同一个RabbitMQ队列里,队列的命名规则是tmc.sub.{clientId}.{clientTag}。有些消息,可能时效性要求比较高,量比较小,而有些消息,时效性要求比较低,但量比较大,放在同一个队列里就会影响后面重要消息的消费时效性。

     而多队列每一个消费者,每一个TMC topic都会放在一个单独的RabbitMQ队列里,命名规则是tmc.sub.{clientId}.{clientTag}.{topic},这样就不会造成高优先级消息被低优先级消息阻塞的问题。

     由于单/多队列的命名规则不同,因此在client端消息入口不同,单队列通过实现MessageConsumer接口,而多队列通过打@TmcListener标注。

 

 

失败和重试

     事务消息的事务,主要体现在生产者和Server之间,因为只有生产者本地事务失败是可以回滚消息的,而只要生产者事务成功提交,消费者是一定要消费掉的,超过一定时间没有消费掉,会给出报警,让人工介入处理。

 

     重试主要用在两个地方

1.    生产者没有主动submit/rollbackServer存在未知状态消息,需要不断去生产者那边check改消息的状态

2.    消息已经成功被submitServer已经投进MQ,但是消费者一直没有ack,需要继续投递,直到被ack

 

     重试的程序实现是,每个分表都有一个扫表线程,这个扫表线程落在哪台机器上,通过Base里的分布式锁抢占,即抢到分表的分布式锁的机器负责扫该表。那每次扫出哪些消息?取决于StatusNextRetryTime,只有Status等于正在进行的并且NextRetryTime小于当前时间的会被扫出来。

     NextRetryTime的更新规则是now + Retry *RetryIntervalRetry是已经重试的次数,这样的重试时间是被逐渐拉长的,举个例子,如果起始时间是0RetryInterval10s,那一次重试是在10s,第二次重试是在30s,第三次重试是在60s,第四次重试是在100s。这样设计是考虑到,当前无法消费成功的消息,短期内立即重试大概率仍然不能消费成功,所以逐渐拉长重试的时间。

     凡是投递超过MaxRetryTimes的消息,更新状态为失败,不在继续重试,并且发送报警给登记的业务负责人。

 

 

消息查询和重新激活

       在控制后台支持通过MsgIdMsgKey以及状态和时间范围查询消息的状态,并且对于失败的消息,可以一键重新激活,重新开始check或者投递。控制后台拥有的功能消息查询,消费查询,消息重试,投递重试,很容易搞混,下面来说明一下。

-     消息查询,是查询消息表的数据,查询消息是否入库,是否被submit,以及消息的内容(原始base64和解码过的字符串格式都能看到),check的次数,下次check时间

-     消费查询,是查询消费表的数据,是否生成消费记录(如果没有说明生产者还没submit),是否被ack,投递的次数,下次投递的时间

-     消息重试,是生产者还没有提交,超过check重试的失败消息,进行重新激活,重新开始check生产者消息的状态

-     投递重试,是超过投递次数,消费者还没有ack的失败消息,进行重新激活,重新开始往MQ投递

 

 

额外的小功能

     业务在使用过程中,提出了一些特殊的小需求,TMC也试着满足他们的需求,列举如下:

-     支持demo环境,由于demo环境consul和线上隔离,因此无法通过consul来获取地址,只能在client端配置上TMC Server地址,同时新消息Server也会记录下发送者的地址用于check

-     测试环境支持生产、消费不同tag,由于公司测试有好几个环境devstablek8stest等等,为方便开发和测试同学,生产和消费都支持指定tag

-     延迟消息,生产者可以指定delay参数,Serversubmit后到达delay的时间才会开始投递

-     顺序消息(未上线),之前有业务提过,想要支持顺序消息,就尝试把RabbitMQ换成支持顺序消息的MQ,如RocketMQ,但后来业务改了方案不需要了,这份代码就暂时没合入主分支,没有上线

上一篇:

下一篇: