当前位置：网站首页>seata源码解析：seata server各种消息处理流程
seata源码解析：seata server各种消息处理流程
2022-08-04 19:23:00 【51CTO】
seata-server消息处理流程
上一篇文章我们分析了seata-server端启动流程。本篇文章我们来看seata-server消息处理流程。
seata中有一个全局事务协调器DefaultCoordinator，它主要是处理来自RM和TM的请求来做相应的操作，但是实际的执行者并不是DefaultCoordinator，而是DefaultCore
DefaultCore的继承关系如下图，从继承图中我们可以看到其实Core类的实现类才是一个事务管理器。在seata中有4种事务管理模式，所以每种模式有一个具体的事务管理器。
而DefaultCore则是聚合了4种具体的事务管理器，根据事务的不同类型调用不同的事务管理器。组件的关系如下图
所以事务协调的主要工作就是接受请求然后调用事务管理器进行相应的操作。
事务协调器接收请求
之前我们说到，所有的消息都会交给AbstractNettyRemotingServer.ServerHandler来处理，而AbstractNettyRemotingServer.ServerHandler根据消息的不同类型，交给不同的RemotingProcessor来处理
所以我们对那种消息感兴趣只需要看对应的RemotingProcessor实现类即可，我们挑几个常见的消息分析以下，思路都差不多。
事务管理器执行操作
RegRmProcessor和RegTmProcessor
tm和rm这部分注册代码看的我有点晕（不重要就没耐心看下去），主要作用就是在tc保存tm和rm的长连接，当tc需要往tm和rm发送消息的时候，就从ChannelManager中找到对应的长连接，然后发送消息
各模式中rm注册的时机如下
xa模式：构建DataSourceProxyXA
at模式：构建DataSourceProxy
tcc模式：GlobalTransactionScanner（Bean初始化后阶段），生成代理对象的时候判定这个方法是tcc的prepare方法
ServerOnRequestProcessor
在TC端，全局事务的状态被保存在GlobalSession对象中，分支事务的状态被保存在BranchSession中
ServerOnRequestProcessor处理消息的公共流程为
- 对应的channel是否注册过，没注册过直接关闭连接，否则到第二步
- 针对不同的消息交给DefaultCoordinator类的不同方法来处理，并返回结果
开启全局事务
// DefaultCoordinator
protected
void
doGlobalBegin(
GlobalBeginRequest
request,
GlobalBeginResponse
response,
RpcContext
rpcContext)
throws
TransactionException {
response.
setXid(
core.
begin(
rpcContext.
getApplicationId(),
rpcContext.
getTransactionServiceGroup(),
request.
getTransactionName(),
request.
getTimeout()));
if (
LOGGER.
isInfoEnabled()) {
LOGGER.
info(
"Begin new global transaction applicationId: {},transactionServiceGroup: {}, transactionName: {},timeout:{},xid:{}",
rpcContext.
getApplicationId(),
rpcContext.
getTransactionServiceGroup(),
request.
getTransactionName(),
request.
getTimeout(),
response.
getXid());
}
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
消息的接收是通过DefaultCoordinator，然后交给DefaultCore来执行对应的操作，DefaultCore生成xid并返回
// DefaultCore
public
String
begin(
String
applicationId,
String
transactionServiceGroup,
String
name,
int
timeout)
throws
TransactionException {
// 创建一个 GlobalSession
GlobalSession
session
=
GlobalSession.
createGlobalSession(
applicationId,
transactionServiceGroup,
name,
timeout);
MDC.
put(
RootContext.
MDC_KEY_XID,
session.
getXid());
// 将 ROOT_SESSION_MANAGER 加入到这个 GlobalSession 的监听器列表中
session.
addSessionLifecycleListener(
SessionHolder.
getRootSessionManager());
// 开启 GlobalSession
session.
begin();
// 发布事件，如果你对这个事件感兴趣，可以注册这个事件
// transaction start event
eventBus.
post(
new
GlobalTransactionEvent(
session.
getTransactionId(),
GlobalTransactionEvent.
ROLE_TC,
session.
getTransactionName(),
applicationId,
transactionServiceGroup,
session.
getBeginTime(),
null,
session.
getStatus()));
// 返回 xid
return
session.
getXid();
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
- 15.
- 16.
- 17.
- 18.
- 19.
- 20.
- 21.
可以只返回一个xid，xid由DefaultCore#begin方法生成，xid的生成策略如下
从GlobalSession#begin方法可以看到GlobalSession用到了观察者模式，当GlobalSession的状态发生变更时，会通过给相应的观察者，观察者都是SessionManager，当接收到相应的事件后，将变更的状态进行持久化存储，当使用db模式存储时，这里会在global_table中插入一条记录。
// GlobalSession
public
void
begin()
throws
TransactionException {
this.
status
=
GlobalStatus.
Begin;
this.
beginTime
=
System.
currentTimeMillis();
this.
active
=
true;
for (
SessionLifecycleListener
lifecycleListener :
lifecycleListeners) {
lifecycleListener.
onBegin(
this);
}
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
注册分支事务
// AbstractCore
public
Long
branchRegister(
BranchType
branchType,
String
resourceId,
String
clientId,
String
xid,
String
applicationData,
String
lockKeys)
throws
TransactionException {
// 根据 xid 从 SessionManager 中获取到 GlobalSession
GlobalSession
globalSession
=
assertGlobalSessionNotNull(
xid,
false);
return
SessionHolder.
lockAndExecute(
globalSession, ()
-> {
globalSessionStatusCheck(
globalSession);
globalSession.
addSessionLifecycleListener(
SessionHolder.
getRootSessionManager());
// 创建新的分支事务即 branchSession
BranchSession
branchSession
=
SessionHelper.
newBranchByGlobal(
globalSession,
branchType,
resourceId,
applicationData,
lockKeys,
clientId);
MDC.
put(
RootContext.
MDC_KEY_BRANCH_ID,
String.
valueOf(
branchSession.
getBranchId()));
// 对分支事务需要的资源加锁，加锁的逻辑在别的文章详解
branchSessionLock(
globalSession,
branchSession);
try {
// 将 branchSession 加到 globalSession 的属性中
globalSession.
addBranch(
branchSession);
}
catch (
RuntimeException
ex) {
branchSessionUnlock(
branchSession);
throw
new
BranchTransactionException(
FailedToAddBranch,
String
.
format(
"Failed to store branch xid = %s branchId = %s",
globalSession.
getXid(),
branchSession.
getBranchId()),
ex);
}
if (
LOGGER.
isInfoEnabled()) {
LOGGER.
info(
"Register branch successfully, xid = {}, branchId = {}, resourceId = {} ,lockKeys = {}",
globalSession.
getXid(),
branchSession.
getBranchId(),
resourceId,
lockKeys);
}
return
branchSession.
getBranchId();
});
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
- 15.
- 16.
- 17.
- 18.
- 19.
- 20.
- 21.
- 22.
- 23.
- 24.
- 25.
- 26.
- 27.
- 28.
- 29.
- 30.
- 根据 xid 从 SessionManager 中获取到 GlobalSession
- 创建新的分支事务即 BranchSession
- 将 branchSession 加到 globalSession 的属性中，此时GlobalSession会发布分支事务注册事件，SessionManager 收到事件后会在 branch_table 中插入一条记录
注意：AT模式下，当分支事务注册的时候，会将修改的数据加锁，如果加锁失败，则抛出异常
提交全局事务
// DefaultCore
public
GlobalStatus
commit(
String
xid)
throws
TransactionException {
// 根据xid找到全局事务对象GlobalSession
GlobalSession
globalSession
=
SessionHolder.
findGlobalSession(
xid);
if (
globalSession
==
null) {
// 已经被commit过了，直接返回成功
return
GlobalStatus.
Finished;
}
// 添加监听器
globalSession.
addSessionLifecycleListener(
SessionHolder.
getRootSessionManager());
// just lock changeStatus
boolean
shouldCommit
=
SessionHolder.
lockAndExecute(
globalSession, ()
-> {
// Highlight: Firstly, close the session, then no more branch can be registered.
// 关闭 GlobalSession 防止再次有新的 BranchSession 注册进来
globalSession.
closeAndClean();
if (
globalSession.
getStatus()
==
GlobalStatus.
Begin) {
// 判断是否可以异步提交
// 目前只有at模式可以异步提交，因为是通过undolog的方式去做的
if (
globalSession.
canBeCommittedAsync()) {
globalSession.
asyncCommit();
return
false;
}
else {
globalSession.
changeStatus(
GlobalStatus.
Committing);
return
true;
}
}
return
false;
});
// 同步提交
// XA/TCC只能同步提交
if (
shouldCommit) {
boolean
success
=
doGlobalCommit(
globalSession,
false);
//If successful and all remaining branches can be committed asynchronously, do async commit.
if (
success
&&
globalSession.
hasBranch()
&&
globalSession.
canBeCommittedAsync()) {
globalSession.
asyncCommit();
return
GlobalStatus.
Committed;
}
else {
return
globalSession.
getStatus();
}
}
else {
// 异步提交
// 只有AT模式能异步提交
return
globalSession.
getStatus()
==
GlobalStatus.
AsyncCommitting
?
GlobalStatus.
Committed :
globalSession.
getStatus();
}
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
- 15.
- 16.
- 17.
- 18.
- 19.
- 20.
- 21.
- 22.
- 23.
- 24.
- 25.
- 26.
- 27.
- 28.
- 29.
- 30.
- 31.
- 32.
- 33.
- 34.
- 35.
- 36.
- 37.
- 38.
- 39.
- 40.
- 41.
- 42.
- 43.
- 44.
- 45.
- 46.
- 47.
public
boolean
doGlobalCommit(
GlobalSession
globalSession,
boolean
retrying)
throws
TransactionException {
boolean
success
=
true;
// start committing event
// 发布事件
eventBus.
post(
new
GlobalTransactionEvent(
globalSession.
getTransactionId(),
GlobalTransactionEvent.
ROLE_TC,
globalSession.
getTransactionName(),
globalSession.
getApplicationId(),
globalSession.
getTransactionServiceGroup(),
globalSession.
getBeginTime(),
null,
globalSession.
getStatus()));
if (
globalSession.
isSaga()) {
success
=
getCore(
BranchType.
SAGA).
doGlobalCommit(
globalSession,
retrying);
}
else {
// 取出所有的分支事务，然后提交
Boolean
result
=
SessionHelper.
forEach(
globalSession.
getSortedBranches(),
branchSession
-> {
// if not retrying, skip the canBeCommittedAsync branches
if (
!
retrying
&&
branchSession.
canBeCommittedAsync()) {
return
CONTINUE;
}
BranchStatus
currentStatus
=
branchSession.
getStatus();
// 一阶段失败
if (
currentStatus
==
BranchStatus.
PhaseOne_Failed) {
globalSession.
removeBranch(
branchSession);
return
CONTINUE;
}
try {
BranchStatus
branchStatus
=
getCore(
branchSession.
getBranchType()).
branchCommit(
globalSession,
branchSession);
switch (
branchStatus) {
case
PhaseTwo_Committed:
globalSession.
removeBranch(
branchSession);
return
CONTINUE;
case
PhaseTwo_CommitFailed_Unretryable:
if (
globalSession.
canBeCommittedAsync()) {
LOGGER.
error(
"Committing branch transaction[{}], status: PhaseTwo_CommitFailed_Unretryable, please check the business log.",
branchSession.
getBranchId());
return
CONTINUE;
}
else {
// 分支事务，不能异步提交，并且还不重试，全局事务执行失败
SessionHelper.
endCommitFailed(
globalSession);
LOGGER.
error(
"Committing global transaction[{}] finally failed, caused by branch transaction[{}] commit failed.",
globalSession.
getXid(),
branchSession.
getBranchId());
return
false;
}
default:
// 当前是否正在重试
// retrying=true，说明是从重试队列进来的任务，不用再往重试队列放了
if (
!
retrying) {
globalSession.
queueToRetryCommit();
return
false;
}
if (
globalSession.
canBeCommittedAsync()) {
LOGGER.
error(
"Committing branch transaction[{}], status:{} and will retry later",
branchSession.
getBranchId(),
branchStatus);
return
CONTINUE;
}
else {
LOGGER.
error(
"Committing global transaction[{}] failed, caused by branch transaction[{}] commit failed, will retry later.",
globalSession.
getXid(),
branchSession.
getBranchId());
return
false;
}
}
}
catch (
Exception
ex) {
StackTraceLogger.
error(
LOGGER,
ex,
"Committing branch transaction exception: {}",
new
String[] {
branchSession.
toString()});
if (
!
retrying) {
globalSession.
queueToRetryCommit();
throw
new
TransactionException(
ex);
}
}
return
CONTINUE;
});
// Return if the result is not null
// result 不为null 则为 false
if (
result
!=
null) {
return
result;
}
//If has branch and not all remaining branches can be committed asynchronously,
//do print log and return false
// 有分支事务，并且不允许异步提交，说明失败了
if (
globalSession.
hasBranch()
&&
!
globalSession.
canBeCommittedAsync()) {
LOGGER.
info(
"Committing global transaction is NOT done, xid = {}.",
globalSession.
getXid());
return
false;
}
}
//If success and there is no branch, end the global transaction.
// 分支事务全部提交成功了
if (
success
&&
globalSession.
getBranchSessions().
isEmpty()) {
// 全局事务状态改为已提交
SessionHelper.
endCommitted(
globalSession);
// committed event
eventBus.
post(
new
GlobalTransactionEvent(
globalSession.
getTransactionId(),
GlobalTransactionEvent.
ROLE_TC,
globalSession.
getTransactionName(),
globalSession.
getApplicationId(),
globalSession.
getTransactionServiceGroup(),
globalSession.
getBeginTime(),
System.
currentTimeMillis(),
globalSession.
getStatus()));
LOGGER.
info(
"Committing global transaction is successfully done, xid = {}.",
globalSession.
getXid());
}
return
success;
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
- 15.
- 16.
- 17.
- 18.
- 19.
- 20.
- 21.
- 22.
- 23.
- 24.
- 25.
- 26.
- 27.
- 28.
- 29.
- 30.
- 31.
- 32.
- 33.
- 34.
- 35.
- 36.
- 37.
- 38.
- 39.
- 40.
- 41.
- 42.
- 43.
- 44.
- 45.
- 46.
- 47.
- 48.
- 49.
- 50.
- 51.
- 52.
- 53.
- 54.
- 55.
- 56.
- 57.
- 58.
- 59.
- 60.
- 61.
- 62.
- 63.
- 64.
- 65.
- 66.
- 67.
- 68.
- 69.
- 70.
- 71.
- 72.
- 73.
- 74.
- 75.
- 76.
- 77.
- 78.
- 79.
- 80.
- 81.
- 82.
- 83.
- 84.
- 85.
- 86.
- 87.
- 88.
- 89.
- 90.
- 91.
- 92.
- 93.
- 94.
- 95.
- 96.
- 97.
可以看到AT模式可以异步提交，因为AT模式全局提交只是删除undoLog，异步提交可以提高执行效率。而其他模式得同步提交，依次向RM发送分支事务提交请求，当所有分支事务都执行成功后，全局事务提交成功。否则，将任务交给管理重试的SessionManager进行重试
全局事务的提交和回滚逻辑差不多，回滚逻辑就不分析了
ServerOnResponseProcessor
当我们需要进行全局提交时，需要向各个RM发送对应的请求，注意发送的是同步请求，阻塞获取结果。
实现思路主要是如下一个map
每个消息有一个消息id，当发送的时候给每条消息创建一个MessageFuture，放在futures中，然后这个MessageFuture（底层其实就是CompletableFuture）阻塞获取结果
而ServerOnResponseProcessor则是用来接收分支提交（请求和响应对应的消息id是一样的），当收到结果后，设置消息对应的MessageFuture为完成，此时阻塞的同步请求就能获取到结果了
public
class
ServerOnResponseProcessor
implements
RemotingProcessor {
@Override
public
void
process(
ChannelHandlerContext
ctx,
RpcMessage
rpcMessage)
throws
Exception {
// 根据消息id找到对应的MessageFuture
MessageFuture
messageFuture
=
futures.
remove(
rpcMessage.
getId());
if (
messageFuture
!=
null) {
messageFuture.
setResultMessage(
rpcMessage.
getBody());
}
else {
// 没有找到对应的消息发送记录
// 删除部分代码
}
}
}
- 1.
- 2.
- 3.
- 4.
- 5.
- 6.
- 7.
- 8.
- 9.
- 10.
- 11.
- 12.
- 13.
- 14.
参考博客
边栏推荐
猜你喜欢
随机推荐
在线生成接口文档
HCIA-R＆S自用笔记（22）STP状态与计时器、STP拓扑变化、STP配置及实验
零基础做出高端堆叠极环图
Usage of collect_list in Scala105-Spark.sql
win10 uwp xaml 绑定接口
Defaced Fingerprint Recovery and Identification
猜数字游戏
使用.NET简单实现一个Redis的高性能克隆版（二）
Storage resource activation system to help new infrastructure
HCIP-R＆S By Wakin自用笔记（1）企业网络高级解决方案
03 ts类型缩小,函数
Yuanguo chain game system development
测试/开发程序员男都秃头？女都满脸痘痘？过好我们“短暂“的一生......
ros2订阅esp32发布的电池电压数据
PostgreSQL的 SPI_接口函数
Orthodontic MIA micro-implant anchorage technology China 10th anniversary exchange meeting was held in Shenyang
切换node版本和切换npm源工具
The upgrade of capacity helps the flow of computing power, the acceleration moment of China's digital economy
MogDB学习笔记-环境准备及单实例安装
win10 uwp json