saga分布式事务

var builder = WebApplication.CreateBuilder(args);

builder.Services.AddDtmcli(x =>
{
    x.DtmUrl = "http://localhost:36789";
});

var app = builder.Build();

app.MapPost("/api/TransIn", (HttpContext httpContext, TransRequest req) =>
{
    //
    Console.WriteLine($"TransIn, QueryString={httpContext.Request.QueryString}");
    Console.WriteLine($"User: {req.UserId}, transfer in {req.Amount} --- forward");

    return Results.Ok(TransResponse.BuildSucceedResponse());
});

app.MapPost("/api/TransInCompensate", (HttpContext httpContext, TransRequest req) =>
{
    Console.WriteLine($"TransInCompensate, QueryString={httpContext.Request.QueryString}");
    Console.WriteLine($"User: {req.UserId}, transfer out {req.Amount} --- reverse compensation");

    return Results.Ok(TransResponse.BuildSucceedResponse());
});

app.Run("http://*:10001");

注意：为简单起见，这里不介绍具体的数据库操作，而是用日志输出代替。

添加一个控制台应用程序来模拟事务发起者。我们应该从 nuget 添加以下包之一。

dotnet add package Dtmcli --version 1.0.0

or

dotnet add package Dtmgrpc --version 1.0.0

注意：Dtmcli 使用 HTTP 协议与 DTM 服务器通信，而 Dtmgrpc 使用 GRPC 协议。

以下示例将用于 Dtmcli 演示如何提交 SAGA 事务。

var services = new ServiceCollection();

services.AddDtmcli(x =>
{
    x.DtmUrl = "http://localhost:36789";
});

var provider = services.BuildServiceProvider();
var dtmClient = provider.GetRequiredService<IDtmClient>();

var outApi = "http://192.168.0.101:10000/api";
var inApi = "http://192.168.0.101:10001/api";

var userOutReq = new Common.TransRequest() { UserId = "1", Amount = -30 };
var userInReq = new Common.TransRequest() { UserId = "2", Amount = 30 };

var cts = new CancellationTokenSource();

var gid = await dtmClient.GenGid(cts.Token);
var saga = new Saga(dtmClient, gid)
        .Add(outApi + "/TransOut", outApi + "/TransOutCompensate", userOutReq)
        .Add(inApi + "/TransIn", inApi + "/TransInCompensate", userInReq)
        .EnableWaitResult()
        ;

await saga.Submit(cts.Token);

Console.WriteLine($"Submit SAGA transaction, gid is {gid}.");

几个简单的步骤，一个完整的SAGA分布式事务就写好了！！

让我们看看这个例子的结果是什么。

运行本例后，从应用日志中可以看到转入转出都成功了，从dtm服务器的日志中可以看到DTM服务器是如何调用微服务API的。

虽然已经完成了一个成功的例子，但是这个例子太简单了，转入转出都是一次性成功，并不能体现微服务世界遇到的网络抖动、机器宕机、进程崩溃等问题。

在事务域中，异常是一个关键的考虑因素，它可能导致不一致。当我们在做分布式事务的时候，那么分布式中的异常出现的频率就更高了，异常的设计和处理就显得更加重要了。

针对这些异常，DTM首创了子事务屏障技术，可以帮助我们轻松解决异常问题，大大降低了使用分布式事务的门槛。

子事务屏障可以达到如下效果，见图。

var saga = new Saga(dtmClient, gid)
        .Add(outApi + "/TransOut", outApi + "/TransOutCompensate", userOutReq)
        .Add(inApi + "/TransInError", inApi + "/TransInCompensate", userInReq)
        .EnableWaitResult()
        ;

在这种情况下会发生什么？

最直接的结果是分布式事务直接失败，因为事务发起者抛出了异常。

事务失败很常见，我们要确保数据的一致性是正确的。

但是还有一种更严重的情况，传输的前向操作失败，但是传输的回滚操作也进行了！本来，受让方在这笔交易中本应多出一些钱，但实际上是亏本。

做正向操作的补偿是容易出错的，因为正向操作的失败可能发生在事务提交之前或之后。子事务屏障为我们处理了这些情况。如果在承诺之后发生前向操作失败，障碍将要求赔偿；如果失败发生在承诺之前，障碍将忽略补偿。

子事务屏障技术的原理是在本地数据库中创建一个分支操作状态表dtm_barrier，唯一键为 gid-branch_id-branch_op.

1. 打开本地事务
2. 对于当前操作op(forward|try|confirm|cancel)，insert忽略一行gid-branchid-[op]，如果插入不成功，commit事务返回成功（常用幂等性控制方式）
3. 如果当前操作是cancel|reverse，则insert忽略一行 gid-branchid-try，如果插入成功（注意是成功），则提交事务并返回成功
4. 调用barrier内的业务逻辑，如果业务返回成功，则事务提交返回成功；如果业务返回失败，则事务回滚并返回失败

接下来让我们尝试这种子事务屏障技术。

SDK 为我们提供了以下方法。

public Task Call(DbConnection db, Func<DbTransaction, Task> busiCall);

我们需要做的是在 busiCall 中编写我们自己的逻辑。

以下代码改进了转入的补偿 API，并演示了如何使用子事务屏障。

app.MapPost("/api/BarrierTransInCompensate", async (HttpContext httpContext, TransRequest req, IBranchBarrierFactory factory) =>
{
    Console.WriteLine($"TransIn, QueryString={httpContext.Request.QueryString}");

    // create barrier from query
    var barrier = factory.CreateBranchBarrier(httpContext.Request.Query);

    using var db = Db.GeConn();
    await barrier.Call(db, async (tx) =>
    {
        // some exception occure, should not output this one.
        Console.WriteLine($"User: {req.UserId}, transfer in {req.Amount} --- sub-transaction handle!");

        await Task.CompletedTask;
    });

    return Results.Ok(TransResponse.BuildSucceedResponse());
});

我们希望 sub-transaction handle 这里不会输出日志。

var saga = new Saga(dtmClient, gid)
        .Add(outApi + "/TransOut", outApi + "/TransOutCompensate", userOutReq)
        .Add(inApi + "/TransInError", inApi + "/BarrierTransInCompensate", userInReq)
        .EnableWaitResult()
        ;

Will not exec busiCall, isNullCompensation=True, isDuplicateOrPend=False