消息最终一致性的架构革命

概述

跨服务更新数据是应用开发常见的任务，如果一些关键数据对一致性的要求较高，而业务上也不需要支持回滚的话，那么通常就会采用本地消息表的方式来保证最终一致。许多公司在处理跨服务更新数据一致性问题时，都会先引入本地消息表，后续随着业务场景复杂化，再引入更多的事务模式

本文提出的二阶消息，是一种新模式，新架构，优雅的解决了消息最终一致性的问题。解决同样的一个问题，可以将本地消息表或者事务消息中，上百行的代码简化为大约五六行，大大简化了架构，提升开发效率，具备非常大的优势。

下面我们以跨行转账作为例子，给大家详解这种新架构。业务场景介绍如下：

我们需要跨行从 A 转给 B 30 元，我们先进行可能失败的转出操作 TransOut ，即进行 A 扣减 30 元。如果 A 因余额不足扣减失败，那么转账直接失败，返回错误；如果扣减成功，那么进行下一步转入操作，因为转入操作没有余额不足的问题，可以假定转入操作一定会成功。

采用新架构开发

新架构基于分布式事务管理器 dtm-labs/dtm

完成上述任务的核心代码如下所示：

msg := dtmcli.NewMsg(DtmServer, gid).
	Add(busi.Busi+"/TransIn", &TransReq{Amount: 30})
err := msg.DoAndSubmitDB(busi.Busi+"/QueryPreparedB", db, func(tx *sql.Tx) error {
	return busi.SagaAdjustBalance(tx, busi.TransOutUID, -req.Amount, "SUCCESS")
})

上述代码是 HTTP 接入，gRPC 的接入和 HTTP 基本一样，这里不再赘述，有需要的读者，可以参考dtm-labs/dtm-examples中的例子

这部分代码中

• 首先生成一个 DTM 的 msg 全局事务，传递 dtm 的服务器地址和全局事务 id
• 给 msg 添加一个分支业务逻辑，这里的业务逻辑为余额转入操作 TransIn ，然后带上这个服务需要传递的数据，金额 30 元
• 然后调用 msg 的 DoAndSubmitDB ，这个函数保证业务成功执行和 msg 全局事务提交，要么同时成功，要么同时失败
  1. 第一个参数为回查 URL ，详细含义稍后说
  2. 第二个参数为 sql.DB ，是业务访问的数据库对象
  3. 第三个参数是业务函数，我们这个例子中的业务是给 A 扣减 30 元余额

由于当前 TransOut 业务操作与 TransIn 不再同一个服务，因此可能发生执行完一个操作后，发生进程 crash ，导致另一个操作未执行，此时 dtm 会通过回查 URL ，查询 TransOut 的业务操作是否成功完成。dtm 里面的回查只需要粘贴如下代码即可，框架会自动完成回查逻辑：

app.GET(BusiAPI+"/QueryPreparedB", dtmutil.WrapHandler2(func(c *gin.Context) interface{} {
    return MustBarrierFromGin(c).QueryPrepared(dbGet())
}))

至此一个完整的二阶段消息的业务完成，接入复杂度、代码量比本地消息表等现有方案，都有巨大的优势，已成为这类问题的首选方案。您可以通过以下命令运行一个完整的例子：

运行 dtm

git clone https://github.com/dtm-labs/dtm && cd dtm
go run main.go

运行例子

git clone https://github.com/dtm-labs/dtm-examples && cd dtm-examples 
go run main.go http_msg_doAndCommit

成功流程

DoAndSubmitDB 是如何保证业务成功执行与 msg 提交的原子性的呢？请看如下的时序图：

一般情况下，时序图中的 5 个步骤会正常完成，整个业务按照预期进行，全局事务完成。这里面有个新的内容需要解释一下，就是 msg 的提交是按照两个阶段发起的，第一阶段调用 Prepare ，第二阶段调用 Commit ，DTM 收到 Prepare 调用后，不会调用分支事务，而是等待后续的 Submit 。只有收到了 Submit ，开始分支调用，最终完成全局事务。

提交后宕机流程

在分布式系统中，各类的宕机和网络异常都是需要考虑的，下面我们来看看可能发生的问题：

首先我们要达到的最重要目标是业务成功执行和 msg 事务是原子操作，因此首先看如果在业务完成提交后，发送 Submit 消息前出现了宕机故障会怎么样，新架构如何保证原子性？

我们来看看这种情况下的时序图：

如果在本地事务提交之后，在发送 Submit 前，出现了进程 Crash 或者机器宕机会怎么样？这个时候 DTM 会在一定超时时间之后，取出只 Prepare 但未 Submit 的 msg 事务，调用 msg 事务指定的回查服务。

您的回查服务逻辑，不需要手动编写，只需要按照之前给出的代码进行调用即可，它会到表里面查询，本地事务是否提交了：

• 已提交： 返回成功，dtm 进行下一步子事务调用
• 已回滚： 返回失败，dtm 终止全局事务，不再进行子事务调用
• 进行中： 这个回查会等待最终结果，然后按照前面的已提交 /已回滚的情况处理

提交前宕机流程

我们来看看本地事务被回滚的时序图：

如果在 dtm 收到 Prepare 调用后，AP 在事务提交前，遇见故障宕机，那么数据库会检测到 AP 的连接断开，自动回滚本地事务。

后续 dtm 轮询取出已经超时的，只 Prepare 但没有 Submit 的全局事务，进行回查。回查服务发现本地事务已回滚，返回结果给 dtm 。dtm 收到已回滚的结果后，将全局事务标记为失败，并结束该全局事务。

易用性

采用新架构处理一致性问题，仅需要：

• 定义好本地业务逻辑，指定下一步处理的服务即可
• 定义 QueryPrepared 处理服务，复制粘贴例子代码即可。

然后我们看看其他方案情况

二阶消息 vs 本地消息表

上述的问题也可以采用本地消息表方案（方案详情参考分布式事务最经典的七种解决方案），来保证数据的最终一致性。如果采用本地消息表，需要的工作包括：

• 在本地事务中执行本地业务逻辑，将消息插入消息表并最后提交
• 编写轮询任务，将本地消息表的消息，发给消息队列
• 消费消息，并将消息发给相应的处理服务

两者对比，二阶消息有以下优点：

• 无需学习或维护任何消息队列
• 不需要处理轮询任务
• 不需要消费消息

二阶消息 vs 事务消息

上述的问题也可以采用 RocketMQ 的事务消息方案（方案详情参考分布式事务最经典的七种解决方案），来保证数据的最终一致性。如果采用本地消息表，需要的工作包括：

如果采用事务消息，需要的工作包括：

• 开启本地事务，发送半消息，提交事务，发送 commit 消息
• 消费超时的半消息，对于收到的超时半消息，查询本地数据库，然后进行 Commit/Rollback
• 消费已提交的消息，并将消息发送给处理服务

两者对比，二阶消息有以下优点：

• 无需学习或维护任何消息队列
• 本地事务与发送消息之间的复杂操作需要手动处理，一不小心，可能出现 bug 。而二阶消息则是全自动处理
• 不需要消费消息

二阶消息在二阶段提交方面，与 RocketMQ 的事务消息相似，是受到 RocketMQ 的事务消息启发后提出的新架构。二阶消息的命名，不再复用 RocketMQ 的事务消息，主要是因为二阶消息在架构上有很大的改变，而另一方面，在分布式事务的上下文中，使用”事务消息“这个名字，容易带来理解上的混淆。

更多的优点

对比于前面讲述的队列方案，二阶消息还有很多额外的优点：

• 二阶消息整个暴露的接口，完全与队列无关，只跟实际的业务和服务调用相关，对开发人员更加友好
• 二阶消息不用考虑消息队列消息堆积及其他故障等问题，因为二阶消息只依赖 dtm ，开发人员可以认为 dtm 与系统中其他一个普通无状态服务一样，只依赖背后的存储 Mysql/Redis 。
• 消息队列是异步的，而二阶消息同时支持异步和同步，默认异步，只需要打开 msg.WaitResult=true ，那么可以同步等待下游服务完成
• 二阶消息还支持同时指定多个下游服务

二阶消息未来展望

二阶消息能够大幅降低消息最终一致性解决方案的难度，获得广泛的应用。未来 dtm 会考虑添加后台，允许动态指定下游服务，提供更高的灵活性。如果您原先采用消息队列来做服务解耦，那么这个 dtm 的后台，允许你直接指定某个消息的多个接收函数，无需编写消息消费者，带来更加简单、直观、易用的开发体验。

回查原理剖析

前面的时序图中，以及接口中都出现了回查服务，在二阶消息中，是复制粘贴代码自动处理的，而 RocketMQ 的事务消息，则是手动处理的。那么自动处理的原理是什么？

要进行回查，首先要在业务数据库实例中，建立一张独立的表，里面保存全局事务 id 。在处理业务事务时，会把 gid 写入到这张表。

当我们用 gid 回查时，如果能够在表中查到 gid ，那么说明本地事务已提交，这样就可以返回 dtm ，告知本地事务已提交。

当我们用 gid 回查时，没有在表中查到 gid ，那么说明本地事务未提交，此时可能的结果是两个，一是事务还在进行中，二是事务已回滚。我查了许多关于 RocketMQ 的资料，未找到有效的解决方案。搜到所有解决方案是，如果未查到结果，那么什么都不做，等待下一次回查，如果 2 分钟或者更久的回查，一直都是查不到的，那么认为本地事务已回滚。

上述这种方案有很大的问题：

• 两分钟还查不到 gid ，并不能认为本地事务已回滚，极端情况下，可能发生数据库故障（例如进程或磁盘卡住了），持续时间超过 2 分钟，最后数据又提交了，那么这个时候，数据就不是最终一致了，就需要人工介入处理了
• 如果一个本地事务，已经回滚了，但是回查操作，还会在两分钟之内，按照 10s 左右的时间间隔，不断的进行轮询，会给服务器造成不必要的压力

而 dtm 的二阶消息方案，则彻底解决了这部分的问题。dtm 的二阶消息工作过程如下：

1. 在处理本地事务时，会将 gid 插入到 dtm_barrier.barrier 表中，同时带上插入原因为 committed 。该表有一个唯一索引，主要字段为 gid 。
2. 当进行回查时，二阶消息的操作不是直接查 gid 是否存在，而是再 insert ignore 一条带有相同 gid 的数据，同时带上插入原因为 rollbacked 。此时如果表中如果已有 gid 的记录，那么新的插入操作就会被 ignore ，否则数据会被插入。
3. 然后再用 gid 查询表中的记录，如果查到记录的 reason 为 committed ，那么说明本地事务已提交；如果查到记录的 reason 为 rollbacked ，那么说明本地事务已回滚。

那么对比 RocketMQ 回查时的常见方案，二阶消息是如何区分出进行中和已回滚呢？其中的技巧在于回查时插入的数据，如果回查时，数据库的事务还在进行中，那么插入操作就会被进行中的事务阻塞，因为插入操作会等待事务中持有的锁。如果插入操作正常返回，那么数据库中的本地事务，必定已结束，必然是已提交或已回滚。

下面给大家留一个问题：二阶消息的操作 3 能否省略，能否只根据步骤 2 的插入是否成功，来判断是否已回滚？欢迎大家留言讨论

普通消息

二阶消息不仅可以替换本地消息表方案，也能够替换普通消息方案。如果直接调用 Submit ，那么就与普通消息方案近似，但是提供了更灵活简单的接口。

假设一个这样的应用场景，界面上有一个参加活动的按钮，如果参加活动，会赠与两本电子书的永久权限。这种情况下，可以再这个按钮的服务端中，类似这样处理：

msg := dtmcli.NewMsg(DtmServer, gid).
	Add(busi.Busi+"/AuthBook", &Req{UID: 1, BookID: 5}).
	Add(busi.Busi+"/AuthBook", &Req{UID: 1, BookID: 6})
err := msg.Submit()

这种方式也提供了异步接口，而不用依赖消息消息队列。在微服务的许多场景中，可以替换原有的异步消息架构。

小结

本文提出的二阶消息，接口简洁优雅，带来了比本地消息表和 Rocket 事务消息更简单的架构，可以帮助大家更好的解决无需回滚的数据一致性问题。

项目地址

关于分布式事务更多的理论知识与实践，可以访问以下项目和公众号：

https://github.com/dtm-labs/dtm ，欢迎访问，并 star 支持我们。

关注 [分布式事务] 公众号，获取更多分布式事务相关知识，同时可以加入我们的社群