二阶段消息是 DTM
新提出的,可以完美代替现有的事务消息和本地消息表架构。无论从复杂度、性能、便利性还是代码量都是完胜现有的方案。
相比现有的消息架构借助于各种消息中间件比如 RocketMQ
等, DTM
自己实现了无需额外的学习成本。它能够保证本地事务的提交和全局事务提交是“原子的”,适合解决不需要回滚的分布式事务场景。
二阶段消息保证提交的原子性和如何保证业务成功执行如下时序图:
二阶段消息主要是指 Prepare
和 Submit
两个阶段,主程序向 DTM
服务发送 Prepare
消息,成功后执行本地事务,完成本地事务后发送 Submit
消息至 DTM
服务,之后 DTM
会调用分支事件执行其他服务,最后完成全局事务。
当发送了 Prepare
但是 Submit
没有提交的话,会进行回调请求来确认消息的情况,具体工作过程如下:
1、在处理本地事务时,会将 gid
插入到 barrier
表中,同时带上插入原因为 committed
。该表有一个唯一索引,主要字段为 gid
。
2、当进行回查时,二阶段消息的操作不是直接查 gid
是否存在,而是再 insert ignore
一条带有相同 gid
的数据,同时带上插入原因为 rollbacked
。此时如果表中如果已有 gid
的记录,那么新的插入操作就会被 ignore
,否则数据会被插入。
3、然后再用 gid
查询表中的记录,如果查到记录的 reason
为 committed
,那么说明本地事务已提交;如果查到记录的 reason
为 rollbacked
,那么说明本地事务已回滚。
我使用二进制包下载安装 地址 ,我是 Window
环境所以下载后解压,点击 dtm.exe
进行运行即可,如下启动成功
启动成功后可以访问 http://localhost:36789
,进入管理后台
我们需要创建一个表处理消息的回查,表里保存全局事务ID,具体作用在后续说明,我这里用的SqlServer数据库,所以执行如下:
CREATE TABLE [dbo].[barrier](
[id] bigint NOT NULL IDENTITY(1,1) PRIMARY KEY,
[trans_type] varchar(45) NOT NULL DEFAULT(''),
[gid] varchar(128) NOT NULL DEFAULT(''),
[branch_id] varchar(128) NOT NULL DEFAULT(''),
[op] varchar(45) NOT NULL DEFAULT(''),
[barrier_id] varchar(45) NOT NULL DEFAULT(''),
[reason] varchar(45) NOT NULL DEFAULT(''),
[create_time] datetime NOT NULL DEFAULT(getdate()) ,
[update_time] datetime NOT NULL DEFAULT(getdate())
)
GO
CREATE UNIQUE INDEX[ix_uniq_barrier] ON[dbo].[barrier]([gid] ASC, [branch_id] ASC, [op] ASC, [barrier_id] ASC)
WITH(IGNORE_DUP_KEY = ON)
GO
这里比较关键的是那个唯一索引,有一个 IGNORE_DUP_KEY = ON
,这个其实就是为了等价 mysql
的 insert ignore
表示存在相关字段的信息则不插入,否则就插入数据
当然还支持很多其他的数据库,建表语句可以从这里查看 地址
我们简单的创建两个 .net core webapi
项目进行测试,两个项目都进行相同的如下操作:
** 1、安装Dtmcli和Microsoft.EntityFrameworkCore.SqlServer**
安装 Dtmcli
是因为其中已经帮我们集成了 DTM
客户端 SDK HTTP
版本,想要 GRPC
版本可以安装 Dtmgrpc
。
安装 Microsoft.EntityFrameworkCore.SqlServer
很显然是为了处理数据库。
Install-Package Dtmcli
Install-Package Microsoft.EntityFrameworkCore.SqlServer
2、配置
接下来我们配置服务,先在配置文件 appsetting.json
中添加如下
"AppSettings": {
"DtmUrl": "http://localhost:36789",
"BusiUrl": "http://localhost:5056",
"QueryPreparedUrl": "http://localhost:5046",
"BarrierConn": "Data Source=.;Initial Catalog=HTGL;TrustServerCertificate=True;;Integrated Security=True"}
DtmUrl
: DTM
的监听地址, http
的是 36789
, grpc
的是 36790
BusiUrl
:访问其他服务的地址
QueryPreparedUrl
:回查的地址
BarrierConn
:数据库连接语句
添加一个配置类:
public classAppSettings
{
public string DtmUrl { get; set; }
public string BusiUrl { get; set; }
public string BarrierConn { get; set; }
public string QueryPreparedUrl { get; set; }
}
之后注入服务如下:
builder.Services.AddDtmcli(dtm =>{
dtm.DtmUrl = builder.Configuration.GetValue<string>("AppSettings:DtmUrl");
dtm.SqlDbType =DtmCommon.Constant.Barrier.DBTYPE_SQLSERVER;
dtm.BarrierSqlTableName = "[HTGL].[dbo].[barrier]";
});
builder.Services.Configure<AppSettings>(builder.Configuration.GetSection("AppSettings"));
SqlDbType
:表示使用的数据库类型
BarrierSqlTableName
: Barrier
表的名字
3、添加代码
我们在其中一个项目添加主程序代码如下:
[ApiController]public classDtmController : ControllerBase
{
private readonly ILogger<DtmController>_logger;
private readonlyIDtmClient _dtmClient;
private readonlyIDtmTransFactory _transFactory;
private readonlyAppSettings _settings;
private readonlyIBranchBarrierFactory _factory;
public DtmController(ILogger<DtmController> logger, IDtmClient dtmClient,IDtmTransFactory transFactory, IOptions<AppSettings>settings, IBranchBarrierFactory factory)
{
_logger =logger;
_dtmClient =dtmClient;
_transFactory =transFactory;
_settings =settings.Value;
_factory =factory;
}
private DbConnection GetConn() => newMicrosoft.Data.SqlClient.SqlConnection(_settings.BarrierConn);
[HttpPost("post-dtm-msg")]
public async Task<IActionResult>Get(CancellationToken cancellationToken)
{
//1、创建gid
var gid = await_dtmClient.GenGid(cancellationToken);
//2、设置分支事务
var msg =_transFactory.NewMsg(gid)
.Add(_settings.BusiUrl + "/TransOut", new { id = 123})
.Add(_settings.BusiUrl + "/TransIn", new { id = 321});//3、执行submit
using (DbConnection conn =GetConn())
{
await msg.DoAndSubmitDB(_settings.QueryPreparedUrl + "/msg-queryprepared", conn, async tx =>{
//4、执行本地事务
awaitTask.CompletedTask;
});
}
_logger.LogInformation("result gid is {0}", gid);
return Content("SUCCESS");
}
[HttpGet("msg-queryprepared")]
public async Task<IActionResult>QueryPrepared(CancellationToken cancellationToken)
{
var bb =_factory.CreateBranchBarrier(Request.Query);
_logger.LogInformation("bb {0}", bb);
using (DbConnection conn =GetConn())
{
//回调查询消息状态
var res = awaitbb.QueryPrepared(conn);
return Ok(new { dtm_result =res });
}
}
}
然后我们向另一个服务项目添加如下代码,作为一个简单的服务方法,没有任何操作只是返回成功:
[ApiController]
public classTransController : ControllerBase
{
private readonly ILogger<TransController>_logger;
private readonlyIBranchBarrierFactory _factory;
private readonlyAppSettings _settings;
private DbConnection GetConn() => newMicrosoft.Data.SqlClient.SqlConnection(_settings.BarrierConn);
public TransController(ILogger<TransController> logger, IBranchBarrierFactory factory, IOptions<AppSettings>settings)
{
_logger =logger;
_factory =factory;
_settings =settings.Value;
}
[HttpPost("TransIn")]
public async Task<IResult>In()
{
return Results.Ok(new { dtm_result = "SUCCESS"});
//return Results.Ok(new { dtm_result = "FAILURE" });
}
[HttpPost("TransOut")]
public async Task<IResult>Out()
{
return Results.Ok(new { dtm_result = "SUCCESS"});
}
}
我们正常执行,可以看到下面的动图结果,在执行完本地事务后会访问分支事务,然后数据库表中添加了一条记录
可以在管理后台看到我们请求成功的信息
如果要演示失败,需要做以下修改直接报错,我们可以看到访问了回调方法,然后数据库中看到 rollback
标记的消息
using (DbConnection conn =GetConn())
{
await msg.DoAndSubmitDB(_settings.QueryPreparedUrl + "/msg-queryprepared", conn, async tx =>{
throw new Exception("报错了");
//4、执行本地事务
awaitTask.CompletedTask;
});
}
提交后再宕机演示比较麻烦,我就不演示了,大家意会即可。
如果分支事务返回的不是SUCCESS而是FAILURE会由DTM隔一段时间重新请求,dtm对每个事务的重试是指数退避策略,具体为间隔是每失败一次,间隔加倍,避免过多的重试,导致系统负载异常上升。
如果您经过长时间的的宕机,因指数退避算法导致要很久才会重试。如果您想要手动触发立即重试,您可以手动把相应事务的next_cron_time(Redis存储引擎的该功能还在开发中)修改为当前时间,就会在数秒内被定时轮询,事务就会继续往前执行。
版权说明 : 本文为转载文章, 版权归原作者所有 版权申明
原文链接 : https://www.cnblogs.com/xwc1996/p/17252311.html
内容来源于网络,如有侵权,请联系作者删除!