Seata Server 整体流程

刺骨的言语ヽ痛彻心扉 2022-11-22 10:13 198阅读 0赞

Seata Server 在分布式事务中, 担任事物协调者(TC)的角色  
![在这里插入图片描述][4697c47143856ac7cd1d39d23ad62891.png_pic_center]

从TC的角度: 我们可以把 **TC** 看成处理 **RM**,**TM** 请求的一个业务系统

# 核心组件初始化 #

为了了解TC的核心实现， 要先介绍核心组件支持

服务端分为八大组件，支持服务的启动和运行  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L216NDEzOA_size_16_color_FFFFFF_t_70_pic_center]

1.  ParameterParser 启动时参数解析，支持k8s,docker
2.  MetricManager 统计分析组件, 默认不启用
3.  ShutdownHook 应用程序关闭时通知已注册的组件卸载
4.  DefaultCoordinator 处理业务逻辑， 例如事物开启，分支注册, 初始化调度
5.  NettyRemotingServer 监听端口，编码解码
6.  SessionManager 会话管理器 管理全局事物，分支事物的状态
7.  ServerHandler 服务处理器， 五大请求处理： 请求，响应，TM注册，RM注册，心跳
8.  UUIDGenerator 生成全局事物ID

服务的启动过程也就是组件的初始化过程

// 代码取自: seata:1.4.0 , 并删减了和组件无关的部分
     public static void main(String[] args) throws IOException { 
            //初始化组件 ParameterParser 
            ParameterParser parameterParser = new ParameterParser(args);
            
             //初始化组件 MetricsManager
            MetricsManager.get().init();
    		
            NettyRemotingServer nettyRemotingServer = new NettyRemotingServer(WORKING_THREADS);
            nettyRemotingServer.setListenPort(parameterParser.getPort());
            
          	//初始化组件 UUIDGenerator
            UUIDGenerator.init(parameterParser.getServerNode());
          	//初始化组件 SessionHolder, log store mode : file, db, redis
            SessionHolder.init(parameterParser.getStoreMode());
    
    		//初始化组件 DefaultCoordinator
            DefaultCoordinator coordinator = new DefaultCoordinator(nettyRemotingServer);
            coordinator.init();
            nettyRemotingServer.setHandler(coordinator);
            // register ShutdownHook
            ShutdownHook.getInstance().addDisposable(coordinator);
            ShutdownHook.getInstance().addDisposable(nettyRemotingServer);
    
    		//初始化组件 NettyRemotingServer , ServerHandler
            nettyRemotingServer.init();
        }

# 请求处理 #

虽然提供了这么多组件, 和处理请求相关的却不多。 既然是服务，就需要接受请求，处理请求

从接口的角度来讲，分为两大模块：

*  接受请求组件: RemotingServer
 *  处理请求组件: TransactionMessageHandler

Seata Server 的主要结构类图

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L216NDEzOA_size_16_color_FFFFFF_t_70_pic_center 1]

## RemotingServer ##

启动时，由NettyRemotingServer 将消息注册到 **消息处理器**，**消息处理器** 的处理器映射关系在AbstractRemotingServer的属性processorTable， 这是一个HashMap， 由于不需要动态注册，也就不需要关心线程安全

在处理消息之前, 首先通过Netty解码器, 获取 RpcMessage对象

TC的通道处理器的顺序如下:

1.  心跳处理器
2.  协议解码器
3.  协议编码器
4.  ServerHandler

消息首先经过协议解码器，然后被ServerHandler处理

协议解码器解析Seata的自定义协议。 通过消息类型、压缩方式、以及序列化方式将客户端数据组装成RpcMessage对象

ServerHandler是AbstractNettyRemoting的内部类， 当收到请求时触发channelRead，处理 RpcMessage

ServerHandler 是 AbstractNettyRemotingServer 的内部类，所以能调用父类方法 AbstractNettyRemoting\#processMessage.

processMessage方法首先根据**请求类型**从processorTable拿到对应的消息处理器和线程池。

*  如果有线程池就在线程池中处理
 *  没有线程池就同步处理

AbstractNettyRemoting\#processMessage 处理消息的方式如下:

// 篇幅限制， 删除大量日志和try,catch
    public abstract class AbstractNettyRemoting { 
    // 一个消息类型， 映射两个对象： 消息处理器， 线程池
    protected final HashMap<Integer/*消息类型*/, 
     						Pair<RemotingProcessor/*消息处理器*/, ExecutorService/*线程池*/>
     					> processorTable = new HashMap<>(32);
     // 处理消息
    protected void processMessage(ChannelHandlerContext ctx, RpcMessage rpcMessage) throws Exception { 
            Object body = rpcMessage.getBody();
            if (body instanceof MessageTypeAware) { 
                MessageTypeAware messageTypeAware = (MessageTypeAware) body;
                // 通过请求的消息类型，从已注册的处理映射器找线程池和处理器
                final Pair<RemotingProcessor, ExecutorService> pair = 
                		this.processorTable.get((int) messageTypeAware.getTypeCode());
                if (pair != null) { 
                	// 如果线程池存在，由线程池处理业务
                    if (pair.getSecond() != null) { 
                            pair.getSecond().execute(() -> { 
                              	// 消息处理器处理业务逻辑
                              	 pair.getFirst().process(ctx, rpcMessage);
                            });
                    } else { 
                    		// 如果没有线程池， 同步处理业务。 目前只有心跳和事务管理器注册的消息是同步处理
                    		// 消息处理器处理业务逻辑
                            pair.getFirst().process(ctx, rpcMessage);
                    }
                }
            }
        }
    }

**消息处理器**

消息处理器在TC一共有五种类型， 服务启动时将消息处理器注册到 AbstractNettyRemoting\#processorTable 。

Seata Server 消息处理器的类型:

1.  请求处理器
2.  响应处理器
3.  注册事物管理器处理器
4.  注册资源管理器处理器
5.  服务端心跳处理器  
    ![在这里插入图片描述][20201103221027659.png_pic_center]

这五种消息处理器， 对应的消息类型和功能如下

<table> 
 <thead> 
  <tr> 
   <th>消息处理器</th> 
   <th>消息类型</th> 
   <th>中文类型</th> 
   <th>描述</th> 
   <th>执行方式</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>ServerOnRequestProcessor</td> 
   <td>TYPE_BRANCH_REGISTER</td> 
   <td>分支事物注册</td> 
   <td>RM发起分支事物注册</td> 
   <td>异步</td> 
  </tr> 
  <tr> 
   <td>ServerOnRequestProcessor</td> 
   <td>TYPE_BRANCH_STATUS_REPORT</td> 
   <td>分支状态报告</td> 
   <td>可选项， 告知TC状态已成功</td> 
   <td>异步</td> 
  </tr> 
  <tr> 
   <td>ServerOnRequestProcessor</td> 
   <td>TYPE_GLOBAL_BEGIN</td> 
   <td>开启全局事物</td> 
   <td>TM发起一阶段开启事物</td> 
   <td>异步</td> 
  </tr> 
  <tr> 
   <td>ServerOnRequestProcessor</td> 
   <td>TYPE_GLOBAL_COMMIT</td> 
   <td>全局事物提交</td> 
   <td>TM发起一阶段事物提交</td> 
   <td>异步</td> 
  </tr> 
  <tr> 
   <td>ServerOnRequestProcessor</td> 
   <td>TYPE_GLOBAL_LOCK_QUERY</td> 
   <td>开启全局锁</td> 
   <td>...</td> 
   <td>异步</td> 
  </tr> 
  <tr> 
   <td>ServerOnRequestProcessor</td> 
   <td>TYPE_GLOBAL_REPORT</td> 
   <td>全局事物报告</td> 
   <td>...</td> 
   <td>异步</td> 
  </tr> 
  <tr> 
   <td>ServerOnRequestProcessor</td> 
   <td>TYPE_GLOBAL_ROLLBACK</td> 
   <td>全局事物回滚</td> 
   <td>TM发起一阶段事物回滚</td> 
   <td>异步</td> 
  </tr> 
  <tr> 
   <td>ServerOnRequestProcessor</td> 
   <td>TYPE_GLOBAL_STATUS</td> 
   <td>获取全局事物状态</td> 
   <td>根据XID返回全局事物的状态</td> 
   <td>异步</td> 
  </tr> 
  <tr> 
   <td>ServerOnRequestProcessor</td> 
   <td>TYPE_SEATA_MERGE</td> 
   <td>合并消息</td> 
   <td>批量处理合并后的消息</td> 
   <td>异步</td> 
  </tr> 
  <tr> 
   <td>ServerOnResponseProcessor</td> 
   <td>TYPE_BRANCH_COMMIT_RESULT</td> 
   <td>分支事物提交</td> 
   <td>RM接受二阶段事物提交</td> 
   <td>异步</td> 
  </tr> 
  <tr> 
   <td>ServerOnResponseProcessor</td> 
   <td>TYPE_BRANCH_ROLLBACK_RESULT</td> 
   <td>分支事物回滚</td> 
   <td>RM接受二阶段事物回滚</td> 
   <td>异步</td> 
  </tr> 
  <tr> 
   <td>RegRmProcessor</td> 
   <td>TYPE_REG_RM</td> 
   <td>注册资源管理器</td> 
   <td>客户端启动时， 向TC发起注册</td> 
   <td>异步</td> 
  </tr> 
  <tr> 
   <td>RegTmProcessor</td> 
   <td>TYPE_REG_CLT</td> 
   <td>注册事务管理器</td> 
   <td>客户端启动时， 向TC发起注册</td> 
   <td>同步</td> 
  </tr> 
  <tr> 
   <td>ServerHeartbeatProcessor</td> 
   <td>TYPE_HEARTBEAT_MSG</td> 
   <td>心跳</td> 
   <td>收到客户端的PING，返回一个PONG</td> 
   <td>同步</td> 
  </tr> 
 </tbody> 
</table>

例如接收到一个分支事物注册（TYPE\_BRANCH\_REGISTER）类型的请求，  
找到处理器: `ServerOnRequestProcessor` ，然后调用 process 方法

ServerOnRequestProcessor 的处理流程如下， 其他处理器处理流程各不一样

1.  判断通道必须已注册
2.  事物消息处理器（TransactionMessageHandler）处理请求
3.  异步发送响应内容

ServerOnRequestProcessor\#processs 代码如下:

// 由于篇幅限制， 删除大部分try,catch. 日志等..
    public class ServerOnRequestProcessor implements RemotingProcessor { 
        @Override
        public void process(ChannelHandlerContext ctx, RpcMessage rpcMessage) throws Exception { 
        	// 判断通道是否已注册，（RM启动时，会发送TYPE_REG_RM类型的请求，服务端会缓存这个通道）
            if (ChannelManager.isRegistered(ctx.channel())) { 
                onRequestMessage(ctx, rpcMessage);
            }
        }
    
        private void onRequestMessage(ChannelHandlerContext ctx, RpcMessage rpcMessage) { 
                final AbstractMessage msg = (AbstractMessage) message;
                // 事物消息处理器（TransactionMessageHandler）处理请求
                AbstractResultMessage result = transactionMessageHandler.onRequest(msg, rpcContext);
                // 异步发送响应内容
                remotingServer.sendAsyncResponse(rpcMessage, ctx.channel(), result);
        }
    }

## TransactionMessageHandler ##

事物消息处理器的实现只有一个: DefaultCoordinator, 在启动时初始化字段 RemotingServer和DefaultCore。  
DefaultCoordinator 是如何根据请求找到具体的方法呢？

利用java的 动态单分派，静态多分派的特性。通过请求类型找到处理业务逻辑的方法

// DefaultCoordinator 的使用方式
         @Override
        public AbstractResultMessage onRequest(AbstractMessage request, RpcContext context) { 
            if (!(request instanceof AbstractTransactionRequestToTC)) { 
                throw new IllegalArgumentException();
            }
            AbstractTransactionRequestToTC transactionRequest = (AbstractTransactionRequestToTC) request;
            transactionRequest.setTCInboundHandler(this);
            // 动态单分派的方式调用每个请求自己的handler
            return transactionRequest.handle(context);
        }
    
    	// Request的handler实现
        public AbstractTransactionResponse handle(RpcContext rpcContext) { 
        	// 静态多分派的方式调用 DefaultCoordinator 的handle, ( handler对象就是DefaultCoordinator )
            return handler.handle(this, rpcContext);
        }

通过两次分派后进入具体的请求处理， 接着进入分支注册处理.  
例如这里的分支注册处理方法：

// 找到DefaultCoordinator中具体的处理方法
      public BranchRegisterResponse handle(BranchRegisterRequest request, final RpcContext rpcContext) { 
            BranchRegisterResponse response = new BranchRegisterResponse();
            // exceptionHandleTemplate 处理事务状态修改. 没抛异常： Success 抛出异常: Failed
            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;
        }

所有由DefaultCoordinator调用的处理方法模板都是一样的， 只是请求对象和响应对象的不同。

分支注册和和其他请求处理的方法声明是类似的， 其他处理声明:

// 开启全局事物
        GlobalBeginResponse handle(GlobalBeginRequest globalBegin, RpcContext rpcContext);
    	// 提交全局事物
        GlobalCommitResponse handle(GlobalCommitRequest globalCommit, RpcContext rpcContext);
    	// 回滚全局事物
        GlobalRollbackResponse handle(GlobalRollbackRequest globalRollback, RpcContext rpcContext);
     	// 分支注册
        BranchRegisterResponse handle(BranchRegisterRequest branchRegister, RpcContext rpcContext);
    	//... 篇幅限制，省略其他的请求方法声明

# 会话管理器 #

Seata 定义了四种会话管理器， 通过SPI方式加载， 使用SessionHolder的静态方法可以获取会话管理器

会话管理器的处理分为实时部分和非实时部分

实时部分用于处理全局事物，分支事物的增删改查  
非实时部分处理异常中断的重新拉起，非实时部分其实就是调度， 调度完善了会话管理器的生命周期， 实现自动关闭，回滚，提交会话。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L216NDEzOA_size_16_color_FFFFFF_t_70_pic_center 2]

会话管理器在后台承担持久层的工作，支持文件、数据库、Redis 三种形式。 对于线上环境 ，务必不能使用文件形式。 DB和Redis都是高可用的， 应该尽可能部署多台。

1.  全局会话管理器, 处理的状态:  
    UnKnown,Begin,Committing,CommitRetrying,Rollbacking,RollbackRetrying,  
    TimeoutRollbacking,TimeoutRollbackRetrying,AsyncCommitting
2.  异步提交会话管理器, 处理的状态: AsyncCommitting
3.  重试提交会话管理器, 处理的状态: CommitRetrying
4.  重试回滚会话管理器, 处理的状态: RollbackRetrying,Rollbacking,TimeoutRollbacking,TimeoutRollbackRetrying

# 调度 #

TC的调度一共有6种， 其中前4种用于完善会话的生命周期

## 重试回滚调度 ##

默认每秒执行一次， 通过配置修改调度 `server.recovery.rollbackingRetryPeriod=3000` 执行周期， 单位为毫秒

执行过程如下

1.  从重试回滚会话管理器获取所有会话进行遍历（如果是DB，就是查询指定状态的会话）
2.  判断如果是重试中并且距离全局事物开始时间大于12秒， 大于就强制重试，否则忽略这个会话
3.  判断是否重试超时，根据配置清理会话， 直接移除全局会话, 会话处理完成
4.  处理全局事物回滚

## 重试提交调度 ##

默认每秒执行一次， 通过配置修改调度 `server.recovery.committingRetryPeriod=3000` 执行周期， 单位为毫秒

通过SessionHolder获取重试提交会话管理器，遍历所有会话

1.  判断是否设置了超时参数`server.maxCommitRetryTimeout`，如果设置的值大于0，并且当前会话达到超时时间则移除当前会话，处理下一个会话
2.  将当前会话添加到全局会话管理器
3.  处理全局事物提交

## 异步提交调度 ##

默认每秒执行一次， 通过配置修改调度 `server.recovery.asynCommittingRetryPeriod=3000` 执行周期， 单位为毫秒

通过SessionHolder获取异步提交会话管理器，遍历所有会话

1.  判断当前会话状态必须是异步提交中( AsyncCommitting), 状态不匹配则处理下一个会话
2.  将当前会话添加到全局会话管理器
3.  处理全局事物提交

## 超时检查调度 ##

默认每秒执行一次， 通过配置修改调度 `server.recovery.timeoutRetryPeriod=3000` 执行周期， 单位为毫秒

通过SessionHolder获取全局会话管理器，遍历所有会话

1.  判断是否超时，如果超时:
    
     *  关闭当前会话
     *  当前会话状态变更为超时回滚中
2.  将当前会话添加到**重试回滚会话管理器**

## 回滚日志调度 ##

从服务器启动后第三分钟开始， 每24小时执行一次， 通过 `undo.logDeletePeriod=86400000` 修改调度周期， 单位为毫秒  
(86400000是一天的毫秒数）

1.  从ChannelManager获取当前在TC注册的所有的资源管理器。
2.  遍历所有资源管理器，发送回滚日志删除请求
3.  RM端收到请求后， 根据TC发送的保存天数， 保存指定天数内的数据。

回滚日志保存天数由TC设置， 默认7天。 设置方式： `undo.logSaveDays=7`

## 移除超时同步请求 ##

默认每3秒执行一次， 不可更改

应用场景: TC同步发送消息时， 会等待客户端的返回结果。

如果等待的时间超时了，就由这个定时器移除.( 将future的值设置为null， 认为客户端没返回)  
默认超时时间 30 秒， 并且不可更改

[4697c47143856ac7cd1d39d23ad62891.png_pic_center]: /images/20221120/52bda31a91124a15917ffd364fc21299.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L216NDEzOA_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/88934bca248a44c9bf855bab7f9aa549.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L216NDEzOA_size_16_color_FFFFFF_t_70_pic_center 1]: https://img-blog.csdnimg.cn/20201107135055453.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L216NDEzOA==,size_16,color_FFFFFF,t_70#pic_center
[20201103221027659.png_pic_center]: https://img-blog.csdnimg.cn/20201103221027659.png#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L216NDEzOA_size_16_color_FFFFFF_t_70_pic_center 2]: https://img-blog.csdnimg.cn/20201106001610598.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L216NDEzOA==,size_16,color_FFFFFF,t_70#pic_center