Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

Commitc86f875

Browse files
authored
Add http xa sample (#71)
* add http xa sample* Add XA pattern comment in README* Change the DBType of DtmOptions to SqlDbType in the configuration section of the README
1 parent37e4945 commitc86f875

File tree

4 files changed

+188
-13
lines changed

4 files changed

+188
-13
lines changed

‎README-cn.md‎

Lines changed: 25 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -87,7 +87,7 @@ services.AddDtmcli(x =>
8787
x.BranchTimeout=10000;
8888

8989
// 子事务屏障的数据库类型, mysql, postgres, sqlserver
90-
x.DBType="mysql";
90+
x.SqlDbType="mysql";
9191

9292
// 子事务屏障的数据表名
9393
x.BarrierTableName="dtm_barrier.barrier";
@@ -108,7 +108,7 @@ services.AddDtmcli(Configuration, "dtm");
108108
"DtmUrl":"http://localhost:36789",
109109
"DtmTimeout":10000,
110110
"BranchTimeout":10000,
111-
"DBType":"mysql",
111+
"SqlDbType":"mysql",
112112
"BarrierTableName":"dtm_barrier.barrier",
113113
}
114114
}
@@ -265,12 +265,32 @@ public class MyBusi
265265
this._globalTransaction=globalTransaction;
266266
}
267267

268-
publicasyncTaskDoBusAsync()
268+
publicasyncTaskDoBusAsync(CancellationTokencancellationToken)
269269
{
270+
varsvc="http://localhost:5005";
271+
270272
await_globalTransaction.ExcecuteAsync(async (Xaxa)=>
271273
{
272-
awaitxa.CallBranch(newTransRequest("1",-30),_settings.BusiUrl+"/XaTransOut",cancellationToken);
273-
awaitxa.CallBranch(newTransRequest("2",30),_settings.BusiUrl+"/XaTransIn",cancellationToken);
274+
// NOTE: XA 模式的限制
275+
// 当前模式仅支持mysql、postgresDB，请修改相应的客户端配置，如SqlDbType等。
276+
// 如使用Mysql并且版本低于8.0需关闭连接池使用
277+
278+
// 调用 XA 子事务
279+
awaitxa.CallBranch(
280+
// 参数
281+
newTransRequest("1",-30),
282+
283+
// 操作的 URL
284+
svc+"/XaTransOut",
285+
286+
// 取消令牌
287+
cancellationToken);
288+
289+
awaitxa.CallBranch(
290+
newTransRequest("2",30),
291+
svc+"/XaTransIn",
292+
cancellationToken);
293+
274294
},cancellationToken);
275295
}
276296
}

‎README.md‎

Lines changed: 25 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -94,7 +94,7 @@ services.AddDtmcli(x =>
9494
x.BranchTimeout=10000;
9595

9696
// barrier database type, mysql, postgres, sqlserver
97-
x.DBType="mysql";
97+
x.SqlDbType="mysql";
9898

9999
// barrier table name
100100
x.BarrierTableName="dtm_barrier.barrier";
@@ -115,7 +115,7 @@ And the configuration file
115115
"DtmUrl":"http://localhost:36789",
116116
"DtmTimeout":10000,
117117
"BranchTimeout":10000,
118-
"DBType":"mysql",
118+
"SqlDbType":"mysql",
119119
"BarrierTableName":"dtm_barrier.barrier",
120120
}
121121
}
@@ -273,12 +273,32 @@ public class MyBusi
273273
this._globalTransaction=globalTransaction;
274274
}
275275

276-
publicasyncTaskDoBusAsync()
276+
publicasyncTaskDoBusAsync(CancellationTokencancellationToken)
277277
{
278+
varsvc="http://localhost:5005";
279+
278280
await_globalTransaction.ExcecuteAsync(async (Xaxa)=>
279281
{
280-
awaitxa.CallBranch(newTransRequest("1",-30),_settings.BusiUrl+"/XaTransOut",cancellationToken);
281-
awaitxa.CallBranch(newTransRequest("2",30),_settings.BusiUrl+"/XaTransIn",cancellationToken);
282+
// NOTE: Limitations of using Xa mode
283+
// The current mode only supports mysql, postgresDB, please modify the corresponding client configuration, such as SqlDbType, etc.
284+
// Connection pooling needs to be turned off for mysql versions below 8.0
285+
286+
// Create XA sub-transaction
287+
awaitxa.CallBranch(
288+
// Arguments of action
289+
newTransRequest("1",-30),
290+
291+
// URL of action
292+
svc+"/XaTransOut",
293+
294+
// Cancel token
295+
cancellationToken);
296+
297+
awaitxa.CallBranch(
298+
newTransRequest("2",30),
299+
svc+"/XaTransIn",
300+
cancellationToken);
301+
282302
},cancellationToken);
283303
}
284304
}

‎samples/DtmSample/Controllers/TransController.cs‎

Lines changed: 46 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,12 @@
1-
usingDtmSample.Dtos;
1+
usingDapper;
2+
usingDtmcli;
3+
usingDtmSample.Dtos;
4+
usingMicrosoft.AspNetCore.Http;
25
usingMicrosoft.AspNetCore.Mvc;
36
usingMicrosoft.Extensions.Logging;
7+
usingMySqlConnector;
8+
usingSystem.Threading;
9+
usingSystem.Threading.Tasks;
410

511
namespaceDtmSample.Controllers
612
{
@@ -10,10 +16,12 @@ namespace DtmSample.Controllers
1016
publicclassTransController:ControllerBase
1117
{
1218
privatereadonlyILogger<TransController>_logger;
19+
privatereadonlyXaLocalTransaction_xaTrans;
1320

14-
publicTransController(ILogger<TransController>logger)
21+
publicTransController(ILogger<TransController>logger,XaLocalTransactionxaTrans)
1522
{
1623
_logger=logger;
24+
_xaTrans=xaTrans;
1725
}
1826

1927
#region TCC
@@ -119,7 +127,42 @@ public IActionResult TransInRevert([FromBody] TransRequest body)
119127
_logger.LogInformation("TransInConfirm, QueryString={0}",Request.QueryString);
120128
_logger.LogInformation("用户: {0},转入 {1} 元---回滚",body.UserId,body.Amount);
121129
returnOk(TransResponse.BuildSucceedResponse());
122-
}
130+
}
131+
#endregion
132+
133+
#region Xa
134+
[HttpPost("XaTransOut")]
135+
publicasyncTask<IActionResult>XaTransOut(CancellationTokentoken)
136+
{
137+
//todo: Connection pooling needs to be turned off for mysql versions below 8.0
138+
usingvarconn=newMySqlConnection("Server=en.dtm.pub; Port=3306; User ID=dtm; Password=passwd123dtm; Database=dtm_busi;Pooling=False");
139+
awaitthis._xaTrans.ExcecuteAsync(this.Request.Query,conn,"mysql",async(dbConn,xa)=>
140+
{
141+
varbody=awaitthis.Request.ReadFromJsonAsync<TransRequest>();
142+
awaitdbConn.ExecuteAsync($"UPDATE dtm_busi.user_account SET balance = balance +{body.Amount} where user_id = '{body.UserId}'");
143+
_logger.LogInformation("XaTransOut, QueryString={0}",Request.QueryString);
144+
_logger.LogInformation("用户: {0},转出 {1} 元",body.UserId,body.Amount);
145+
},token);
146+
147+
returnOk(TransResponse.BuildSucceedResponse());
148+
}
149+
150+
[HttpPost("XaTransIn")]
151+
publicasyncTask<IActionResult>XaTransIn(CancellationTokentoken)
152+
{
153+
//todo: Connection pooling needs to be turned off for mysql versions below 8.0
154+
usingvarconn=newMySqlConnection("Server=en.dtm.pub; Port=3306; User ID=dtm; Password=passwd123dtm; Database=dtm_busi;Pooling=False");
155+
awaitthis._xaTrans.ExcecuteAsync(this.Request.Query,conn,"mysql",async(dbConn,xa)=>
156+
{
157+
varbody=awaitthis.Request.ReadFromJsonAsync<TransRequest>();
158+
awaitdbConn.ExecuteAsync($"UPDATE dtm_busi.user_account SET balance = balance +{body.Amount} where user_id = '{body.UserId}'");
159+
_logger.LogInformation("XaTransIn, QueryString={0}",Request.QueryString);
160+
_logger.LogInformation("用户: {0},转入 {1} 元",body.UserId,body.Amount);
161+
},token);
162+
163+
returnOk(TransResponse.BuildSucceedResponse());
164+
}
123165
#endregion
124166
}
125167
}
168+
Lines changed: 92 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,92 @@
1+
usingDtmcli;
2+
usingDtmSample.Dtos;
3+
usingMicrosoft.AspNetCore.Mvc;
4+
usingMicrosoft.Extensions.Logging;
5+
usingMicrosoft.Extensions.Options;
6+
usingSystem;
7+
usingSystem.Threading;
8+
usingSystem.Threading.Tasks;
9+
10+
namespaceDtmSample.Controllers
11+
{
12+
/// <summary>
13+
/// XA 示例
14+
/// </summary>
15+
[ApiController]
16+
[Route("/api")]
17+
publicclassXaTestController:ControllerBase
18+
{
19+
20+
privatereadonlyILogger<TccTestController>_logger;
21+
privatereadonlyXaGlobalTransaction_globalTransaction;
22+
privatereadonlyAppSettings_settings;
23+
24+
publicXaTestController(ILogger<TccTestController>logger,IOptions<AppSettings>optionsAccs,XaGlobalTransactiontransaction)
25+
{
26+
_logger=logger;
27+
_settings=optionsAccs.Value;
28+
_globalTransaction=transaction;
29+
}
30+
31+
/// <summary>
32+
/// Xa 成功提交
33+
/// </summary>
34+
/// <param name="cancellationToken"></param>
35+
/// <returns></returns>
36+
[HttpPost("commit")]
37+
publicasyncTask<IActionResult>Commit(CancellationTokencancellationToken)
38+
{
39+
//todo: Currently only supported by mysql, please modify the appsettings.json
40+
try
41+
{
42+
await_globalTransaction.ExcecuteAsync(async(Xaxa)=>
43+
{
44+
//// 用户1 转出30元
45+
varres1=awaitxa.CallBranch(newTransRequest("1",-30),_settings.BusiUrl+"/XaTransOut",cancellationToken);
46+
47+
//// 用户2 转入30元
48+
varres2=awaitxa.CallBranch(newTransRequest("2",30),_settings.BusiUrl+"/XaTransIn",cancellationToken);
49+
},cancellationToken);
50+
51+
returnOk(TransResponse.BuildSucceedResponse());
52+
}
53+
catch(Exceptionex)
54+
{
55+
_logger.LogError(ex,"Xa Error");
56+
returnOk(TransResponse.BuildFailureResponse());
57+
}
58+
}
59+
60+
61+
/// <summary>
62+
/// Xa 失败回滚
63+
/// </summary>
64+
/// <param name="cancellationToken"></param>
65+
/// <returns></returns>
66+
[HttpPost("rollbcak")]
67+
publicasyncTask<IActionResult>Rollbcak(CancellationTokencancellationToken)
68+
{
69+
//todo: Currently only supported by mysql, please modify the appsettings.json
70+
try
71+
{
72+
await_globalTransaction.ExcecuteAsync(async(Xaxa)=>
73+
{
74+
//// 用户1 转出30元
75+
varres1=awaitxa.CallBranch(newTransRequest("1",-30),_settings.BusiUrl+"/XaTransOut",cancellationToken);
76+
77+
//// 用户2 转入30元
78+
varres2=awaitxa.CallBranch(newTransRequest("2",30),_settings.BusiUrl+"/XaTransIn",cancellationToken);
79+
80+
thrownewException("rollbcak");
81+
},cancellationToken);
82+
83+
returnOk(TransResponse.BuildSucceedResponse());
84+
}
85+
catch(Exceptionex)
86+
{
87+
_logger.LogError(ex,"Xa Error");
88+
returnOk(TransResponse.BuildFailureResponse());
89+
}
90+
}
91+
}
92+
}

0 commit comments

Comments
 (0)
[8]ページ先頭
©2009-2025 Movatter.jp