使用事件和消息队列实现分布式事务（转+补充）

虽然本文并不是笔者原创，可是咱们在非强依赖的事务中原理上也是采用这种方式处理的，不过由于没有仔细去总结，最近在整理和总结时看到了，故转载并作部分根据咱们实际状况的完善和补充。

不一样于单一架构应用(Monolith), 分布式环境下, 进行事务操做将变得困难, 由于分布式环境一般会有多个数据源, 只用本地数据库事务难以保证多个数据源数据的一致性. 这种状况下, 可使用两阶段或者三阶段提交协议来完成分布式事务.可是使用这种方式通常来讲性能较差, 由于事务管理器须要在多个数据源之间进行屡次等待. 有一种方法一样能够解决分布式事务问题, 而且性能较好, 这就是我这篇文章要介绍的使用事件,本地事务以及消息队列来实现分布式事务.

咱们从一个简单的实例入手. 基本全部互联网应用都会有用户注册的功能. 在这个例子中, 咱们对于用户注册有两步操做: 
1. 注册成功, 保存用户信息.
2. 须要给用户发放一张代金券, 目的是鼓励用户进行消费.
若是是一个单一架构应用, 实现这个功能很是简单: 在一个本地事务里, 往用户表插一条记录, 而且在代金券表里插一条记录, 提交事务就完成了. 可是若是咱们的应用是用微服务实现的, 可能用户和代金券是两个独立的服务, 他们有各自的应用和数据库, 那么就没有办法简单的使用本地事务来保证操做的原子性了. 如今来看看如何使用事件机制和消息队列来实现这个需求.(我在这里使用的消息队列是kafka, 原理一样适用于ActiveMQ/RabbitMQ等其余队列)

咱们会为用户注册这个操做建立一个事件, 该事件就叫作用户建立事件(USER_CREATED). 用户服务成功保存用户记录后, 会发送用户建立事件到消息队列, 代金券服务会监听用户建立事件, 一旦接收到该事件, 代金券服务就会在本身的数据库中为该用户建立一张代金券. 好了, 这些步骤看起来都至关的简单直观, 可是怎么保证事务的原子性呢? 考虑下面这两个场景:
1. 用户服务在保存用户记录, 还没来得及向消息队列发送消息以前就宕机了. 怎么保证用户建立事件必定发送到消息队列了?
2. 代金券服务接收到用户建立事件, 还没来得及处理事件就宕机了. 从新启动以后如何消费以前的用户建立事件?
这两个问题的本质是: 如何让操做数据库和操做消息队列这两个操做成为一个原子操做. 不考虑2PC, 这里咱们能够经过事件表来解决这个问题. 下面是类图. 

EventPublish是记录待发布事件的表. 其中:
id: 每一个事件在建立的时候都会生成一个全局惟一ID, 例如UUID.
status: 事件状态, 枚举类型. 如今只有两个状态: 待发布(NEW), 已发布(PUBLISHED).
payload: 事件内容. 这里咱们会将事件内容转成json存到这个字段里.
eventType: 事件类型, 枚举类型. 每一个事件都会有一个类型, 好比咱们以前提到的建立用户USER_CREATED就是一个事件类型.
EventProcess是用来记录待处理的事件. 字段与EventPublish基本相同.

咱们首先看看事件的发布过程. 下面是用户服务发布用户建立事件的顺序图. 
1. 用户服务在接收到用户请求后开启事务, 在用户表建立一条用户记录, 而且在EventPublish表建立一条status为NEW的记录, payload记录的是事件内容, 提交事务.
2. 用户服务中的定时器首先开启事务, 而后查询EventPublish是否有status为NEW的记录, 查询到记录以后, 拿到payload信息, 将消息发布到kafka中对应的topic.
发送成功以后, 修改数据库中EventPublish的status为PUBLISHED, 提交事务.

下面是代金券服务处理用户建立事件的顺序图. 
1. 代金券服务接收到kafka传来的用户建立事件(其实是代金券服务主动拉取的消息, 先忽略消息队列的实现), 在EventProcess表建立一条status为NEW的记录, payload记录的是事件内容, 若是保存成功, 向kafka返回接收成功的消息.
2. 代金券服务中的定时器首先开启事务, 而后查询EventProcess是否有status为NEW的记录, 查询到记录以后, 拿到payload信息, 交给事件回调处理器处理, 这里是直接建立代金券记录. 处理成功以后修改数据库中EventProcess的status为PROCESSED, 最后提交事务.

回过头来看咱们以前提出的两个问题:
1. 用户服务在保存用户记录, 还没来得及向消息队列发送消息以前就宕机了. 怎么保证用户建立事件必定发送到消息队列了?
根据事件发布的顺序图, 咱们把建立事件和发布事件分红了两步操做. 若是事件建立成功, 可是在发布的时候宕机了. 启动以后定时器会从新对以前没有发布成功的事件进行发布. 若是事件在建立的时候就宕机了, 由于事件建立和业务操做在一个数据库事务里, 因此对应的业务操做也失败了, 数据库状态的一致性获得了保证.
2. 代金券服务接收到用户建立事件, 还没来得及处理事件就宕机了. 从新启动以后如何消费以前的用户建立事件?
根据事件处理的顺序图, 咱们把接收事件和处理事件分红了两步操做. 若是事件接收成功, 可是在处理的时候宕机了. 启动以后定时器会从新对以前没有处理成功的事件进行处理. 若是事件在接收的时候就宕机了, kafka会从新将事件发送给对应服务.

经过这种方式, 咱们不用2PC, 也保证了多个数据源之间状态的最终一致性.
和2PC/3PC这种同步事务处理的方式相比, 这种异步事务处理方式具备异步系统一般都有的优势:
1. 事务吞吐量大. 由于不须要等待其余数据源响应.
2. 容错性好. A服务在发布事件的时候, B服务甚至能够不在线.
缺点:
1. 编程与调试较复杂.
2. 容易出现较多的中间状态. 好比上面的例子, 在用户服务已经保存了用户并发布了事件, 可是代金券服务还没来得及处理以前, 用户若是登陆系统, 会发现本身是没有代金券的. 这种状况可能在有些业务中是可以容忍的, 可是有些业务却不行. 因此开发以前要考虑好.

另外, 上面的流程在实现的过程当中还有一些能够改进的地方:
1. 定时器在更新EventPublish状态为PUBLISHED的时候, 能够一次批量更新多个EventProcess的状态.
2. 定时器查询EventProcess并交给事件回调处理器处理的时候, 可使用线程池异步处理, 加快EventProcess处理周期.
3. 在保存EventPublish和EventProcess的时候同时保存到Redis, 以后的操做能够对Redis中的数据进行, 可是要当心处理缓存和数据库可能状态不一致问题.
4. 针对Kafka, 由于Kafka的特色是可能重发消息, 因此在接收事件而且保存到EventProcess的时候可能报主键冲突的错误(由于重复消息id是相同的), 这个时候能够直接丢弃该消息.

补充点：

一、咱们使用的是rabbitmq cluster；

二、有HA要求，就要防止重复SPOF，而涉及到scheduler的时候，就须要防止重复调度，对此，可使用quartz cluster（能够容忍必定延时）或者分布式事务协调器好比zookeeper，前者相对后者系统结构简单得多，对于这些非强依赖的状况，由于仅在没有等待处理的队列才会到下一个调度间隔，所以进一步使用多线程以及数据库优化以后，延时一般在10ms内所有已经push到对方处理了。

三、对于须要回滚和性能要求极高的业务，上述模式没法直接套用。好比在证券买卖中，延时性是个极为重要的特性，所以不可能在资金冻结以后最长可能超过10ms才发出买单，对于此，建议的设计是这些功能总体VIP化+可信请求处理。而对于须要回滚的操做，其中一种处理方式是，对于每一个业务功能，都增长一个对应的对冲逻辑，当对端处理失败的时候，push相应的对冲消息便可。

四、对于某些在短期内发出大量请求的业务逻辑，好比对于一个帐户，一会儿发出成千上万请求的业务逻辑，如此一般服务端是经过线程池处理，若是所有请求到到一个线程池中处理，占据了全部的线程和数据库链接，会致使其余用户所有出现hang的状况，这种状况下，服务端须要采用pools of single thread executor，而非通用pools的模式，可能还要进一步针对帐号进行二次mod。

在一个系统中，有可能这三种方式都涉及么，若是确实须要分布式系统的规模，那么大部分状况下会须要两种结构并存，某些行业和场景中，三种结构都须要。

对于分布式系统的架构，必定要先按照子系统独立拆分（这是为了减小升级、开发、运维复杂性和成本），而后按照业务数据分库分表拆分（这是为了尽量晚的引入分布式事务，分布式事务一旦引入，极可能使得开发、测试成本剧增，并且在绝大部分状况下不少系统是拆分过分、架构不合理，而非没有采用分布式的缘由），分布式系统绝对不是、也不能由于数据量、并发数很大就构造分布式系统，而是由于业务子系统太多、太复杂、又不得不进行大量的子系统间交互才衍生而来。各独立运行系统间的交互是不能有太多功能的，不然业务架构的设计上就存在疑虑，为了分布式而分布式会使得系统的维护成本极为高昂，同时分布式系统大量依赖于MQ和RPC框架，可是不是使用了RPC和MQ就必定要将系统复杂到分布式系统。