Seata解析-TC会话管理器SessionManager

x33g5p2x  于2021-12-21 转载在 其他  
字(5.1k)|赞(0)|评价(0)|浏览(595)

本文基于seata 1.3.0版本

在介绍ServerOnRequestProcessor处理器时,多次提到了会话管理器,比如处理分支事务注册请求时,对全局事务对象加锁就是调用了根会话管理器,再如处理全局事务提交请求中,会将全局事务对象添加到异步提交管理器中,这个异步提交管理器就是会话管理器的一种,还有在分析代码的时候,多次看到往globalSession对象中添加监听器,其实这个监听器就是会话管理器。
在seata中,事务也叫作会话,所以会话管理器其实就是事务管理器。因此会话管理器在seata中的作用非常重要,本文就来详细介绍会话管理器。

一、SessionManager

SessionManager是一个接口,它有三个最终实现类,下图是会话管理器的继承结构。

SessionManager继承了接口SessionLifecycleListener,这是一个会话生命周期监听器,会话管理器能够充当监听器,就是该接口在起作用。SessionLifecycleListener接口的方法定义了要监听的事件。

  1. public interface SessionLifecycleListener {
  2. /** * 监听全局事务的开启,当处理全局事务开启请求时,就会调用该方法 */
  3. void onBegin(GlobalSession globalSession) throws TransactionException;
  4. /** * 监听全局事务对象GlobalSession的状态变化,只要是GlobalSession的状态发生变化,就会调用该方法 */
  5. void onStatusChange(GlobalSession globalSession, GlobalStatus status) throws TransactionException;
  6. /** * 监听分支事务状态的变化,在处理分支状态报告请求时,会调用该方法 */
  7. void onBranchStatusChange(GlobalSession globalSession, BranchSession branchSession, BranchStatus status)
  8. throws TransactionException;
  9. /** * 监听新的分支事务注册 */
  10. void onAddBranch(GlobalSession globalSession, BranchSession branchSession) throws TransactionException;
  11. /** * 监听分支事务从全局事务对象中移除, * 当处理全局事务回滚请求全局事务提交请求时,都会有移除分支事务的动作,因此都会触发该方法 */
  12. void onRemoveBranch(GlobalSession globalSession, BranchSession branchSession) throws TransactionException;
  13. /** * 监听全局事务关闭,也就是监听GlobalSession的close方法。 * 在处理全局事务提交请求和全局事务回滚请求时,都会调用GlobalSession的close方法。 */
  14. void onClose(GlobalSession globalSession) throws TransactionException;
  15. /** * 监听全局事务终止,也就是监听GlobalSession的end方法。 * 当要求全局事务提交或者回滚时,无论最后成功与否,seata都会调用GlobalSession的end方法,因此都会触发onEnd */
  16. void onEnd(GlobalSession globalSession) throws TransactionException;
  17. }

我在每个方法上都做了注释,介绍了每个方法都是在哪些场景下会被调用。
SessionManager接口的方法主要是侧重于管理:

  1. public interface SessionManager extends SessionLifecycleListener, Disposable {
  2. /** * 将全局事务对象添加到会话管理器中,当全局事务异步提交或者异步回滚时,都会调用该方法 */
  3. void addGlobalSession(GlobalSession session) throws TransactionException;
  4. /** * 根据XID查找GlobalSession */
  5. GlobalSession findGlobalSession(String xid) ;
  6. /** * 不同的存储模式下,本方法和上面的方法实现不同,如果存储模式是file,则两个方法完全一致, * 如果存储模式是db,则上面的方法相当于调用findGlobalSession(xid, true) * 如果第二个参数为true,表示返回的GlobalSession对象中带有分支事务集合 */
  7. GlobalSession findGlobalSession(String xid, boolean withBranchSessions);
  8. /** * 更新事务对象的状态 */
  9. void updateGlobalSessionStatus(GlobalSession session, GlobalStatus status) throws TransactionException;
  10. /** * 从管理器中移除GlobalSession * 当异步提交重试超时时,会调用该方法 */
  11. void removeGlobalSession(GlobalSession session) throws TransactionException;
  12. /** * 向GlobalSession中添加分支事务对象,当分支事务注册时,会调用该方法 */
  13. void addBranchSession(GlobalSession globalSession, BranchSession session) throws TransactionException;
  14. /** * 更新分支事务状态 */
  15. void updateBranchSessionStatus(BranchSession session, BranchStatus status) throws TransactionException;
  16. /** * 从全局事务中移除分支事务,当全局事务提交或者回滚时,会调用该方法 */
  17. void removeBranchSession(GlobalSession globalSession, BranchSession session) throws TransactionException;
  18. /** * 返回所有的全局会话对象 */
  19. Collection<GlobalSession> allSessions();
  20. /** * 根据条件查找符合要求的GlobalSession */
  21. List<GlobalSession> findGlobalSessions(SessionCondition condition);
  22. /** * 对全局事务对象加锁,当修改全局事务对象的状态时,都会加锁 */
  23. <T> T lockAndExecute(GlobalSession globalSession, GlobalSession.LockCallable<T> lockCallable)
  24. throws TransactionException;
  25. }

seata提供了SessionManager管理器的三个实现类:DataBaseSessionManager、FileSessionManager、RedisSessionManager
这三个实现类都继承了抽象类:AbstractSessionManager。

二、AbstractSessionManager

AbstractSessionManager实现了两个接口SessionManager和SessionLifecycleListener。
AbstractSessionManager提供了两个属性:

  1. //事务存储管理器,
  2. //TransactionStoreManager有三种实现,分别对应了三种存储模式,
  3. //TransactionStoreManager用于将事务日志写入数据库、redis或者文件
  4. protected TransactionStoreManager transactionStoreManager;
  5. //表示管理器的名字,如果管理器的实现类是FileSessionManager,那么name的值是文件名
  6. protected String name;

下图是AbstractSessionManager的方法:

上面这些方法除了onClose、writeSession、destroy、setTransactionStoreManager四个方法之外,其他的方法都是直接或者间接调用writeSession方法,该方法的作用是写事务日志。从这里可以看出,凡是修改分支事务和全局事务状态,增加或删除分支事务,注册或者删除全局事务,都会写事务日志。
writeSession方法也非常简单,直接调用事务存储管理器的writeSession方法写入日志,如果事务日志写入失败,则抛出对应的异常,在下一篇文章中会在介绍该方法:

  1. private void writeSession(LogOperation logOperation, SessionStorable sessionStorable) throws TransactionException {
  2. if (!transactionStoreManager.writeSession(logOperation, sessionStorable)) {
  3. if (LogOperation.GLOBAL_ADD.equals(logOperation)) {
  4. throw new GlobalTransactionException(TransactionExceptionCode.FailedWriteSession,
  5. "Fail to store global session");
  6. } else if (LogOperation.GLOBAL_UPDATE.equals(logOperation)) {
  7. throw new GlobalTransactionException(TransactionExceptionCode.FailedWriteSession,
  8. "Fail to update global session");
  9. } else if (LogOperation.GLOBAL_REMOVE.equals(logOperation)) {
  10. throw new GlobalTransactionException(TransactionExceptionCode.FailedWriteSession,
  11. "Fail to remove global session");
  12. } else if (LogOperation.BRANCH_ADD.equals(logOperation)) {
  13. throw new BranchTransactionException(TransactionExceptionCode.FailedWriteSession,
  14. "Fail to store branch session");
  15. } else if (LogOperation.BRANCH_UPDATE.equals(logOperation)) {
  16. throw new BranchTransactionException(TransactionExceptionCode.FailedWriteSession,
  17. "Fail to update branch session");
  18. } else if (LogOperation.BRANCH_REMOVE.equals(logOperation)) {
  19. throw new BranchTransactionException(TransactionExceptionCode.FailedWriteSession,
  20. "Fail to remove branch session");
  21. } else {
  22. throw new BranchTransactionException(TransactionExceptionCode.FailedWriteSession,
  23. "Unknown LogOperation:" + logOperation.name());
  24. }
  25. }
  26. }

相关文章