Seata源码分析之Session_综合

一、GlobalSession

GlobalSession是seata协调器DefaultCoordinator管理维护的重要部件，当用户开启全球分布式事务，TM调用begin方法请求至TC，TC则创建GlobalSession实例对象，返回唯一的xid。它实现SessionLifecycle接口，提供begin，changeStatus，changeBranchStatus，addBranch，removeBranch等操作session和branchSession的方法。

public class GlobalSession implements SessionLifecycle, SessionStorable {private String xid;private long transactionId;private volatile GlobalStatus status;private String applicationId;private String transactionServiceGroup;private String transactionName;private int timeout;private long beginTime;private String applicationData;private boolean active = true;private final ArrayList<BranchSession> branchSessions = new ArrayList<>();// 添加branchSessionpublic boolean add(BranchSession branchSession) {return branchSessions.add(branchSession);}// 删除branchSessionpublic boolean remove(BranchSession branchSession) {return branchSessions.remove(branchSession);}// 是否超时public boolean isTimeout() {return (System.currentTimeMillis() - beginTime) > timeout;}@Override// session开启，设置状态为Beginpublic void begin() throws TransactionException {this.status = GlobalStatus.Begin;this.beginTime = System.currentTimeMillis();this.active = true;for (SessionLifecycleListener lifecycleListener : lifecycleListeners) {lifecycleListener.onBegin(this);}}// 修改seseion状态public void changeStatus(GlobalStatus status) throws TransactionException {this.status = status;for (SessionLifecycleListener lifecycleListener : lifecycleListeners) {lifecycleListener.onStatusChange(this, status);}}// 修改branchSession状态public void changeBranchStatus(BranchSession branchSession, BranchStatus status)throws TransactionException {branchSession.setStatus(status);for (SessionLifecycleListener lifecycleListener : lifecycleListeners) {lifecycleListener.onBranchStatusChange(this, branchSession, status);}}@Override// 添加branchSessionpublic void addBranch(BranchSession branchSession) throws TransactionException {for (SessionLifecycleListener lifecycleListener : lifecycleListeners) {lifecycleListener.onAddBranch(this, branchSession);}branchSession.setStatus(BranchStatus.Registered);add(branchSession);}@Override// 删除branchSessionpublic void removeBranch(BranchSession branchSession) throws TransactionException {for (SessionLifecycleListener lifecycleListener : lifecycleListeners) {lifecycleListener.onRemoveBranch(this, branchSession);}branchSession.unlock();remove(branchSession);}
}

二、GlobalSessionLock

GlobalSessionLock是GlobalSession中的内部类，它内部持有ReentrantLock对象，利用ReentrantLock的lock和unlock机制。GlobalSession中默认持有一个globalSessionLock对象，并提供lockAndExcute在GlobalSession中的线程安全操作的模板方法。

public class GlobalSession implements SessionLifecycle, SessionStorable {private GlobalSessionLock globalSessionLock = new GlobalSessionLock();public void lock() throws TransactionException {globalSessionLock.lock();}public void unlock() {globalSessionLock.unlock();}public void lockAndExcute(LockRunnable excuteRunnable) throws TransactionException {this.lock();try {excuteRunnable.run();} finally {this.unlock();}}public <T> T lockAndExcute(LockCallable<T> lockCallable) throws TransactionException {this.lock();try {return lockCallable.call();} finally {this.unlock();}}private class GlobalSessionLock {private Lock globalSessionLock = new ReentrantLock();private static final int GLOBAL_SESSION_LOCK_TIME_OUT_MILLS = 2 * 1000;public void lock() throws TransactionException {try {if (globalSessionLock.tryLock(GLOBAL_SESSION_LOCK_TIME_OUT_MILLS, TimeUnit.MILLISECONDS)) {return;}} catch (InterruptedException e) {LOGGER.error("Interrupted error", e);}throw new TransactionException(TransactionExceptionCode.FailedLockGlobalTranscation);}public void unlock() {globalSessionLock.unlock();}}@FunctionalInterfacepublic interface LockRunnable {void run() throws TransactionException;}@FunctionalInterfacepublic interface LockCallable<V> {V call() throws TransactionException;}
}

三、BranchSession

BranchSession为分支session，管理分支数据，受globalSession统一调度管理，它的lock和unlock方法由lockManger实现。

public class BranchSession implements Lockable, Comparable<BranchSession>, SessionStorable {private String xid;private long transactionId;private long branchId;private String resourceGroupId;private String resourceId;private String lockKey;private BranchType branchType;private BranchStatus status = BranchStatus.Unknown;private String clientId;private String applicationData;@Overridepublic boolean lock() throws TransactionException {return LockerFactory.getLockManager().acquireLock(this);}@Overridepublic boolean unlock() throws TransactionException {return LockerFactory.getLockManager().releaseLock(this);}
}

四、DefaultLockManager

DefaultLockManager是LockManager的默认实现，它获取branchSession的lockKey，转换成List<RowLock>，委派Locker进行处理。

public class DefaultLockManager extends AbstractLockManager {private static Locker locker = null;@Overridepublic boolean acquireLock(BranchSession branchSession) throws TransactionException {String lockKey = branchSession.getLockKey();if (StringUtils.isNullOrEmpty(lockKey)) {//no lockreturn true;}//get locks of branchList<RowLock> locks = collectRowLocks(branchSession);if (CollectionUtils.isEmpty(locks)) {//no lockreturn true;}return getLocker(branchSession).acquireLock(locks);}@Overridepublic boolean releaseLock(BranchSession branchSession) throws TransactionException {List<RowLock> locks = collectRowLocks(branchSession);if (CollectionUtils.isEmpty(locks)) {//no lockreturn true;}try {return getLocker(branchSession).releaseLock(locks);} catch (Exception t) {LOGGER.error("unLock error, branchSession:" + branchSession, t);return false;}}@Overridepublic boolean isLockable(String xid, String resourceId, String lockKey) throws TransactionException {List<RowLock> locks = collectRowLocks(lockKey, resourceId, xid);try {return getLocker().isLockable(locks);} catch (Exception t) {LOGGER.error("isLockable error, xid:" + xid + ", resourceId:" + resourceId + ", lockKey:" + lockKey, t);return false;}}@Overridepublic void cleanAllLocks() throws TransactionException {getLocker().cleanAllLocks();}protected Locker getLocker() {return getLocker(null);}protected Locker getLocker(BranchSession branchSession) {return LockerFactory.get(branchSession);}
}

五、Locker

Locker接口提供根据行数据获取锁，释放锁，是否锁住和清除所有锁的方法。有DataBaseLocker和MemoryLocker两种实现方式，可以根据用户配置db或file进行不同的加载。

public interface Locker {boolean acquireLock(List<RowLock> rowLock) ;boolean releaseLock(List<RowLock> rowLock);boolean isLockable(List<RowLock> rowLock);void cleanAllLocks();
}

六、MemoryLocker

MemoryLocker即是使用seata服务器的内存map进行缓存实现锁功能。

public class MemoryLocker extends AbstractLocker {private static final int BUCKET_PER_TABLE = 128;private static final ConcurrentHashMap<String/* resourceId */,ConcurrentHashMap<String/* tableName */,ConcurrentHashMap<Integer/* bucketId */,Map<String/* pk */, Long/* transactionId */>>>>LOCK_MAP= new ConcurrentHashMap<String, ConcurrentHashMap<String, ConcurrentHashMap<Integer, Map<String, Long>>>>();protected BranchSession branchSession = null;public MemoryLocker(BranchSession branchSession) {this.branchSession = branchSession;}@Overridepublic boolean acquireLock(List<RowLock> rowLocks) {if (CollectionUtils.isEmpty(rowLocks)) {//no lockreturn true;}String resourceId = branchSession.getResourceId();long transactionId = branchSession.getTransactionId();ConcurrentHashMap<Map<String, Long>, Set<String>> bucketHolder = branchSession.getLockHolder();ConcurrentHashMap<String, ConcurrentHashMap<Integer, Map<String, Long>>> dbLockMap = LOCK_MAP.get(resourceId);if (dbLockMap == null) {LOCK_MAP.putIfAbsent(resourceId,new ConcurrentHashMap<String, ConcurrentHashMap<Integer, Map<String, Long>>>());dbLockMap = LOCK_MAP.get(resourceId);}for (RowLock lock : rowLocks) {String tableName = lock.getTableName();String pk = lock.getPk();ConcurrentHashMap<Integer, Map<String, Long>> tableLockMap = dbLockMap.get(tableName);if (tableLockMap == null) {dbLockMap.putIfAbsent(tableName, new ConcurrentHashMap<Integer, Map<String, Long>>());tableLockMap = dbLockMap.get(tableName);}int bucketId = pk.hashCode() % BUCKET_PER_TABLE;Map<String, Long> bucketLockMap = tableLockMap.get(bucketId);if (bucketLockMap == null) {tableLockMap.putIfAbsent(bucketId, new ConcurrentHashMap<String, Long>());bucketLockMap = tableLockMap.get(bucketId);}synchronized (bucketLockMap) {Long lockingTransactionId = bucketLockMap.get(pk);if (lockingTransactionId == null) {//No existing lockbucketLockMap.put(pk, transactionId);Set<String> keysInHolder = bucketHolder.get(bucketLockMap);if (keysInHolder == null) {bucketHolder.putIfAbsent(bucketLockMap, new ConcurrentSet<String>());keysInHolder = bucketHolder.get(bucketLockMap);}keysInHolder.add(pk);} else if (lockingTransactionId.longValue() == transactionId) {// Locked by mecontinue;} else {LOGGER.info("Global lock on [" + tableName + ":" + pk + "] is holding by " + lockingTransactionId);try {// Release all acquired locks.branchSession.unlock();} catch (TransactionException e) {throw new FrameworkException(e);}return false;}}}return true;}@Overridepublic boolean releaseLock(List<RowLock> rowLock) {ConcurrentHashMap<Map<String, Long>, Set<String>> lockHolder = branchSession.getLockHolder();if (lockHolder == null || lockHolder.size() == 0) {return true;}Iterator<Map.Entry<Map<String, Long>, Set<String>>> it = lockHolder.entrySet().iterator();while (it.hasNext()) {Map.Entry<Map<String, Long>, Set<String>> entry = it.next();Map<String, Long> bucket = entry.getKey();Set<String> keys = entry.getValue();synchronized (bucket) {for (String key : keys) {Long v = bucket.get(key);if (v == null) {continue;}if (v.longValue() == branchSession.getTransactionId()) {bucket.remove(key);}}}}lockHolder.clear();return true;}@Overridepublic boolean isLockable(List<RowLock> rowLocks) {if (CollectionUtils.isEmpty(rowLocks)) {//no lockreturn true;}Long transactionId = rowLocks.get(0).getTransactionId();String resourceId = rowLocks.get(0).getResourceId();ConcurrentHashMap<String, ConcurrentHashMap<Integer, Map<String, Long>>> dbLockMap = LOCK_MAP.get(resourceId);if (dbLockMap == null) {return true;}for (RowLock rowLock : rowLocks) {String xid = rowLock.getXid();String tableName = rowLock.getTableName();String pk = rowLock.getPk();ConcurrentHashMap<Integer, Map<String, Long>> tableLockMap = dbLockMap.get(tableName);if (tableLockMap == null) {continue;}int bucketId = pk.hashCode() % BUCKET_PER_TABLE;Map<String, Long> bucketLockMap = tableLockMap.get(bucketId);if (bucketLockMap == null) {continue;}Long lockingTransactionId = bucketLockMap.get(pk);if (lockingTransactionId == null || lockingTransactionId.longValue() == transactionId) {// Locked by mecontinue;} else {LOGGER.info("Global lock on [" + tableName + ":" + pk + "] is holding by " + lockingTransactionId);return false;}}return true;}@Overridepublic void cleanAllLocks() {LOCK_MAP.clear();}
}