聊聊分布式解决方案Saga模式-蒲公英云

Saga模式

Saga模式使用一系列本地事务来提供事务管理，而一个本地事务对应一个Saga参与者，在Saga流程里面每一个本地事务只操作本地数据库，然后通过消息或事件来触发下一个本地事务，如果其中一个本地事务失败了，Saga就会执行一系列补偿事务来实现回滚操作。（补偿事务简单来讲就是对之前本地事务做的修改导致不一致的情况执行反向操作来消除掉不一致的状态）。

上图左侧是正常的事务流程，当执行事务T3时出现异常，则开始反向执行右边的事务补偿，其中C3是T3的补偿，C2是T2的补偿，C1是T1的补偿，将T3，T2，T1已经修改的数据做补偿处理。

实现分析

对Saga事务流程进行排序，当Ti事务完成之后，需要决定下一步要怎么进行。如果成功执行T(i+1)分支，如果失败，则执行C(i-1)分支。这类似一个工作流或是状态机的概念。从实现来看，有两种方式：

集中式实现

集中式协调器负责服务调用以及事务协调（Orchestration）即编排实现：集中式协调器负责服务调用以及事务协调。Saga提供一个控制类，其方便参与者之间的协调工作。事务执行的命令从控制类发起，按照逻辑顺序请求Saga的参与者，从参与者那里接受到反馈以后，控制类在发起向其他参与者的调用。所有Saga的参与者都围绕这个控制类进行沟通和协调工作。

去中心化实现

分布式的实现方式——通过事件驱动的方式进行事务协调（Choreography）即协同实现：Saga参与者（子事务）之间的调用、分配、决策和排序，通过交换事件进行进行。是一种去中心化的模式，参与者之间通过消息机制进行沟通，通过监听器的方式监听其他参与者发出的消息，从而执行后续的逻辑处理。由于没有中间协调点，靠参与者自己进行相互协调。

实现比对

我个人认为在计算机的世界里没有银弹！任何的解决方案只能说是合适与不合适，而没有完美的契合并解决。

如上两种解决方式都有一定的弊端；对于集中式的实现方式，其弊端如下：

必须额外实现一个协调器，相当于增加了系统复杂度
需要考虑协调器自身发生故障时应对措施

分布式的实现方式，其弊端如下：

添加新的事务步骤时比较麻烦，需要确定哪个Saga参与者订阅了哪个事件。
有可能出现循环依赖的问题，每一个Saga参与者都可能订阅其他参与者的事件。
集成测试异常复杂，需要运行所有服务来模拟事务。

实现方式

目前看到市面上已经有很多的saga实现，他们都具备saga的基本功能。

这些实现，可以大致可以分为两类

状态机实现

Seata

这一类的典型实现有seata的saga，他引入了一个DSL语言定义的状态机，允许用户做以下操作：

在某一个子事务结束后，根据这个子事务的结果，决定下一步做什么
能够把子事务执行的结果保存到状态机，并在后续的子事务中作为输入
允许没有依赖的子事务之间并发执行。

优点：
功能强大，事务可以灵活自定义
缺点：
状态机的使用门槛非常高，需要了解相关DSL，可读性差，出问题难调试。官方例子是一个包含两个子事务的全局事务，Json格式的状态机定义大约有95行，较难入门。
接口入侵强，只能使用特定的输入输出接口参数类型，在云原生时代，对强类型的gRPC不友好（gRPC协议，在TM拿不到用户自定义的输入输出pb文件，因此无法解析结果中的字段）

Masstransit Saga State Machines

Masstransit是一个免费、开源的.NET 分布式应用框架。其功能之一就是提供了强大的状态机编排能力。通过集成消息队列中间件，基于C#高效易用的语法，支持了状态机的编排。其使用语法示例如下

/ 下单 初始化 → 已初始化
/ 翻译：当前状态是Initial且执行OrderProcessInitializationEvent事件时，Then(然后)执行xxxx，最后将状态转换(TransitionTo)为OrderProcessInitializedState
During(Initial,
    When(OrderProcessInitializationEvent)
        .Then(x => {
            x.Saga.OrderStartDate = DateTime.Now;
        })
        .TransitionTo(OrderProcessInitializedState));
/ 库存 已初始化 → 校验库存
/ 翻译：当前状态是OrderProcessInitializedState且执行CheckProductStockEvent事件时，Then(然后)执行xxxx，最后将状态转换(TransitionTo)为CheckProductStockState
During(OrderProcessInitializedState,
    When(CheckProductStockEvent)
    .Then(x => {
        System.Console.WriteLine(x.Message.OrderId);
        })
        .TransitionTo(CheckProductStockState));
/ 支付 校验库存 → 支付
During(CheckProductStockState,
    When(TakePaymentEvent)
        .TransitionTo(TakePaymentState));
/ 订单 支付 → 创建订单
During(TakePaymentState,
    When(CreateOrderEvent).Then(x => {
        System.Console.WriteLine(x.Message.OrderId);
    })
        .TransitionTo(CreateOrderState));
/ 创建订单失败
DuringAny(When(CreateOrderFaultEvent)
    .TransitionTo(CreateOrderFaultedState)
    .Then(context => context.Publish<Fault<TakePaymentEvent>>(new {context.Message})));
/ 支付失败
DuringAny(When(TakePaymentEventFaultEvent)
    .TransitionTo(TakePaymentFaultedState)
    .Then(context => context.Publish<Fault<CheckProductStockEvent>>(new {context.Message})));
/ 校验库存失败
DuringAny(When(CheckProductStockFaultEvent)
    .TransitionTo(CheckProductStockFaultedState)
    .Then(context => context.Publish<Fault<OrderProcessInitializationEvent>>(new {context.Message})));
/ 下单失败
DuringAny(When(OrderProcessInitializationFaultEvent)
    .TransitionTo(OrderProcessInitializedFaultedState)
    .Then(context => context.Publish<OrderProcessFailedEvent>(new {OrderId = context.Saga.CorrelationId})));
/ 下单流程失败
DuringAny(When(OrderProcessFailedEvent)
    .TransitionTo(OrderProcessFailedState));