seadt：金融级分布式事务解决方案（二）—— TCC应用与实现
点击关注公众号👆，探索更多Shopee技术实践
seadt 是 Shopee Financial Products 团队使用 Golang，针对真实的业务场景提供的分布式事务解决方案。
上一篇文章介绍了 seadt 的设计，本章介绍业务在对接中的实际应用，以及对应的代码实现。
seadt 在实现的过程中遇到了不少问题，这些问题大部分是 Golang 技术栈不够丰富引起的。我们提供了一些解决思路，希望能够帮助各位 Golang 开发者解决类似问题。
目录
1. 业务对接
1.1 应用场景
1.2 发起者对接
1.3 参与者对接
2. seadt 实现
2.1 TM 代码
2.2 RM 代码
2.3 TC 代码
2.4 异常处理
1. 业务对接
我们先温习上篇文章中，关于 seadt 的 TCC 模式整体流程：
seadt 的设计目标有两个：
事务本地化：分布式事务发起者像操作本地事务一样发起分布式事务，把外部调用、子事务等当成本地事务中的 DB 操作一样； 服务原子化：分布式事务参与者（业务被调用方）提供简单的原子操作接口，无需考虑分布式系统中的各种异常问题（并发、幂等、空提交、空回滚、事务悬挂等）。
那么什么场景适合使用 seadt 的 TCC 模式呢？
1.1 应用场景
适用场景分类：
类型一：调用多个下游事务接口
上游调多个下游的事务操作，将本地事务与多个下游的事务做成一个类似于本地事务的效果。
敲黑板：下游的事务操作不仅仅是其他服务的写接口，中间件的写也可以认为是一种事务操作。 举例：系统升级/模块拆分过程中的双写方案，转账场景等。
类型二：调用单个下游事务接口
上游调一个下游的事务操作，将本地事务与下游绑定成一个事务。
敲黑板：很多时候大家直觉会认为这不是一个分布式事务，不适合用 TCC。适不适合用 TCC 另说，不过它的确是一个分布式事务，只不过大部分团队很轻松地通过轮询等方式实现了数据一致性。 举例：调下游冻结额度。
我们上篇文章例子中的贷款流程，它属于类型一，是标准的分布式事务场景，可以使用 TCC 模式。
接下来看业务程序作为分布式事务发起者，如何启动分布式事务。
1.2 发起者对接
📍 seadt 发起者设计目标：事务本地化
分布式事务发起者像操作本地事务一样发起分布式事务，把外部调用、子事务等当成本地事务中的 DB 操作一样。
作为业务程序服务 biz-server，此时需要在一个接口中调用下游服务的 ②冻结额度、③冻结优惠券、①④本地业务逻辑处理。
这是一个非常常见的业务处理场景，业务开发人员肯定都希望这四步操作能像本地事务一样处理简单。如果业务自行处理异常场景，会非常麻烦，尤其是有两个外部调用的时候。
例如上图所示的业务流程中，连续调用两个外部的写接口，接入 seadt 后只需要这样写代码即可。事务发起者启动代码：
func *** DoSubmit(***) {
// 开启分布式事务
tcc.WithGlobalTransaction(ctx, func(ctx context.Context) {
// 调用业务参与者A的Try接口
***.TryFrozenQuotaForLoan(***)
// 调用业务参与者B的Try接口
***.TryFrozenVoucherForLoan(***)
// 自身事务操作操作
***.Insert(ctx, record)
}, &tcc_model.GlobalTransactionExtraInfo{
// 事务超时时间
TimeOutSecond: int64(constant.DEFAULT_ASYNC_PROCESS_TIMEOUT),
// 事务标识，可选
TransactionName: "quota_frozen",
})
}
是不是简单得就如同操作本地事务一样？接下来再看看参与者的对接。
1.3 参与者对接
📍 seadt 参与者设计目标：服务原子化
分布式事务参与者（业务被调用方）提供简单的原子操作接口，无需考虑分布式系统中的各种异常问题（并发、幂等、空提交、空回滚、事务悬挂等）。
在 TCC 模式下，所有参与者将资源控制接口（写接口）以 TCC 的模式提供出去，就需要每一个资源控制都实现 TCC 三个接口，分别是 Try、Confirm、Cancel 接口。
业务参与者在 TCC 模式下，除了自身的业务逻辑处理外，其余的都是公共处理逻辑，例如：Try 过程中注册分支事务，RM 本地保存分支事务，控制分支事务状态流转，分支事务状态上报，异常的统一处理等。
因此我们选择 AOP（Aspect Oriented Programming，面向切面编程）切面的方式完成这些公共处理，让业务无感知，只关注业务核心逻辑。
为了便于 AOP 的实现，让业务对接代码更加精简，我们将业务对接的接口设计如下：
其中，seadt-SDK 为参与者提供一套 TCC 的三个接口，但是只替参与者提供二阶段的 API 接口 pb，并不提供一阶段的 Try 接口；Try 接口需要业务参与者自行提供 pb。
为什么这样设计，原本有三种可行方案：
方案一：全部由业务参与者的 pb 中提供 TCC 接口，即 biz-API 提供 Try、Confirm、Cancel 接口。 方案二：全部由 seadt-SDK 的 pb 提供 TCC 的 API 接口，即 seadt-SDK-API 提供 Try、Confirm、Cancel 接口。 方案三：当前设计，业务参与者提供 Try 接口，seadt-SDK-API 提供 Confirm、Cancel 接口。
接下来对比三种方案的差异。
方案一：全部由业务参与者的 pb 中提供 TCC 接口
事务发起者需要依赖参与者 pb 并且调其接口，这是必然的。但是 TC 需要调用参与者的二阶段 Confirm 和 Cancel 接口，驱动二阶段执行完成，保证事务最终一致性。因此需要 TC 调用参与者提供的二阶段接口，这就意味着 TC 需要依赖所有参与者的 pb 文件来发起 gRPC 调用。
如果业务参与者实现并将 pb 提供给 TC 调用，会有如下问题：
TC 依赖业务 pb，导致 TC 维护成本高； 扩展性差。如果框架升级，需要在 Confirm 和 Cancel 请求里增加一些分布式事务的字段，那么所有接入的业务都需要做代码变更。
TC 作为一个分布式事务的协调者，只负责推动分布式事务状态流转，完成事务，不应该和业务有耦合。所以这种方式不合理，不利于 seadt 未来的发展。
除此之外，我们有另一种 TC 不依赖参与者 pb 的方式，即 TC 通过 grpc-json 的方式调用这些业务接口，做到 TC 和业务的解耦。
实现代码如下：
jsonClient := grpc.NewClient(func(options *client.Options) {
options.PoolSize = serviceConfig.ClientPoolSize
options.Selector = selector.NewSelector(
cacheRegistry.TTL(time.Duration(serviceConfig.RegistryCacheTTL) * time.Minute),
)
}, grpc.Codec("application/json", emptyCodec{}))
// 通过json的方式，调用下游grpc接口
func JsonCall(***) (***) {
err := jsonClient.Call(ctx, jsonClient.NewRequest(service, method, req, func(o *client.RequestOptions) {
o.ContentType = "application/json"
}), &resp)
return resp, err
}
通过 JSON 调用 gRPC，不只需要改造调用方，还需要改造被调用方，因为要保持上下游使用的 JSON 序列器一致，才能正确解析请求体 request。
type jsonCodec struct {
}
func (j jsonCodec) Marshal(v interface{}) ([]byte, error) {
return json.Marshal(v)
}
func (j jsonCodec) Unmarshal(data []byte, v interface{}) error {
decoder := jsoniter.NewDecoder(strings.NewReader(string(data)))
decoder.UseNumber()
err := decoder.Decode(v)
return err
}
这种方式虽然能解决 TC 依赖业务 pb 的问题，但依然存在如下缺点：
业务方升级成本高，不易统一管理； 业务 pb 更新风险高。弱约束的情况下，如果业务变更了 pb 接口的 service name，可能会导致存量的未完成的分布式事务永远无法完成（这里具体原因和 grpc-json 这种动态 RPC 调用有关）。
因此 TC 通过 grpc-json 的方式调用业务二阶段接口，也不太可取。接下来看方案二。
方案二：全部由 seadt-SDK 的 pb 提供 TCC 的 API 接口
由 TCC 框架定义通用接口会存在如下问题：
效率低。通用接口如果做成 JSON 入参和出参（否则无法适应不同的业务场景），需要业务发起者和参与者手动进行 JSON 序列化和反序列化，比原生的业务 pb 接口多了一次序列化和反序列化； 业务逻辑可读性下降。假设 RM 需要对外提供多套 TCC 接口时，业务请求入口都是框架提供的标准 pb 接口，无法直观地通过接口本身区分业务场景。
所以我们也不选择方案二：全部由 seadt-SDK 的 pb 提供 TCC 的 API 接口。接下来看我们当前的实现方式。
方案三：业务参与者提供 Try 接口，seadt-SDK 提供二阶段接口
相比于方案二，业务参与者提供 Try 的业务 pb 接口：
发起者直接依赖调用方 pb 接口，调用方法语义更明确； 效率高，所有调用均用 Go 原生 pb 协议； 易维护，seadt 无需依赖业务 pb； 理解成本高，Try 接口需要特殊处理，额外封装 pb 接口，而 Confirm/Cancel 却不用。
综上所述，采用方案三：业务参与者提供 Try 接口，seadt-SDK 提供二阶段接口。
Try 接口，通过业务 pb 接口提供； Comfirm/Cancel 接口，通过框架 pb 提供，业务层无需感知。
该方式实现的接口调用时序如下：
1.3.1 参与者实现 TCC 接口
基于上述的设计，参与者对接首先实现 seadt-RM 提供的三个 API 接口。
特别说明：这里的 TCC 接口并不是提供给外部的 API 接口，而是一套 Proxy 的接口。
type ITccResourceService interface {
Try(ctx context.Context, payload interface{}) (bool, error)
Confirm(ctx context.Context, payload interface{}) bool
Cancel(ctx context.Context, payload interface{}) bool
…
}
在 Try/Confirm/Cancel 接口中的 payload，类似于 gRPC 接口中的 request，用于携带 TCC 处理的业务请求参数。其中 TM 在调用 Try 时指定 payload，seadt 在二阶段回调时自动将 Try 的 payload 注入给 Confirm 和 Cancel。
拿本文中的例子，参与者对接代码的具体业务实现：
// 业务实现Try接口
func (t *QuotaFrozenTccImpl) Try(ctx context.Context, payload interface{}) (bool, error) {
if req, ok := payload.(*api.ClFrozenQuotaReq); ok {
// 执行业务逻辑，对额度进行冻结
***.QuotaAvailableToFrozen(ctx,req)
return true, nil
} else {
return false, nil
}
}
// 业务实现Confirm接口
func (t *QuotaFrozenTccImpl) Confirm(ctx context.Context, payload interface{}) bool {
// 示例场景中，Confirm阶段不需要操作额度，因此这里Confirm空转
log.Info(*** "call confirm"})
return true
}
// 业务实现Cancel接口
func (t *QuotaFrozenTccImpl) Cancel(ctx context.Context, payload interface{}) bool {
if req, ok := payload.(*api.ClFrozenQuotaReq); ok {
// 执行业务逻辑，将冻结的额度进行解冻
***.QuotaFrozenToAvailable(ctx,req)
return true
} else {
return false
}
}
1.3.2 封装 Try 接口并提供 pb
// 参与者直接 pb 提供的api接口，给到上游调用
rpc TryClFrozenQuota (***) returns (***) {
};
// pb 提供的api接口实现
func (t ***) TryClFrozenQuota(ctx context.Context,*) error {
…
// 调用TCC代理的Try方法，由SDK提供接口，业务实现
res, err := seadt.GetQuotaFrozenProxy().Try(ctx, req)
// err处理
…
}
参与者对接 seadt，也同样简单。业务只需要处理业务逻辑即可，不再需要关注分布式事务的空回滚、空提交、事务幂等、事务悬挂等问题，由 seadt 统一处理。
至于 seadt 是如何处理这些问题的，会在本文的“2.4 异常处理”中统一介绍。
2. seadt 实现
业务的实现已经简单到接近本地事务处理了，背后则是 seadt-SDK 替业务负重前行。先看下 seadt TM、RM、TC 提供的接口以及对应调用关系。
业务系统发起一个分布式事务处理流程：
事务发起者通过 TM 的启动分布式事务（StartGTX），内部由 TM 向 TC 创建全局分布式事务； 事务发起者在闭包方法块中执行处理逻辑，包括向事务参与者调用 Try 一阶段方法； 事务参与者集成 seadt-RM 后，在 Try 方法的切面中，由 RM 向 seadt-TC 创建分支事务（CreateBTX），成功后再执行本地真正的业务逻辑。执行完成后向 seadt-TC 上报分支事务状态（ReportBTXstatus），该接口异步上报，允许超时/失败； 事务发起者结束分布式事务 GTX，由 seadt-TM 向 TC 上报 GTX 状态（ReportGTXstatus），允许超时/失败等。如果发生异常，TC 未收到 GTX 结果，则在设定时间后由 seadt-TC 的 Recovery 对发起者的 TM 进行反查（QueryGTXresult）。 seadt-TC 拿到 GTX 的结果后，向 RM 广播二阶段结果（Confirm/Cancel）。
这整套流程中，比较棘手的问题有下面几点：
TM 的 tcc.WithGlobalTransaction()如何与本地事务融为一体，并做到一致性；TM 如何避免事务结束对 TC 依赖，不感知 seadt 的存在； 如何让参与者对接 RM 简单； RM 如何提事务参与者解决空提交、空回滚、事务悬挂问题； TC 调用 TM 和 RM 的接口，如何摆脱依赖业务 pb 的问题； TC 的空提交、空回滚、事务悬挂问题； TC 的高可用，低延时（本文暂时不涉及）。
带着上面几个问题，我们接下来看各个组件是如何解决和实现的。
2.1 TM 代码
TM 实现的过程中，需要解决下面两个难题：
TM 的 tcc.WithGlobalTransaction()如何与本地事务融为一体，并做到一致性；TM 如何避免事务结束对 TC 依赖，事务发起者不感知 seadt 的存在。
2.1.1 TM 事务启动
TM 的分布式事务启动如何与本地事务融为一体，我们从“1.2 发起者对接”中可以看到，发起者只需要如同本地启动事务即可。接下来看看分布式事务启动 tcc.WithGlobalTransaction() 伪代码。
func WithGlobalTransaction(ctx ***, process func(***) (****) {
****
// 基于本地事务模型做扩展，支持TCC分布式事务。
transaction.WithTransaction(ctx, func(ctx context.Context) {
rootContext := tm.RefTransactionManager().StartGlobalTransaction(ctx, extraInfo)
****
process(rootContext)
})
}
StartGlobalTransaction 为开启全局事务的核心方法，下图为开启全局事务注册的触发器，以及触发器内部的处理。流程如下：
StartGlobalTransaction 伪代码如下：
func (t ***) StartGlobalTransaction(***) *sd_context.RootContext {
****
createGbTxResp := t.createGlobalTransaction(rootContext, extraInfo)
// 开新协程处理，在新协程中保存全局事务
***.CallWithNewGoroutineSync(ctx, func(ctx context.Context) {
transaction.WithTransaction(ctx, func(ctx context.Context) {
repo.SaveGlobalTransaction(ctx, gbTransPo)
})
})
// 全局事务注册事务触发器
t.registerGlobalTransactionCallback(rootContext, gbTransPo)
return rootContext
}
上述代码中，为什么在 SaveGlobalTransaction 和 TransferGlobalTransactionToCancel 需要开启新协程，并在新协程中开始事务执行呢？
因为当前的事务模板默认只支持 PROPAGATION_REQUIRED 的事务传播行为，这里要求创建全局事务的本地事务和将全局事务状态设置为回滚状态的本地事务提交独立于外层事务，即 PROPAGATION_REQUIRED_NEW，通过新开协程方式达到 PROPAGATION_REQUIRED_NEW 的目的。
为什么不将 StartGlobalTransaction 的处理放入 transaction.WithTransaction 中？
由于 transaction.WithTransaction 是原有的普通事务模板，而 StartGlobalTransaction 需要做的处理只针对开启全局事务才需要。
因此未在 transaction.WithTransaction 原有的基础上改，而是对其进行扩展，并且提供一个全新的事务模板方法 WithGlobalTransaction() 给需要全局事务的发起者使用，并且可以完全兼容系统中原有的普通事务。
2.1.2 TM 事务模板&触发器
接下来看下如何利用本地事务模板：transaction.WithTransaction()。
func WithTransaction(ctx context.Context, process func(ctx context.Context)) {
****
BeforeTransaction(ctx)
defer func() {
// 避免异常，放在此处保证执行
AfterTransaction(ctx, err)
}()
process(ctx)
}
在事务执行 process() 前后，分别加上前置后置处理切面。
事务执行前：BeforeTransaction 主要为开启事务前置处理，包括获取 Session、初始化事务上下文，和初始化各类核心触发器。
这些 Callback 正是 TCC 全局事务需要用到的核心触发器，由 Session 管理。这些 Calback 的注册和回调触发接口如下：
// 事务模板Callback接口
type ITransactionCallback interface{
// 注册接口
RegisterBeforeCommitCallback(ctx context.Context, callback BeforeCommitCallback)
RegisterBeforeRollbackCallback(ctx context.Context, callback BeforeRollbackCallback)
RegisterBeforeCompletionCallback(ctx context.Context, callback BeforeCompletionCallback)
RegisterCommitCallback(ctx context.Context, callback AfterCommitCallback)
RegisterRollbackCallback(ctx context.Context, callback AfterRollbackCallback)
RegisterCompletionCallback(ctx context.Context, callback AfterCompletionCallback)
// 触发执行接口
BeforeCommit(ctx context.Context) error
BeforeRollback(ctx context.Context) error
BeforeCompletion(ctx context.Context)
AfterCommit(ctx context.Context)
AfterRollback(ctx context.Context)
AfterCompletion(ctx context.Context)
}
事务执行后：AfterTransaction 在业务执行完 DB 操作后，最终需要提交事务，在处理事务前后我们需要执行各种触发器。
我们先看看事务模板和核心触发器关系图：
其中数据库事务 commit/rollback 前触发器有：BeforeCommit、BeforeRollback、BeforeCompletion 三个触发器。
而数据库事务 commit/rollback 后触发器有：AfterCommit、AfterRollback、AfterCompletion 三个触发器。
业务 DB 变更完成，调用真正的事务执行的 AfterTransaction，内部处理逻辑伪代码如下：
func AfterTransaction(ctx context.Context, originErr error) {
****
// 嵌套事务处理，在最外层事务处理结束后清理事务相关信息
if nestTransactionDepth <= 0 {
***
}
****
if isCommitted {
***.ExecuteBeforeCommitCallbacks(ctx)
} else {
***.ExecuteBeforeRollbackCallbacks(ctx)
}
// 即将进入本地事务真正提交到db，BeforeCompletion
***.ExecuteBeforeCompletionCallbacks(ctx)
if isCommitted {
// 执行真正的db事务提交
tx.CommitTrans()
// 事务commit完成，处理AfterCommit
***.ExecuteCommitCallbacks(ctx)
} else {
// 执行真正的db事务回滚
tx.RollbackTrans()
// 事务rollback完成，处理AfterRollback
***.ExecuteRollbackCallbacks(ctx)
}
***.ExecuteAfterCompletionCallbacks(ctx)
}
为了事务模板触发器设计的完整性，我们设计了 AfterCommit 和 AfterRollback 触发器，实际中我们并没有用到。接下来我们看下 TCC 在执行分布式事务中是如何结合这些触发器保证事务一致性的。
启动分布式事务的时候，首先会注册一堆事务触发器，我们只举例 BeforeCommit 触发器的注册部分代码片段：
// 注册事务提交前回调BeforeCommit
func (local ***) RegisterBeforeCommitCallback(***) {
callbacks := local.GetBeforeCommitCallbacks(ctx)
// 将当前注册的callback添加进callback列表
callbacks = append(callbacks, callback)
// 将callback列表存入当前事务对应的routineLocal中(类似于threadLocal)
local.beforeCommitCallbacks.Put(callbacks)
}
注册完触发器后，就执行事务模板中的业务回调处理，最后在事务提交的时候才触发这些触发器。我们只举例 commit 正常场景，BeforeCommit 触发器和 AfterCompletion 触发器的执行顺序和核心伪代码。
接下来是 BeforeCommit、BeforeCompletion、AfterCompletion 触发器内部的回调处理。
BeforeCommitCallback(***){
// 检查全局事务是否超时
if !time.Now().Before(globalTransPo.ExpireTime) {
****
// 设置全局事务超时回滚标志
rootContext.Set(constant.LocalKey_GlobalTimeout, true)
// 触发事务超时回滚
panic(exception.ExGlobalTransactionTimeoutError)
}
// 设置全局事务为commit (不需要开启子事务，跟随当前事务即可)
repo.***.TransferGlobalTransactionToConfirm(ctx, globalTransPo)
})
BeforeCompletionCallback(rootContext, func(ctx context.Context) {
***.ClearRoutineLocalInfo()
})
AfterCompletionCallback(rootContext, func(ctx context.Context) {
****
// 上报Global Transaction状态到TC
tc.***.ReportGlobalTransactionStatus(ctx, ***)
// 出现err只记录日志，不影响原事务结果，由TC驱动反查。
****
}
})
那这些 Callback 是如何触发回调执行的呢？答案就是在本地事务业务逻辑 precess() 处理完成之后，进入 AfterTransaction 部分，在这一部分进行事务的真正提交到 DB 和触发各种对应 Callback 的回调执行。
另：在上文中，我们大量使用了 RoutineLocal 来进行事务上下文的缓存，事务模板核心触发器信息缓存。RoutineLocal 是我们实现的类似于 ThreadLocal 的，与当前协程绑定的内存缓存，其实现逻辑大体如下：
// tls的结构为 Sync.Map<goroutineId, Sync.Map<routineLocalName, interface{}>>
func init() {
gp := g.G()
if gp == nil {
return
}
tlsDataMap.Store(gp, &tlsData{
data: dataMap{},
})
}
2.2 RM 代码
参与者接入的 RM 需要达到下面两个要求：
让参与者对接 RM 简单; RM 替事务参与者解决空提交、空回滚、事务悬挂问题。
上文的“2.1 参与者对接”中，我们已经详细了讲述了接口的设计思路，让参与者对接 RM 足够简单。背后其实主要是通过代理模式将公共的处理统一交由 RM 处理，这里面用到的核心点就是 AOP 技术。
我们先来看下参与者中 seadt-SDK 提供给 TC 调用的二阶段接口：
// seadt-SDK的二阶段grpc接口，由TC自动回调处理，无需业务关系
type SeadtResourceManagerHandler interface {
// 分支事务二阶段Confirm
SDConfirm(context.Context, *SDPhaseTwoReq, *SDPhaseTwoResp) error
// 分支事务二阶段Cancel
SDCancel(context.Context, *SDPhaseTwoReq, *SDPhaseTwoResp) error
}
其中，一阶段的 Try 由业务自行提供，该接口无需业务关注和实现。业务真正需要关注和实现的是 RM 定义的一套标准 TCC 接口：
// TCC协议的接口，业务需要实现 tcc 两个阶段三个接口。seadt-SDK在此接口基础上实现代理Proxy
type ITccResourceService interface {
Try(ctx context.Context, payload interface{}) (bool, error)
Confirm(ctx context.Context, payload interface{}) bool
Cancel(ctx context.Context, payload interface{}) bool
}
三个接口中的 payload 为 TCC 协议的业务参数，可类比于 gRPC 接口的 request。
为了让业务接入 RM 尽量简单，业务每个 TCC 场景的 TccResource 只需要实现 ITccResourceService 接口并将其注册到 seadt 即可。在实现完整的 TCC 协议的背后，是 seadt 在 SDK 内部基于业务实现的 ITccResourceService，封装成新的 Proxy 对象，并在 Proxy 对象中进行 AOP 处理和反射处理。
参与者对接 RM 生成 Proxy 对象过程如下：
其中由 RM 生成的关键对象 Proxy 中包含 TCC 三个接口方法的 MethodDescripor，数据结构如下：
// RegisterResource返回 保存TccResourceServiceProxy的反射调用信息
type ResourceServiceDescriptor struct {
Name string
ReflectType reflect.Type
ReflectValue reflect.Value
Methods sync.Map // string -> *MethodDescriptor
}
// 保存Method(ITccResourceService的Try/Confirm/Cancel)的反射调用描述信息
type MethodDescriptor struct {
Method reflect.Method
CallerValue reflect.Value
CtxType reflect.Type
ArgsType []reflect.Type
ArgsNum int
ReturnValuesType []reflect.Type
ReturnValuesNum int
}
生成 Proxy 的伪代码如下：
func RegisterTccResourceService(***) *** {
****
// ITccResourceService的Try的切面逻辑封装
makeCallProxy := func(***) func(in []reflect.Value) []reflect.Value {
*****
// 处理切面的逻辑
returnValues, _ := proceed(methodDesc, branchActionContext, resource)
return returnValues
}
}
// 遍历字段并反射注入proxy中。
for i := 0; i < numField; i++ {
***
commitMethodDesc := proxy.Register(proxyService, ConfirmMethod)
cancelMethodDesc := proxy.Register(proxyService, CancelMethod)
tryMethodDesc := proxy.Register(proxyService, methodName)
// 反射注入
f.Set(reflect.MakeFunc(f.Type(), makeCallProxy(tryMethodDesc, tccResource)))
}
return tccResourceServiceProxy
}
我们重新看下 seadt 的交互时序，重点关注业务参与者与 RM 的工作分工：
RM 统一替业务参与者创建并注册了分支事务、处理事务状态等，而这些处理都是放在上面讲到的 RM 生成的 Proxy 中处理的。下面是 Proxy 的处理逻辑：
// RM一阶段处理：AOP切面处理逻辑
func proceed(***) (***) {
// 前置处理
****
// 向TC注册分支事务，并记录到本地
branchTransaction, err := registerAndInitBranchAction(branchActionContext, resource)
transaction.WithTransaction(branchActionContext, func(ctx context.Context) {
// 执行真正业务try方法
returnValues = proxy.Invoke(methodDesc, branchActionContext, args)
// 记录分支事务为try完成（不需要开新协程，保证跟业务事务一起成功或者失败） repo.***.TransferBranchTransactionToTried(ctx, branchTransaction)
})
// 向TC上报分支事务状态（这里上报出错无影响）
rm.***.ReportBranchTransactionStatus(****)
****
}
到这里，RM 的 TCC 协议的 ITccResourceService 的实现注册，seadt-SDK 内部的 TccResource 的注册管理，Method 的反射调用的封装管理，Try 的 AOP 切面逻辑的实现和内部的业务逻辑基本已经介绍完毕。
RM 在二阶段方法中最核心的处理就两点：
统一处理空提交、空回滚、事务悬挂； 回调业务的二阶段业务处理。
BranchCommit 和 BranchRollback 的实现原理类似，下面以 BranchCommit 为例：
func (t *TCCResourceManager) BranchCommit(***) (***) {
****
// 二阶段前置检查，包括空提交，空回滚，事务悬挂，幂等等处理
checkResult := rm.***.CheckResourcePhaseTwoTransfer(ctx, &rm.ResourceCheckReq{})
// Phase2CommonInterceptor为业务自定义的二阶段前置处理拦截器，会在二阶段真正处理前调用
tcc_interceptor.Phase2CommonInterceptor(func(ctx context.Context, payload interface{}) {
***
returnValues = proxy.Invoke(tccResource.ConfirmMethodDescriptor, businessActionContext, args)
})(businessActionContext.Context, businessActionContext.Payload)
****
}
空提交、空回滚、事务悬挂，到底是什么？它们是如何产生的？又该如何处理？这些问题不仅在业务参与者侧会发生，在 TC 侧同样会发生，因此我们在本文“2.4 异常处理”统一介绍。
2.3 TC 代码
我们看下 TC 在整个分布式事务中的位置和作用。如下图：
TC 相对于 TM 和 RM，其实实现相对简单。从上图可以看出，实际上主要是提供了 4 个对外接口，分别是：创建全局事务 CreateGTX、上报全局事务状态 ReportGTXstatus、创建分支事务 CreateBTX、上报分支事务状态 ReportBTXstatus；以及二阶段处理器：反查 TM 状态处理和推进 RM 二阶段处理。
这 4 个接口实现比较简单，都是单事务的原子 DB 操作，不涉及外部调用，很容易通过普通事务实现。TC 的难点有下面三个：
TC 调用 TM 和 RM 的接口，如何摆脱依赖业务 pb 的问题; TC 的空提交、空回滚、事务悬挂问题; TC 的高可用，低延时（本文暂时不涉及）。
上文介绍业务参与者对接 RM 的时候，已经详细说明了由业务方提供一阶段的 Try 方法，由 seadt-SDK 提供二阶段的 Confirm/Cancel 方法。因此 TC 完全不依赖参与者 pb，需要做好业务参与者一阶段 Try 方法和二阶段 Confirm/Cancel 的关联关系，调用到 RM，由 RM 找到业务具体的二阶段处理逻辑。
先看下 TC 创建全局事务接口处理逻辑：
// 创建全局事务
func (i *TCCTransactionImpl) CreateGTX(***) string {
db.WithTransaction(ctx, func(ctx context.Context) {
// 创建全局事务记录
transRecord := ***.InitGlobalRecord(lesseeId, timeOut, globalId, txName)
// 创建回调信息记录，address是TM的地址，用于反查TM
***.InitGlobalInvokeInfo(transRecord.Txid, address)
// 事务记录保存
repo.***.save(transRecord)
})
***
}
在创建分支事务的时候，TC 需要建立好参与者 Try 与 Confirm/Cancel 接口的映射关系：
// 创建分支事务
func (i *TCCTransactionImpl) CreateBTX(***) string {
db.WithTransaction(ctx, func(ctx context.Context) {
***
// 避免空回滚和事务悬挂
if globalRecord.TxStatus != ***Status_Initial {
***
}
// 创建分支事务记录
***.InitBranchRecord(txid, resourceKey)
// 创建回调信息记录，address是RM的地址，用于二阶段调用
***.InitBranchInvokeInfo(txid, BranchTxid, address, resourceKey)
repo.***.save(transRecord)
})
return branchId
}
最后是事务恢复处理器 Recovery 的补偿处理，它有两个触发点：
在 TM 上报的时候，二阶段状态是立刻触发； 在 1 失败后，定期触发重试。
func twoPhaseHandle(ctx context.Context, globalRecord *model.GlobalTransaction, globalState int32) {
***
execRecords := repo.QueryByFilter(ctx, filter)
for _, execRecord := range execRecords {
// 遍历处理未完成的分支事务
branchHandle(ctx, execRecord, globalState)
}
if finish {
// 分布式事务已经完成
globalFinishHandle(ctx, globalRecord, globalState)
} else {
// 对悬挂分支事务进行上报
utils.HoldingReport(ctx, globalRecord)
}
}
从上面代码片段中可以看到，TC 也做了事务悬挂判断和处理，为什么 TC 会有事务悬挂呢？它和 RM 的事务悬挂有何区别？接下来看下 seadt 实现方式中的异常问题及异常处理。
2.4 异常处理
细心的读者会发现，seadt 的设计与其他几个常见的分布式解决方案相比，在 RPC 调用方面会比较多。如下图所示：
两个参与者的 TCC 正向调用，其中同步 RPC 调用会有 8 次之多，其中除了⑥上报全局事务结果，⑦⑧二阶段执行 Confirm/Cancel 外，其余 5 个均为不可重试 RPC 同步调用。
每一个 RPC 调用，都会面临着成功、失败、超时、网络拥塞、不可达等异常情况，而这些情况会互相叠加而导致异常场景非常多。
为什么要设计如此多的 RPC 调用？我们的设计理念就是让业务用得舒服，让用户把分布式事务当成本地事务处理，同时也让参与者无需考虑诸多异常场景，更多关注业务逻辑。因此才会将空提交、空回滚、事务悬挂等异常通通包揽到 seadt 处理。而为了处理这些异常，不得不在 TM 和 RM 侧保留好事务数据，以及维护这些数据的状态。
当然，除了 RPC 调用的异常外，还会有很多并发、幂等、逻辑控制等问题需要处理。而 RPC 的异常最为典型，因此我们也着重介绍 RPC 引起的异常。
2.4.1 空提交
什么是空提交？
空提交就是业务 RM 实际没有执行一阶段 Try 成功，但收到来自 TC 的 Confirm 调用。
空提交是如何发生的？
只要业务逻辑正确，理论上不会出现空提交。
空提交如何处理？
设计规范中是禁止空提交出现，如果出现空提交，程序会报错并将异常上报监控告警，进行人工问题排查和处理。
2.4.2 空回滚
什么是空回滚？
同空提交类似，如果业务参与者 RM 没有执行一阶段 Try 成功，但是收到来自 TC 的 Cancel 调用。
空回滚是如何发生的？
空回滚场景 1：参与者向 TC 注册分支事务成功后（包括超时但是成功），RM 后续未执行。
空回滚场景 2：参与者向 TC 注册分支事务成功后，RM 执行业务回调失败。
空回滚场景 3：
如图所示，当发起者向 TC 注册分布式事务后，由于网络原因等导致 Try 无法正常调到参与者，而整个事务失败，但是 TC 由于判断事务最终状态为回滚后，需要向参与者调用 Cancel，于是就产生了参与者没有发生 Try 的场景下需要进行 Cancel 操作的情况。
我们这里是另一种情况，参与者收到 Try，在向 TC 注册中超时，或者注册成功后异常，会导致只有注册分支事务，并没有真正地执行 Try。
空回滚如何处理？
空回滚的处理是不是如同空提交一样，直接报错呢？针对空回滚，并不能这样做，要允许空回滚。RM 在接收 TC 的 Cancel 后，向 TC 返回处理成功，让 TC 结束整个事务。
为什么空提交和空回滚的处理策略截然不同呢？
TCC 是一个标准的 2pc，满足 ACP（Atomic Commitment Protocol）协议规范，即只有所有的参与者都 Try 成功，才允许 TC 决策为 Confirm，否则应该为 Cancel。既然如此，就不允许参与者 Try 都未执行成功，而 TC 决策为 Confirm，并向参与者广播，所以禁止空提交。但是回滚不同，ACP 协议允许某个参与者未执行 Try，而 TC 决策为 Cancel 并向所有参与者广播，所以允许空回滚。
为什么 TC 不在接收到参与者 Try 成功后才调用参与者的 rollback 呢？
由于各种网络原因，RM 本地执行 Try 成功后， ReportBranchStatus的请求可能无法到达 TC，因此 TC 对 RM 是否成执行 Try 的感知不是完全准确的，所以只能广播 Rollback 给所有已经创建 Branch 事务的 RM。
2.4.3 事务悬挂
什么是事务悬挂？
如果全局事务已结束，但是参与者的 Try 在之后执行成功，并占用资源。导致资源无法释放。
事务悬挂如何产生？
由于网络等异常，导致参与者注册分支事务时间在 TC 执行完成全局事务之后，导致 TC 无法再处理分支事务，而导致参与者 Try 之后永远无法执行二阶段逻辑，整个事务就被挂起了。
事务悬挂场景 1：
事务悬挂场景 2：
事务悬挂如何处理？
事务悬挂场景 1：TC 会拒绝已经结束的全局事务注册分支事务，让参与者的 Try 失败； 事务悬挂场景 2：RM 会在空回滚的时候插入一条分支事务 Cancel 完成的记录，通过唯一约束阻止参与者 Try 的时候重复创建并且执行后续逻辑处理。
我们拿 RM 的核心代码举例。在“2.2 RM 代码”的核心代码中有提到校验二阶段状态合法性，这里主要是 RM 处理空提交、空回滚、事务悬挂的处理：
func (***) CheckResourcePhaseTwoTransfer(ctx…) (***) {
transaction.WithTransaction(ctx, func(ctx context.Context) {
//******* 分支事务不存在场景处理 *******//
if branchTransaction == nil {
if req.TxStatus == ***Status_Committed {
// 禁止空提交场景: 未触发Try，不支持confirm
***
} else {
// 空回滚处理场景：空回滚需要插入分支事务记录到db，防止TC的分支注册成功返回后到，进而执行Try导致分支事务悬挂
repo.***().SaveBranchTransaction(ctx, protectedBranchTrans)
}
}
//******* 分支事务存在场景处理 *******//
// 检查状态流转是否合理
if branchTransaction.TxStatus < ***Status_Tried {
// 禁止空提交场景: Try未执行成功，不支持Confirm
if req.TxStatus == ***Status_Committed) {
***
}
// 空回滚处理场景：Cancel 允许状态正常流转，框架将分支事务流转到700，但不触发业务的Cancel接口
if req.TxStatus == ***Status_Canceled {
repo.***.TransferBranchTransactionToCancel(*)
***
}
}
// 分支事务状态流转规范约束处理，以及幂等处理
***
})
}
2.4.4 并发异常
异常一：如果③注册分支事务由于网络因素导致重复注册，TC 会创建两个分支事务记录，并不会幂等处理（有可能参与者真的会多次参与某一个分布式事务）； 异常二：TC 像事务参与者广播二阶段的时候，由于网络等其他因素，可能存在并发调用参与者的 Confirm/Cancel 接口； 异常三：在事务悬挂场景 2 中，RM 向 TC 注册分支事务并本地创建分支事务成功，但同时 TC 向 RM 发起 Cancel 请求，这个时候会导致 RM 的空回滚和 RM 一阶段 Try 的业务回调同时执行。而此事 RM 处理空回滚的时候，由于本地分支事务存在，于是直接将本地分支事务状态置为 Cancelled，而 Try 并行执行完毕，依然存在事务悬挂问题。
对于并发异常，我们统一处理逻辑为：一锁二判三更新。
首先在 RM 侧加上分支事务记录锁，然后判断分支事务的状态，最后更新。
如上图，对于最复杂的异常三场景，无论是发生空回滚，还是空回滚与一阶段 Try 业务回调并行执行，由于都存在分支事务状态，加上分支事务锁，可以保证两者串行化，避免并发问题。
至此，就已经将 seadt 实现 TCC 过程中，遇到的问题和解决办法全盘托出了。
目前 seadt 组件已经在部分核心业务流程中得到使用，大大减少了原有业务自实现中的开发内容。接下来 seadt 会基于现有架构和设计支持 Saga 模式，让业务团队在事务处理中更加轻松自如。
后续我们将针对 seadt 的应用、新功能、新规划以及难点设计等输出文档说明，大家敬请期待。
本文作者
Marshal、Ansen、Yongchang，来自 Shopee Financial Products 团队。
技术编辑
Yang，来自 Shopee Financial Products 团队，Shopee 技术委员会 BE 通道委员。
加入我们
基于 Shopee 服务的市场，Financial Products 致力于打造信贷、保险和投资理财等金融服务。信贷为客户和商户提供消费贷和现金贷服务。在诸多消费场景，提供各种类型的保险购买服务；以及提供即时的基金购买和赎回服务，满足用户的投资理财需求。我们围绕零售金融持续打造资金、资产、核算、交易、风控、用户、承保、理赔等金融核心服务；另外，在金融方面的资金账务严要求情形下，我们的业务还呈现出场景多、金额小、交易并发高的特点，我们的团队需解决业务扩展性、数据一致性、高并发等多维度的挑战。
目前团队大量岗位持续招聘中，海量 HC 涵盖后端、前端、测试、大数据等，感兴趣的同学可将简历发送至：rachel.chen@shopee.com（亦可进行咨询，注明来自 Shopee 技术博客）。