Seata源码—6.Seata AT模式的数据源代理二
itomcoil 2025-06-15 16:58 8 浏览
大纲
1.Seata的Resource资源接口源码
2.Seata数据源连接池代理的实现源码
3.Client向Server发起注册RM的源码
4.Client向Server注册RM时的交互源码
5.数据源连接代理与SQL句柄代理的初始化源码
6.Seata基于SQL句柄代理执行SQL的源码
7.执行SQL语句前取消自动提交事务的源码
8.执行SQL语句前后构建数据镜像的源码
9.构建全局锁的key和UndoLog数据的源码
10.Seata Client发起分支事务注册的源码
11.Seata Server处理分支事务注册请求的源码
12.将UndoLog写入到数据库与提交事务的源码
13.通过全局锁重试策略组件执行事务的提交
14.注册分支事务时获取全局锁的入口源码
15.Seata Server获取全局锁的具体逻辑源码
16.全局锁和分支事务及本地事务总结
17.提交全局事务以及提交各分支事务的源码
18.全局事务回滚的过程源码
11.Seata Server处理分支事务注册请求的源码
(1)Seata Server收到分支事务注册请求后的处理
(2)
BranchRegisterRequest.handle()的处理
(3)
DefaultCore.branchRegister()的处理
(1)Seata Server收到分支事务注册请求后的处理
Seata Server收到Seata Client发送过来的分支事务注册请求后，首先会将分支事务注册请求交给ServerOnRequestProcessor的process()方法进行处理，然后再将请求交给DefaultCoordinator的onRequest()方法进行处理。
public abstract class AbstractNettyRemotingServer extends AbstractNettyRemoting implements RemotingServer {
...
@ChannelHandler.Sharable
class ServerHandler extends ChannelDuplexHandler {
@Override
public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exception {
if (!(msg instanceof RpcMessage)) {
return;
}
//接下来调用processMessage()方法对解码完毕的RpcMessage对象进行处理
processMessage(ctx, (RpcMessage) msg);
}
}
}
public abstract class AbstractNettyRemoting implements Disposable {
...
protected void processMessage(ChannelHandlerContext ctx, RpcMessage rpcMessage) throws Exception {
if (LOGGER.isDebugEnabled()) {
LOGGER.debug(String.format("%s msgId:%s, body:%s", this, rpcMessage.getId(), rpcMessage.getBody()));
}
Object body = rpcMessage.getBody();
if (body instanceof MessageTypeAware) {
MessageTypeAware messageTypeAware = (MessageTypeAware) body;
//根据消息类型获取到一个Pair对象，该Pair对象是由请求处理组件和请求处理线程池组成的
//processorTable里的内容，是NettyRemotingServer在初始化时，通过调用registerProcessor()方法put进去的
//所以下面的代码实际上会由ServerOnRequestProcessor的process()方法进行处理
final Pair<RemotingProcessor, ExecutorService> pair = this.processorTable.get((int) messageTypeAware.getTypeCode());
if (pair != null) {
if (pair.getSecond() != null) {
try {
pair.getSecond().execute(() -> {
try {
pair.getFirst().process(ctx, rpcMessage);
} catch (Throwable th) {
LOGGER.error(FrameworkErrorCode.NetDispatch.getErrCode(), th.getMessage(), th);
} finally {
MDC.clear();
}
});
} catch (RejectedExecutionException e) {
...
}
} else {
try {
pair.getFirst().process(ctx, rpcMessage);
} catch (Throwable th) {
LOGGER.error(FrameworkErrorCode.NetDispatch.getErrCode(), th.getMessage(), th);
}
}
} else {
LOGGER.error("This message type [{}] has no processor.", messageTypeAware.getTypeCode());
}
} else {
LOGGER.error("This rpcMessage body[{}] is not MessageTypeAware type.", body);
}
}
...
}
public class ServerOnRequestProcessor implements RemotingProcessor, Disposable {
private final RemotingServer remotingServer;
...
@Override
public void process(ChannelHandlerContext ctx, RpcMessage rpcMessage) throws Exception {
if (ChannelManager.isRegistered(ctx.channel())) {
onRequestMessage(ctx, rpcMessage);
} else {
try {
if (LOGGER.isInfoEnabled()) {
LOGGER.info("closeChannelHandlerContext channel:" + ctx.channel());
}
ctx.disconnect();
ctx.close();
} catch (Exception exx) {
LOGGER.error(exx.getMessage());
}
if (LOGGER.isInfoEnabled()) {
LOGGER.info(String.format("close a unhandled connection! [%s]", ctx.channel().toString()));
}
}
}
private void onRequestMessage(ChannelHandlerContext ctx, RpcMessage rpcMessage) {
Object message = rpcMessage.getBody();
//RpcContext线程本地变量副本
RpcContext rpcContext = ChannelManager.getContextFromIdentified(ctx.channel());
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("server received:{},clientIp:{},vgroup:{}", message, NetUtil.toIpAddress(ctx.channel().remoteAddress()), rpcContext.getTransactionServiceGroup());
} else {
try {
BatchLogHandler.INSTANCE.getLogQueue().put(message + ",clientIp:" + NetUtil.toIpAddress(ctx.channel().remoteAddress()) + ",vgroup:" + rpcContext.getTransactionServiceGroup());
} catch (InterruptedException e) {
LOGGER.error("put message to logQueue error: {}", e.getMessage(), e);
}
}
if (!(message instanceof AbstractMessage)) {
return;
}
//the batch send request message
if (message instanceof MergedWarpMessage) {
...
} else {
//the single send request message
final AbstractMessage msg = (AbstractMessage) message;
//最终调用到DefaultCoordinator的onRequest()方法来处理RpcMessage
//此时传入的msg其实就是客户端发送请求时的BranchRegisterRequest对象
AbstractResultMessage result = transactionMessageHandler.onRequest(msg, rpcContext);
//返回响应给客户端
remotingServer.sendAsyncResponse(rpcMessage, ctx.channel(), result);
}
}
...
}
public class DefaultCoordinator extends AbstractTCInboundHandler implements TransactionMessageHandler, Disposable {
...
@Override
public AbstractResultMessage onRequest(AbstractMessage request, RpcContext context) {
if (!(request instanceof AbstractTransactionRequestToTC)) {
throw new IllegalArgumentException();
}
//此时传入的request其实就是客户端发送请求时的BranchRegisterRequest对象
AbstractTransactionRequestToTC transactionRequest = (AbstractTransactionRequestToTC) request;
transactionRequest.setTCInboundHandler(this);
return transactionRequest.handle(context);
}
...
}(2)
BranchRegisterRequest.handle()的处理
在DefaultCoordinator的onRequest()方法中，会调用BranchRegisterRequest的handle()方法来处理分支事务注册请求，该handle()方法又会调用DefaultCoordinator的doBranchRegister()方法，所以最后会调用DefaultCore的branchRegister()方法来具体处理分支事务注册请求。
public class BranchRegisterRequest extends AbstractTransactionRequestToTC {
...
@Override
public AbstractTransactionResponse handle(RpcContext rpcContext) {
return handler.handle(this, rpcContext);
}
...
}
public interface TCInboundHandler {
...
//Handle branch register response.
BranchRegisterResponse handle(BranchRegisterRequest branchRegister, RpcContext rpcContext);
}
public abstract class AbstractTCInboundHandler extends AbstractExceptionHandler implements TCInboundHandler {
...
@Override
public BranchRegisterResponse handle(BranchRegisterRequest request, final RpcContext rpcContext) {
BranchRegisterResponse response = new BranchRegisterResponse();
exceptionHandleTemplate(new AbstractCallback<BranchRegisterRequest, BranchRegisterResponse>() {
@Override
public void execute(BranchRegisterRequest request, BranchRegisterResponse response) throws TransactionException {
try {
doBranchRegister(request, response, rpcContext);
} catch (StoreException e) {
throw new TransactionException(TransactionExceptionCode.FailedStore, String.format("branch register request failed. xid=%s, msg=%s", request.getXid(), e.getMessage()), e);
}
}
}, request, response);
return response;
}
//Do branch register.
protected abstract void doBranchRegister(BranchRegisterRequest request, BranchRegisterResponse response, RpcContext rpcContext) throws TransactionException;
...
}
public class DefaultCoordinator extends AbstractTCInboundHandler implements TransactionMessageHandler, Disposable {
private final DefaultCore core;
...
@Override
protected void doBranchRegister(BranchRegisterRequest request, BranchRegisterResponse response, RpcContext rpcContext) throws TransactionException {
MDC.put(RootContext.MDC_KEY_XID, request.getXid());
//调用DefaultCore的branchRegister()方法处理分支事务注册请求
response.setBranchId(core.branchRegister(request.getBranchType(), request.getResourceId(), rpcContext.getClientId(), request.getXid(), request.getApplicationData(), request.getLockKey()));
}
...
}(3)
DefaultCore.branchRegister()的处理
DefaultCore的branchRegister()方法其实会继续调用其抽象父类AbstractCore的branchRegister()方法来处理注册分支事务请求，具体的过程如下：
一.根据xid获取全局事务会话
二.根据全局事务会话创建分支事务会话
三.通过MDC将分支事务ID存到线程本地变量副本
四.注册分支事务需要先获取全局锁
五.把分支事务会话加入到全局事务会话中并持久化
public class DefaultCore implements Core {
private static Map<BranchType, AbstractCore> coreMap = new ConcurrentHashMap<>();
public DefaultCore(RemotingServer remotingServer) {
List<AbstractCore> allCore = EnhancedServiceLoader.loadAll(AbstractCore.class, new Class[] {RemotingServer.class}, new Object[] {remotingServer});
if (CollectionUtils.isNotEmpty(allCore)) {
for (AbstractCore core : allCore) {
coreMap.put(core.getHandleBranchType(), core);
}
}
}
@Override
public Long branchRegister(BranchType branchType, String resourceId, String clientId, String xid, String applicationData, String lockKeys) throws TransactionException {
return getCore(branchType).branchRegister(branchType, resourceId, clientId, xid, applicationData, lockKeys);
}
public AbstractCore getCore(BranchType branchType) {
AbstractCore core = coreMap.get(branchType);
if (core == null) {
throw new NotSupportYetException("unsupported type:" + branchType.name());
}
return core;
}
...
}
public abstract class AbstractCore implements Core {
protected RemotingServer remotingServer;
public AbstractCore(RemotingServer remotingServer) {
if (remotingServer == null) {
throw new IllegalArgumentException("remotingServer must be not null");
}
this.remotingServer = remotingServer;
}
//注册分支事务
@Override
public Long branchRegister(BranchType branchType, String resourceId, String clientId, String xid, String applicationData, String lockKeys) throws TransactionException {
//1.根据xid获取全局事务会话GlobalSession
GlobalSession globalSession = assertGlobalSessionNotNull(xid, false);
return SessionHolder.lockAndExecute(globalSession, () -> {
globalSessionStatusCheck(globalSession);
globalSession.addSessionLifecycleListener(SessionHolder.getRootSessionManager());
//2.创建分支事务会话BranchSession，根据全局事务开启一个分支事务
//传入的参数依次是：全局事务会话、事务类型、资源ID、应用数据、全局锁keys、客户端ID
BranchSession branchSession = SessionHelper.newBranchByGlobal(globalSession, branchType, resourceId, applicationData, lockKeys, clientId);
//3.把分支事务的ID存放到线程本地变量副本中，也就是MDC中
MDC.put(RootContext.MDC_KEY_BRANCH_ID, String.valueOf(branchSession.getBranchId()));
//4.注册分支事务时会获取全局锁
//分支事务会话branchSession尝试获取一个全局锁，获取失败会抛异常，说明分支事务注册失败
branchSessionLock(globalSession, branchSession);
try {
//5.把分支事务会话加入到全局事务会话中
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();
});
}
private GlobalSession assertGlobalSessionNotNull(String xid, boolean withBranchSessions) throws TransactionException {
//根据xid寻找全局事务会话GlobalSession
GlobalSession globalSession = SessionHolder.findGlobalSession(xid, withBranchSessions);
if (globalSession == null) {
throw new GlobalTransactionException(TransactionExceptionCode.GlobalTransactionNotExist, String.format("Could not found global transaction xid = %s, may be has finished.", xid));
}
return globalSession;
}
//获取全局锁，获取全局锁失败则抛异常
protected void branchSessionLock(GlobalSession globalSession, BranchSession branchSession) throws TransactionException {
}
...
}
public class SessionHolder {
...
//根据xid获取全局事务会话GlobalSession
public static GlobalSession findGlobalSession(String xid, boolean withBranchSessions) {
return getRootSessionManager().findGlobalSession(xid, withBranchSessions);
}
...
}
@LoadLevel(name = "db", scope = Scope.PROTOTYPE)
public class DataBaseSessionManager extends AbstractSessionManager implements Initialize {
...
//根据xid获取全局事务会话GlobalSession
@Override
public GlobalSession findGlobalSession(String xid, boolean withBranchSessions) {
return transactionStoreManager.readSession(xid, withBranchSessions);
}
...
}
public class DataBaseTransactionStoreManager extends AbstractTransactionStoreManager implements TransactionStoreManager {
...
//根据xid获取全局事务会话GlobalSession
@Override
public GlobalSession readSession(String xid, boolean withBranchSessions) {
//global transaction
GlobalTransactionDO globalTransactionDO = logStore.queryGlobalTransactionDO(xid);
if (globalTransactionDO == null) {
return null;
}
//branch transactions
List<BranchTransactionDO> branchTransactionDOs = null;
//reduce rpc with db when branchRegister and getGlobalStatus
if (withBranchSessions) {
branchTransactionDOs = logStore.queryBranchTransactionDO(globalTransactionDO.getXid());
}
return getGlobalSession(globalTransactionDO, branchTransactionDOs);
}
...
}
public class SessionHelper {
...
//创建分支事务会话
public static BranchSession newBranchByGlobal(GlobalSession globalSession, BranchType branchType, String resourceId, String applicationData, String lockKeys, String clientId) {
BranchSession branchSession = new BranchSession();
branchSession.setXid(globalSession.getXid());
branchSession.setTransactionId(globalSession.getTransactionId());
branchSession.setBranchId(UUIDGenerator.generateUUID());
branchSession.setBranchType(branchType);
branchSession.setResourceId(resourceId);
branchSession.setLockKey(lockKeys);
branchSession.setClientId(clientId);
branchSession.setApplicationData(applicationData);
return branchSession;
}
...
}
public class GlobalSession implements SessionLifecycle, SessionStorable {
private List<BranchSession> branchSessions;
...
//把分支事务会话加入到全局事务会话中
@Override
public void addBranch(BranchSession branchSession) throws TransactionException {
for (SessionLifecycleListener lifecycleListener : lifecycleListeners) {
lifecycleListener.onAddBranch(this, branchSession);
}
branchSession.setStatus(BranchStatus.Registered);
add(branchSession);
}
//把分支事务会话加入到全局事务会话中
public boolean add(BranchSession branchSession) {
if (null != branchSessions) {
return branchSessions.add(branchSession);
} else {
//db and redis no need to deal with
return true;
}
}
...
}
public abstract class AbstractSessionManager implements SessionManager, SessionLifecycleListener {
...
@Override
public void onAddBranch(GlobalSession globalSession, BranchSession branchSession) throws TransactionException {
addBranchSession(globalSession, branchSession);
}
@Override
public void addBranchSession(GlobalSession session, BranchSession branchSession) throws TransactionException {
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("MANAGER[{}] SESSION[{}] {}", name, branchSession, LogOperation.BRANCH_ADD);
}
writeSession(LogOperation.BRANCH_ADD, branchSession);
}
//持久化全局事务会话
private void writeSession(LogOperation logOperation, SessionStorable sessionStorable) throws TransactionException {
//transactionStoreManager.writeSession()会对全局事务会话进行持久化
if (!transactionStoreManager.writeSession(logOperation, sessionStorable)) {
...
}
}
...
}12.将UndoLog写入到数据库与提交事务的源码
在数据源连接代理ConnectionProxy的
processGlobalTransactionCommit()方法中：
一.首先会注册完分支事务
二.然后会将UndoLog写入到数据库
三.最后才提交目标数据源连接的事务
//数据源连接代理
public class ConnectionProxy extends AbstractConnectionProxy {
private final LockRetryPolicy lockRetryPolicy = new LockRetryPolicy(this);
...
@Override
public void commit() throws SQLException {
try {
//通过全局锁重试策略组件LockRetryPolicy来执行本地事务的提交
lockRetryPolicy.execute(() -> {
doCommit();
return null;
});
} catch (SQLException e) {
if (targetConnection != null && !getAutoCommit() && !getContext().isAutoCommitChanged()) {
rollback();
}
throw e;
} catch (Exception e) {
throw new SQLException(e);
}
}
private void doCommit() throws SQLException {
if (context.inGlobalTransaction()) {
processGlobalTransactionCommit();
} else if (context.isGlobalLockRequire()) {
processLocalCommitWithGlobalLocks();
} else {
targetConnection.commit();
}
}
private void processLocalCommitWithGlobalLocks() throws SQLException {
//检查全局锁keys
checkLock(context.buildLockKeys());
try {
//目标数据源连接提交事务
targetConnection.commit();
} catch (Throwable ex) {
throw new SQLException(ex);
}
context.reset();
}
private void processGlobalTransactionCommit() throws SQLException {
try {
//1.注册分支事务
register();
} catch (TransactionException e) {
recognizeLockKeyConflictException(e, context.buildLockKeys());
}
try {
//2.将UndoLog写入到数据库
UndoLogManagerFactory.getUndoLogManager(this.getDbType()).flushUndoLogs(this);
//3.目标数据源连接提交事务
targetConnection.commit();
} catch (Throwable ex) {
LOGGER.error("process connectionProxy commit error: {}", ex.getMessage(), ex);
report(false);
throw new SQLException(ex);
}
if (IS_REPORT_SUCCESS_ENABLE) {
report(true);
}
context.reset();
}
...
}
public class UndoLogManagerFactory {
private static final Map<String, UndoLogManager> UNDO_LOG_MANAGER_MAP = new ConcurrentHashMap<>();
//获取UndoLog管理器
public static UndoLogManager getUndoLogManager(String dbType) {
return CollectionUtils.computeIfAbsent(UNDO_LOG_MANAGER_MAP, dbType,
key -> EnhancedServiceLoader.load(UndoLogManager.class, dbType));
}
}
public abstract class AbstractUndoLogManager implements UndoLogManager {
...
@Override
public void flushUndoLogs(ConnectionProxy cp) throws SQLException {
ConnectionContext connectionContext = cp.getContext();
if (!connectionContext.hasUndoLog()) {
return;
}
String xid = connectionContext.getXid();
long branchId = connectionContext.getBranchId();
BranchUndoLog branchUndoLog = new BranchUndoLog();
branchUndoLog.setXid(xid);
branchUndoLog.setBranchId(branchId);
branchUndoLog.setSqlUndoLogs(connectionContext.getUndoItems());
UndoLogParser parser = UndoLogParserFactory.getInstance();
byte[] undoLogContent = parser.encode(branchUndoLog);
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("Flushing UNDO LOG: {}", new String(undoLogContent, Constants.DEFAULT_CHARSET));
}
CompressorType compressorType = CompressorType.NONE;
if (needCompress(undoLogContent)) {
compressorType = ROLLBACK_INFO_COMPRESS_TYPE;
undoLogContent = CompressorFactory.getCompressor(compressorType.getCode()).compress(undoLogContent);
}
//插入UndoLog到数据库
insertUndoLogWithNormal(xid, branchId, buildContext(parser.getName(), compressorType), undoLogContent, cp.getTargetConnection());
}
//insert uodo log when normal
protected abstract void insertUndoLogWithNormal(String xid, long branchId, String rollbackCtx, byte[] undoLogContent, Connection conn) throws SQLException;
...
}
@LoadLevel(name = JdbcConstants.MYSQL)
public class MySQLUndoLogManager extends AbstractUndoLogManager {
...
@Override
protected void insertUndoLogWithNormal(String xid, long branchId, String rollbackCtx, byte[] undoLogContent, Connection conn) throws SQLException {
insertUndoLog(xid, branchId, rollbackCtx, undoLogContent, State.Normal, conn);
}
private void insertUndoLog(String xid, long branchId, String rollbackCtx, byte[] undoLogContent, State state, Connection conn) throws SQLException {
try (PreparedStatement pst = conn.prepareStatement(INSERT_UNDO_LOG_SQL)) {
pst.setLong(1, branchId);
pst.setString(2, xid);
pst.setString(3, rollbackCtx);
pst.setBytes(4, undoLogContent);
pst.setInt(5, state.getValue());
pst.executeUpdate();
} catch (Exception e) {
if (!(e instanceof SQLException)) {
e = new SQLException(e);
}
throw (SQLException) e;
}
}
...
}13.通过全局锁重试策略组件执行事务的提交
当设置完禁止自动提交事务、构建前镜像、执行SQL、构建后镜像，执行到数据源连接代理ConnectionProxy的commit()方法提交本地事务时，便会通过全局锁重试策略LockRetryPolicy来执行本地事务的提交。
全局锁重试策略LockRetryPolicy，会确保先获取到全局锁才提交本地事务。也就是如果获取不到全局锁，则重试获取。此外，注册分支事务时，获取到全局锁才能注册成功。
public class ConnectionProxy extends AbstractConnectionProxy {
private final LockRetryPolicy lockRetryPolicy = new LockRetryPolicy(this);
...
@Override
public void commit() throws SQLException {
try {
//通过全局锁重试策略组件LockRetryPolicy来执行本地事务的提交
lockRetryPolicy.execute(() -> {
doCommit();
return null;
});
} catch (SQLException e) {
if (targetConnection != null && !getAutoCommit() && !getContext().isAutoCommitChanged()) {
rollback();
}
throw e;
} catch (Exception e) {
throw new SQLException(e);
}
}
...
public static class LockRetryPolicy {
protected static final boolean LOCK_RETRY_POLICY_BRANCH_ROLLBACK_ON_CONFLICT = ConfigurationFactory.getInstance().
getBoolean(ConfigurationKeys.CLIENT_LOCK_RETRY_POLICY_BRANCH_ROLLBACK_ON_CONFLICT, DEFAULT_CLIENT_LOCK_RETRY_POLICY_BRANCH_ROLLBACK_ON_CONFLICT);
protected final ConnectionProxy connection;
public LockRetryPolicy(ConnectionProxy connection) {
this.connection = connection;
}
public <T> T execute(Callable<T> callable) throws Exception {
//the only case that not need to retry acquire lock hear is
//LOCK_RETRY_POLICY_BRANCH_ROLLBACK_ON_CONFLICT == true && connection#autoCommit == true
//because it has retry acquire lock when AbstractDMLBaseExecutor#executeAutoCommitTrue
if (LOCK_RETRY_POLICY_BRANCH_ROLLBACK_ON_CONFLICT && connection.getContext().isAutoCommitChanged()) {
//不需要重试
return callable.call();
} else {
//LOCK_RETRY_POLICY_BRANCH_ROLLBACK_ON_CONFLICT == false
//or LOCK_RETRY_POLICY_BRANCH_ROLLBACK_ON_CONFLICT == true && autoCommit == false
return doRetryOnLockConflict(callable);
}
}
protected <T> T doRetryOnLockConflict(Callable<T> callable) throws Exception {
LockRetryController lockRetryController = new LockRetryController();
while (true) {
try {
return callable.call();
} catch (LockConflictException lockConflict) {
onException(lockConflict);
//AbstractDMLBaseExecutor#executeAutoCommitTrue the local lock is released
if (connection.getContext().isAutoCommitChanged() && lockConflict.getCode() == TransactionExceptionCode.LockKeyConflictFailFast) {
lockConflict.setCode(TransactionExceptionCode.LockKeyConflict);
}
//休眠一会再去重试
lockRetryController.sleep(lockConflict);
} catch (Exception e) {
onException(e);
throw e;
}
}
}
//Callback on exception in doLockRetryOnConflict.
protected void onException(Exception e) throws Exception {
}
}
...
}
public abstract class AbstractDMLBaseExecutor<T, S extends Statement> extends BaseTransactionalExecutor<T, S> {
...
private static class LockRetryPolicy extends ConnectionProxy.LockRetryPolicy {
LockRetryPolicy(final ConnectionProxy connection) {
super(connection);
}
@Override
public <T> T execute(Callable<T> callable) throws Exception {
if (LOCK_RETRY_POLICY_BRANCH_ROLLBACK_ON_CONFLICT) {
return doRetryOnLockConflict(callable);
} else {
return callable.call();
}
}
@Override
protected void onException(Exception e) throws Exception {
ConnectionContext context = connection.getContext();
//UndoItems can't use the Set collection class to prevent ABA
context.removeSavepoint(null);
//回滚目标数据源连接对SQL的执行
connection.getTargetConnection().rollback();
}
public static boolean isLockRetryPolicyBranchRollbackOnConflict() {
return LOCK_RETRY_POLICY_BRANCH_ROLLBACK_ON_CONFLICT;
}
}
...
}14.注册分支事务时获取全局锁的入口源码
在Seata Server中，只有当全局锁获取成功后，分支事务才能注册成功。AbstractCore的branchRegister()方法会通过调用ATCore的branchSessionLock()方法来获取全局锁，而ATCore的branchSessionLock()方法最终则是靠调用AbstractLockManager的acquireLock()方法来尝试获取全局锁的。获取全局锁失败会抛出异常，说明注册分支事务失败。
public abstract class AbstractCore implements Core {
...
//注册分支事务
@Override
public Long branchRegister(BranchType branchType, String resourceId, String clientId, String xid, String applicationData, String lockKeys) throws TransactionException {
//1.根据xid获取全局事务会话GlobalSession
GlobalSession globalSession = assertGlobalSessionNotNull(xid, false);
return SessionHolder.lockAndExecute(globalSession, () -> {
globalSessionStatusCheck(globalSession);
globalSession.addSessionLifecycleListener(SessionHolder.getRootSessionManager());
//2.创建分支事务会话，根据全局事务开启一个分支事务
//传入的参数依次是：全局事务会话、事务类型、资源ID、应用数据、全局锁keys、客户端ID
BranchSession branchSession = SessionHelper.newBranchByGlobal(globalSession, branchType, resourceId, applicationData, lockKeys, clientId);
//3.把分支事务的ID存放到线程本地变量副本中，也就是MDC中
MDC.put(RootContext.MDC_KEY_BRANCH_ID, String.valueOf(branchSession.getBranchId()));
//4.注册分支事务时会加全局锁
//分支事务会话branchSession尝试获取一个全局锁，获取失败会抛异常，说明分支事务注册失败
branchSessionLock(globalSession, branchSession);
try {
//5.把分支事务会话加入到全局事务会话中
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();
});
}
...
}
public class ATCore extends AbstractCore {
...
@Override
protected void branchSessionLock(GlobalSession globalSession, BranchSession branchSession) throws TransactionException {
//从应用程序数据里提取出一些属性进行属性赋值
String applicationData = branchSession.getApplicationData();
boolean autoCommit = true;
boolean skipCheckLock = false;
if (StringUtils.isNotBlank(applicationData)) {
if (objectMapper == null) {
objectMapper = new ObjectMapper();
}
try {
//ObjectMapper是一个对象映射框架，它可以把ApplicationData对象里的属性值读取出来，然后写入到HashMap里
Map<String, Object> data = objectMapper.readValue(applicationData, HashMap.class);
Object clientAutoCommit = data.get(AUTO_COMMIT);
if (clientAutoCommit != null && !(boolean)clientAutoCommit) {
autoCommit = (boolean)clientAutoCommit;
}
Object clientSkipCheckLock = data.get(SKIP_CHECK_LOCK);
if (clientSkipCheckLock instanceof Boolean) {
skipCheckLock = (boolean)clientSkipCheckLock;
}
} catch (IOException e) {
LOGGER.error("failed to get application data: {}", e.getMessage(), e);
}
}
try {
//分支事务会话branchSession尝试获取一个全局锁，获取失败会抛异常，说明分支事务注册失败
if (!branchSession.lock(autoCommit, skipCheckLock)) {
throw new BranchTransactionException(LockKeyConflict, String.format("Global lock acquire failed xid = %s branchId = %s", globalSession.getXid(), branchSession.getBranchId()));
}
} catch (StoreException e) {
if (e.getCause() instanceof BranchTransactionException) {
throw new BranchTransactionException(((BranchTransactionException)e.getCause()).getCode(), String.format("Global lock acquire failed xid = %s branchId = %s", globalSession.getXid(), branchSession.getBranchId()));
}
throw e;
}
}
...
}
public class BranchSession implements Lockable, Comparable<BranchSession>, SessionStorable {
...
public boolean lock(boolean autoCommit, boolean skipCheckLock) throws TransactionException {
if (this.getBranchType().equals(BranchType.AT)) {
//尝试获取全局锁
return LockerManagerFactory.getLockManager().acquireLock(this, autoCommit, skipCheckLock);
}
return true;
}
...
}
public class LockerManagerFactory {
private static final Configuration CONFIG = ConfigurationFactory.getInstance();
private static volatile LockManager LOCK_MANAGER;
public static LockManager getLockManager() {
if (LOCK_MANAGER == null) {
init();
}
return LOCK_MANAGER;
}
public static void init() {
init(null);
}
public static void init(String lockMode) {
if (LOCK_MANAGER == null) {
synchronized (LockerManagerFactory.class) {
if (LOCK_MANAGER == null) {
if (StringUtils.isBlank(lockMode)) {
lockMode = CONFIG.getConfig(ConfigurationKeys.STORE_LOCK_MODE, CONFIG.getConfig(ConfigurationKeys.STORE_MODE, SERVER_DEFAULT_STORE_MODE));
}
if (StoreMode.contains(lockMode)) {
LOCK_MANAGER = EnhancedServiceLoader.load(LockManager.class, lockMode);
}
}
}
}
}
}
public abstract class AbstractLockManager implements LockManager {
...
@Override
public boolean acquireLock(BranchSession branchSession, boolean autoCommit, boolean skipCheckLock) throws TransactionException {
if (branchSession == null) {
throw new IllegalArgumentException("branchSession can't be null for memory/file locker.");
}
String lockKey = branchSession.getLockKey();
if (StringUtils.isNullOrEmpty(lockKey)) {
//no lock
return true;
}
//get locks of branch
//获取到分支事务里需要的所有行锁
List<RowLock> locks = collectRowLocks(branchSession);
if (CollectionUtils.isEmpty(locks)) {
//no lock
return true;
}
//具体进行获取锁
return getLocker(branchSession).acquireLock(locks, autoCommit, skipCheckLock);
}
@Override
public List<RowLock> collectRowLocks(BranchSession branchSession) {
if (branchSession == null || StringUtils.isBlank(branchSession.getLockKey())) {
return Collections.emptyList();
}
String lockKey = branchSession.getLockKey();
String resourceId = branchSession.getResourceId();
String xid = branchSession.getXid();
long transactionId = branchSession.getTransactionId();
long branchId = branchSession.getBranchId();
return collectRowLocks(lockKey, resourceId, xid, transactionId, branchId);
}
protected List<RowLock> collectRowLocks(String lockKey, String resourceId, String xid, Long transactionId, Long branchID) {
List<RowLock> locks = new ArrayList<>();
String[] tableGroupedLockKeys = lockKey.split(";");
for (String tableGroupedLockKey : tableGroupedLockKeys) {
int idx = tableGroupedLockKey.indexOf(":");
if (idx < 0) {
return locks;
}
String tableName = tableGroupedLockKey.substring(0, idx);
String mergedPKs = tableGroupedLockKey.substring(idx + 1);
if (StringUtils.isBlank(mergedPKs)) {
return locks;
}
String[] pks = mergedPKs.split(",");
if (pks == null || pks.length == 0) {
return locks;
}
for (String pk : pks) {
if (StringUtils.isNotBlank(pk)) {
RowLock rowLock = new RowLock();
rowLock.setXid(xid);
rowLock.setTransactionId(transactionId);
rowLock.setBranchId(branchID);
rowLock.setTableName(tableName);
rowLock.setPk(pk);
rowLock.setResourceId(resourceId);
locks.add(rowLock);
}
}
}
return locks;
}
...
}
public class RowLock {
private String xid;//全局事务xid
private Long transactionId;//全局事务ID
private Long branchId;//分支事务ID
private String resourceId;//资源ID
private String tableName;//表名称
private String pk;//主键
private String rowKey;//行键
private String feature;//功能特性
...
}15.Seata Server获取全局锁的具体逻辑源码
调用AbstractLockManager的acquireLock()方法获取全局锁时，其实调用的是DataBaseLocker的acquireLock()方法 -> LockStoreDataBaseDAO的acquireLock()方法。
在LockStoreDataBaseDAO的acquireLock()方法中，首先会查询数据库中是否存在要申请的全局锁的记录，然后根据这些锁记录 + xid判断是否由当前全局事务获取的(这是核心)。
如果不是，则说明其他全局事务先获取到了要申请的全局锁，此时当前事务获取全局锁失败。
如果是，则把当前事务已经获取过的全局锁过滤出来，然后尝试写入当前分支事务还需获取的全局锁记录到数据库。如果写入成功，则表示当前分支事务成功获取到全局锁。如果写入失败，则表示其他分支事务已经获取到全局锁。
@LoadLevel(name = "db")
public class DataBaseLockManager extends AbstractLockManager implements Initialize {
private Locker locker;
@Override
public void init() {
//init dataSource
String datasourceType = ConfigurationFactory.getInstance().getConfig(ConfigurationKeys.STORE_DB_DATASOURCE_TYPE);
DataSource lockStoreDataSource = EnhancedServiceLoader.load(DataSourceProvider.class, datasourceType).provide();
locker = new DataBaseLocker(lockStoreDataSource);
}
@Override
public Locker getLocker(BranchSession branchSession) {
return locker;
}
...
}
public class DataBaseLocker extends AbstractLocker {
private LockStore lockStore;
public DataBaseLocker(DataSource logStoreDataSource) {
lockStore = new LockStoreDataBaseDAO(logStoreDataSource);
}
...
@Override
public boolean acquireLock(List<RowLock> locks, boolean autoCommit, boolean skipCheckLock) {
if (CollectionUtils.isEmpty(locks)) {
//no lock
return true;
}
try {
//通过执行MySQL来获取全局锁
return lockStore.acquireLock(convertToLockDO(locks), autoCommit, skipCheckLock);
} catch (StoreException e) {
throw e;
} catch (Exception t) {
LOGGER.error("AcquireLock error, locks:{}", CollectionUtils.toString(locks), t);
return false;
}
}
...
}
public class LockStoreDataBaseDAO implements LockStore {
...
@Override
public boolean acquireLock(List<LockDO> lockDOs, boolean autoCommit, boolean skipCheckLock) {
//数据库操作三剑客：连接、句柄、结果
Connection conn = null;
PreparedStatement ps = null;
ResultSet rs = null;
Set<String> dbExistedRowKeys = new HashSet<>();
boolean originalAutoCommit = true;
if (lockDOs.size() > 1) {
lockDOs = lockDOs.stream().filter(LambdaUtils.distinctByKey(LockDO::getRowKey)).collect(Collectors.toList());
}
try {
//从全局锁数据源里获取到一个连接
conn = lockStoreDataSource.getConnection();
//关闭自动提交事务
if (originalAutoCommit = conn.getAutoCommit()) {
conn.setAutoCommit(false);
}
//需要获取的锁，有可能多个
List<LockDO> unrepeatedLockDOs = lockDOs;
//check lock
if (!skipCheckLock) {
boolean canLock = true;
//query，针对全局锁表查询某个数据加了全局锁的全局事务xid
//LockStoreSqlFactory是全局锁存储的SQL工厂
String checkLockSQL = LockStoreSqlFactory.getLogStoreSql(dbType).getCheckLockableSql(lockTable, lockDOs.size());
ps = conn.prepareStatement(checkLockSQL);
for (int i = 0; i < lockDOs.size(); i++) {
ps.setString(i + 1, lockDOs.get(i).getRowKey());
}
//执行查询
rs = ps.executeQuery();
//获取到当前要加全局锁的事务xid
String currentXID = lockDOs.get(0).getXid();
boolean failFast = false;
//如果查询到的结果rs是空，则表示当前全局锁没有被事务获取占用
while (rs.next()) {
String dbXID = rs.getString(ServerTableColumnsName.LOCK_TABLE_XID);
//如果获取到全局锁的是别的全局事务xid，那么获取全局锁失败，设置canLock为false
if (!StringUtils.equals(dbXID, currentXID)) {
if (LOGGER.isInfoEnabled()) {
String dbPk = rs.getString(ServerTableColumnsName.LOCK_TABLE_PK);
String dbTableName = rs.getString(ServerTableColumnsName.LOCK_TABLE_TABLE_NAME);
long dbBranchId = rs.getLong(ServerTableColumnsName.LOCK_TABLE_BRANCH_ID);
LOGGER.info("Global lock on [{}:{}] is holding by xid {} branchId {}", dbTableName, dbPk, dbXID, dbBranchId);
}
if (!autoCommit) {
int status = rs.getInt(ServerTableColumnsName.LOCK_TABLE_STATUS);
if (status == LockStatus.Rollbacking.getCode()) {
failFast = true;
}
}
canLock = false;
break;
}
dbExistedRowKeys.add(rs.getString(ServerTableColumnsName.LOCK_TABLE_ROW_KEY));
}
if (!canLock) {
conn.rollback();
if (failFast) {
throw new StoreException(new BranchTransactionException(LockKeyConflictFailFast));
}
return false;
}
//If the lock has been exists in db, remove it from the lockDOs
if (CollectionUtils.isNotEmpty(dbExistedRowKeys)) {
//过滤当前事务已经获取过的全局锁
unrepeatedLockDOs = lockDOs.stream().filter(lockDO -> !dbExistedRowKeys.contains(lockDO.getRowKey())).collect(Collectors.toList());
}
if (CollectionUtils.isEmpty(unrepeatedLockDOs)) {
conn.rollback();
return true;
}
}
//lock
if (unrepeatedLockDOs.size() == 1) {
LockDO lockDO = unrepeatedLockDOs.get(0);
//尝试加锁，表示全局锁被当前的分支事务获取了
if (!doAcquireLock(conn, lockDO)) {
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Global lock acquire failed, xid {} branchId {} pk {}", lockDO.getXid(), lockDO.getBranchId(), lockDO.getPk());
}
conn.rollback();
return false;
}
} else {
if (!doAcquireLocks(conn, unrepeatedLockDOs)) {
if (LOGGER.isInfoEnabled()) {
LOGGER.info("Global lock batch acquire failed, xid {} branchId {} pks {}", unrepeatedLockDOs.get(0).getXid(), unrepeatedLockDOs.get(0).getBranchId(), unrepeatedLockDOs.stream().map(lockDO -> lockDO.getPk()).collect(Collectors.toList()));
}
conn.rollback();
return false;
}
}
conn.commit();
return true;
} catch (SQLException e) {
throw new StoreException(e);
} finally {
IOUtil.close(rs, ps);
if (conn != null) {
try {
if (originalAutoCommit) {
conn.setAutoCommit(true);
}
conn.close();
} catch (SQLException e) {
}
}
}
}
protected boolean doAcquireLock(Connection conn, LockDO lockDO) {
PreparedStatement ps = null;
try {
//insert
String insertLockSQL = LockStoreSqlFactory.getLogStoreSql(dbType).getInsertLockSQL(lockTable);
ps = conn.prepareStatement(insertLockSQL);
ps.setString(1, lockDO.getXid());//全局事务xid
ps.setLong(2, lockDO.getTransactionId());//全局事务id
ps.setLong(3, lockDO.getBranchId());//分支事务id
ps.setString(4, lockDO.getResourceId());//资源id
ps.setString(5, lockDO.getTableName());//表名称
ps.setString(6, lockDO.getPk());//主键
ps.setString(7, lockDO.getRowKey());//rowkey
ps.setInt(8, LockStatus.Locked.getCode());//locked
return ps.executeUpdate() > 0;
} catch (SQLException e) {
throw new StoreException(e);
} finally {
IOUtil.close(ps);
}
}
...
}相关推荐
- Java 如何从一个 List 中随机获得元素
-
概述从一个List中随机获得一个元素是有关List的一个基本操作，但是这个操作又没有非常明显的实现。本页面主要向你展示如何有效的从List中获得一个随机的元素和可以使用的一些方法。选择一个...
- 想月薪过万吗?计算机安卓开发之"集合"
-
集合的总结:/***Collection*List(存取有序,有索引,可以重复)*ArrayList*底层是数组实现的,线程不安全,查找和修改快,增和删比较慢*LinkedList*底层是...
- China Narrows AI Talent Gap With U.S. as Research Enters Engineering Phase: Report
-
ImagegeneratedbyAITMTPOST--ChinaisclosinginontheU.S.intheAIindustry-academia-research...
- 大促系统优化之应用启动速度优化实践
-
作者：京东零售宋维飞一、前言本文记录了在大促前针对SpringBoot应用启动速度过慢而采取的优化方案，主要介绍了如何定位启动速度慢的阻塞点，以及如何解决这些问题。希望可以帮助大家了解如何定位该类问...
- MyEMS开源能源管理系统核心代码解读004
-
本期解读：计量表能耗数据规范化算法：myems/myems-normalization/meter.py代码见底部这段代码是一个用于计算和存储能源计量数据（如电表读数）的小时值的Python脚本。它主...
- Java接口与抽象类:核心区别、使用场景与最佳实践
-
Java接口与抽象类：核心区别、使用场景与最佳实践一、核心特性对比1.语法定义接口：interface关键字定义，支持extends多继承接口javapublicinterfaceDrawabl...
- 超好看 vue2.x 音频播放器组件Vue-APlayer
-
上篇文章给大家分享了视频播放器组件vue-aliplayer，这次给大家推荐一款音频插件VueAplayer。vue-aplayer一个好看又好用的轻量级vue.js音乐播放器组件。清爽漂亮的U...
- Linq 下的扩展方法太少了，MoreLinq 来啦
-
一：背景1.讲故事前几天看同事在用linq给内存中的两个model做左连接，用过的朋友都知道，你一定少不了一个叫做DefaultIfEmpty函数，这玩意吧，本来很流畅的from......
- MapReduce过程详解及其性能优化(详细)
-
从JVM的角度看Map和ReduceMap阶段包括：第一读数据：从HDFS读取数据1、问题:读取数据产生多少个Mapper？？Mapper数据过大的话，会产生大量的小文件，由于Mapper是基于虚拟...
- 手把手教你使用scrapy框架来爬取北京新发地价格行情(实战篇)
-
来源：Python爬虫与数据挖掘作者：霖hero前言关于Scrapy理论的知识，可以参考我的上一篇文章，这里不再赘述，直接上干货。实战演练爬取分析首先我们进入北京新发地价格行情网页并打开开发者工具，如...
- 屏蔽疯狂蜘蛛，防止CPU占用100%（mumu模拟器和雷电模拟器哪个更占用cpu）
-
站点总是某个时间段莫名的cpu100%，资源占用也不高，这就有必要怀疑爬虫问题。1.使用"robots.txt"规范在网站根目录新建空白文件，命名为"robots.txt&#...
- Web黑客近年神作Gospider:一款基于Go语言开发的Web爬虫，要收藏
-
小白看黑客技术文章，一定要点首小歌放松心情哈，我最爱盆栽！开始装逼！Gospider是一款运行速度非常快的Web爬虫程序，对于爱好白帽黑客的小白来说，可谓是佳作！Gospider采用厉害的Go语言开发...
- 用宝塔面板免费防火墙屏蔽织梦扫描网站
-
今天教大家在免费的基础上屏蔽织梦扫描，首先您要安装宝塔面板，然后再安装免费的防火墙插件，我用的是Nginx免费防火墙，然后打开这个插件。设置GET-URL过滤设置一条简单的宝塔面板的正则规则就可以屏蔽...
- 蜘蛛人再捞4千万美元 连续三周蝉联北美票房冠军
-
7月15日讯老马追踪票房数据的北美院线联盟今天表示，“蜘蛛人：离家日”（Spider-Man:FarFromHome）击退两部新片的挑战，连续第2周勇夺北美票房冠军，海捞4530万美元。法新...
- 夏天到了，需要提防扁虱，真是又小又恐怖的动物
-
夏天马上要到了，你知道吗，扁虱是这个夏天最危险的动物之一，很少有动物能比它还凶猛。Whenitcomestosummer'slittledangers,fewarenastiert...
- 一周热门
- 最近发表
-
- Java 如何从一个 List 中随机获得元素
- 想月薪过万吗?计算机安卓开发之"集合"
- China Narrows AI Talent Gap With U.S. as Research Enters Engineering Phase: Report
- 大促系统优化之应用启动速度优化实践
- MyEMS开源能源管理系统核心代码解读004
- Java接口与抽象类:核心区别、使用场景与最佳实践
- 超好看 vue2.x 音频播放器组件Vue-APlayer
- Linq 下的扩展方法太少了，MoreLinq 来啦
- MapReduce过程详解及其性能优化(详细)
- 手把手教你使用scrapy框架来爬取北京新发地价格行情(实战篇)
- 标签列表
-
- ps图案在哪里 (33)
- super().__init__ (33)
- python 获取日期 (34)
- 0xa (36)
- super().__init__()详解 (33)
- python安装包在哪里找 (33)
- linux查看python版本信息 (35)
- python怎么改成中文 (35)
- php文件怎么在浏览器运行 (33)
- eval在python中的意思 (33)
- python安装opencv库 (35)
- python div (34)
- sticky css (33)
- python中random.randint()函数 (34)
- python去掉字符串中的指定字符 (33)
- python入门经典100题 (34)
- anaconda安装路径 (34)
- yield和return的区别 (33)
- 1到10的阶乘之和是多少 (35)
- python安装sklearn库 (33)
- dom和bom区别 (33)
- js 替换指定位置的字符 (33)
- python判断元素是否存在 (33)
- sorted key (33)
- shutil.copy() (33)