Seata 分布式事务启动配置分析

红太狼 2023-06-05 05:56 10阅读 0赞

想要掌握 Seata 的配置，必须了解 Seata 的启动过程，了解启动时的各项配置，才能在配置时知道该干什么。

## 用到的配置 ##

这里先列出 Server 启动过程中实际用到的配置配置，下文会具体分析。

<table> 
 <thead> 
  <tr> 
   <th align="left">属性</th> 
   <th align="left">读取值</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">config.type</td> 
   <td align="left">file</td> 
  </tr> 
  <tr> 
   <td align="left">config.file.name</td> 
   <td align="left">file.conf</td> 
  </tr> 
  <tr> 
   <td align="left">metrics.enabled</td> 
   <td align="left">FALSE</td> 
  </tr> 
  <tr> 
   <td align="left">recovery.asyn-committing-retry-period</td> 
   <td align="left">1000</td> 
  </tr> 
  <tr> 
   <td align="left">recovery.committing-retry-period</td> 
   <td align="left">1000</td> 
  </tr> 
  <tr> 
   <td align="left">recovery.rollbacking-retry-period</td> 
   <td align="left">1000</td> 
  </tr> 
  <tr> 
   <td align="left">recovery.timeout-retry-period</td> 
   <td align="left">1000</td> 
  </tr> 
  <tr> 
   <td align="left">registry.type</td> 
   <td align="left">file</td> 
  </tr> 
  <tr> 
   <td align="left">service.max.commit.retry.timeout</td> 
   <td align="left">-1</td> 
  </tr> 
  <tr> 
   <td align="left">service.max.rollback.retry.timeout</td> 
   <td align="left">-1</td> 
  </tr> 
  <tr> 
   <td align="left">store.file.dir</td> 
   <td align="left">sessionStore</td> 
  </tr> 
  <tr> 
   <td align="left">store.file.file-write-buffer-cache-size</td> 
   <td align="left">16384</td> 
  </tr> 
  <tr> 
   <td align="left">store.file.flush-disk-mode</td> 
   <td align="left">async</td> 
  </tr> 
  <tr> 
   <td align="left">store.file.session.reload.read_size</td> 
   <td align="left">100</td> 
  </tr> 
  <tr> 
   <td align="left">store.mode</td> 
   <td align="left">file</td> 
  </tr> 
  <tr> 
   <td align="left">transaction.undo.log.delete.period</td> 
   <td align="left">86400000</td> 
  </tr> 
  <tr> 
   <td align="left">transport.heartbeat</td> 
   <td align="left">TRUE</td> 
  </tr> 
  <tr> 
   <td align="left">transport.server</td> 
   <td align="left">NIO</td> 
  </tr> 
  <tr> 
   <td align="left">transport.thread-factory.boss-thread-prefix</td> 
   <td align="left">NettyBoss</td> 
  </tr> 
  <tr> 
   <td align="left">transport.thread-factory.boss-thread-size</td> 
   <td align="left">1</td> 
  </tr> 
  <tr> 
   <td align="left">transport.thread-factory.share-boss-worker</td> 
   <td align="left">FALSE</td> 
  </tr> 
  <tr> 
   <td align="left">transport.thread-factory.worker-thread-prefix</td> 
   <td align="left">NettyServerNIOWorker</td> 
  </tr> 
  <tr> 
   <td align="left">transport.thread-factory.worker-thread-size</td> 
   <td align="left">8</td> 
  </tr> 
  <tr> 
   <td align="left">transport.type</td> 
   <td align="left">TCP</td> 
  </tr> 
 </tbody> 
</table>

## Server 入口 ##

`io.seata.server.Server` 类是整个服务的入口，从这里的 `main` 方式入手。

public static void main(String[] args) throws IOException { 
        //initialize the metrics
        MetricsManager.get().init();//1
    
        //initialize the parameter parser
        ParameterParser parameterParser = new ParameterParser(args);
    
        System.setProperty(ConfigurationKeys.STORE_MODE, parameterParser.getStoreMode());
    
        RpcServer rpcServer = new RpcServer(WORKING_THREADS);
        //server port
        rpcServer.setListenPort(parameterParser.getPort());
        UUIDGenerator.init(parameterParser.getServerNode());
        //log store mode : file、db
        SessionHolder.init(parameterParser.getStoreMode());
    
        DefaultCoordinator coordinator = new DefaultCoordinator(rpcServer);
        coordinator.init();
        rpcServer.setHandler(coordinator);
        // register ShutdownHook
        ShutdownHook.getInstance().addDisposable(coordinator);
    
        //127.0.0.1 and 0.0.0.0 are not valid here.
        if (NetUtil.isValidIp(parameterParser.getHost(), false)) { 
            XID.setIpAddress(parameterParser.getHost());
        } else { 
            XID.setIpAddress(NetUtil.getLocalIp());
        }
        XID.setPort(rpcServer.getListenPort());
    
        rpcServer.init();
    
        System.exit(0);
    }

下面逐层对上面代码进行分析。

## 1. `MetricsManager.get().init()` ##

`MetricsManager` 是一个单例实现，在 `init` 方法中，首先调用了 `ConfigurationFactory.getInstance()` 方法，该方法是最重要的一个配置入口，继续深入进去来看。

### 1.1 `ConfigurationFactory.getInstance()` ###

`ConfigurationFactory` 返回的 `io.seata.config.Configuration<T>` 的单例，该单例通过 `buildConfiguration()` 创建。

在 `buildConfiguration()` 中，又首先使用了 `CURRENT_FILE_INSTANCE.getConfig` 方法。

这里的 `CURRENT_FILE_INSTANCE` 创建方式如下：

private static final String REGISTRY_CONF_PREFIX = "registry";
    private static final String REGISTRY_CONF_SUFFIX = ".conf";
    private static final String ENV_SYSTEM_KEY = "SEATA_ENV";
    private static final String ENV_PROPERTY_KEY = "seataEnv";
    
    private static final String SYSTEM_PROPERTY_SEATA_CONFIG_NAME = "seata.config.name";
    
    private static final String ENV_SEATA_CONFIG_NAME = "SEATA_CONFIG_NAME";
    
    public static final Configuration CURRENT_FILE_INSTANCE;
    
    static { 
        //0.8.1 增加这个配置后，配置文件不用局限在 conf 目录（类路径cp）下面了，通过 file: 前缀可以配置为任意的路径
        String seataConfigName = System.getProperty(SYSTEM_PROPERTY_SEATA_CONFIG_NAME);
        if (null == seataConfigName) { 
            seataConfigName = System.getenv(ENV_SEATA_CONFIG_NAME);
        }
        if (null == seataConfigName) { 
            seataConfigName = REGISTRY_CONF_PREFIX;
        }
        String envValue = System.getProperty(ENV_PROPERTY_KEY);
        if (null == envValue) { 
            envValue = System.getenv(ENV_SYSTEM_KEY);
        }
        CURRENT_FILE_INSTANCE = (null == envValue) ? new FileConfiguration(seataConfigName + REGISTRY_CONF_SUFFIX)
                : new FileConfiguration(seataConfigName + "-" + envValue + REGISTRY_CONF_SUFFIX);
    }

> **2019-10-12** 更新上述代码为 0.8.1 版本，0.8.1 之后配置的路径没有了限制，可以更方便的在容器或者K8S中动态配置。

默认的 registry.conf 配置文件如下：

registry {
      # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa
      type = "file"
      # 为了简短，只保留默认的 file 相关内容
      file {
        name = "file.conf"
      }
    }
    
    config {
      # file、nacos 、apollo、zk、consul、etcd3
      type = "file"
      file {
        name = "file.conf"
      }
    }

### 1.2 获取 seata 配置 ###

通过上述方式得到了 `CURRENT_FILE_INSTANCE`，现在回到 `buildConfiguration()` 方法：

private static Configuration buildConfiguration() { 
        ConfigType configType = null;
        String configTypeName = null;
        try { 
            // 获取 config.type 名称，参考上面默认配置，值为 file
            configTypeName = CURRENT_FILE_INSTANCE.getConfig(
                ConfigurationKeys.FILE_ROOT_CONFIG + ConfigurationKeys.FILE_CONFIG_SPLIT_CHAR
                    + ConfigurationKeys.FILE_ROOT_TYPE);
            // 对应枚举值为 File
            configType = ConfigType.getType(configTypeName);
        } catch (Exception e) { 
            throw new NotSupportYetException("not support register type: " + configTypeName, e);
        }
        // 默认配置这里是 File
        if (ConfigType.File == configType) { 
            // 获取 config.file.name 值
            String pathDataId = ConfigurationKeys.FILE_ROOT_CONFIG + ConfigurationKeys.FILE_CONFIG_SPLIT_CHAR
                + FILE_TYPE + ConfigurationKeys.FILE_CONFIG_SPLIT_CHAR
                + NAME_KEY;
            // 上述默认值为 file.conf
            String name = CURRENT_FILE_INSTANCE.getConfig(pathDataId);
            // 读取 file.conf 配置
            return new FileConfiguration(name);
        } else { 
            return EnhancedServiceLoader.load(ConfigurationProvider.class, Objects.requireNonNull(configType).name())
                .provide();
        }
    }

从上面逻辑看，这里就是从支持的多种 registry 中获取具体的配置，默认是 file。

### 1.3 处理 metrics 配置 ###

回到上一层方法：

// ConfigurationFactory.getInstance() 中是 file.conf 配置
    boolean enabled = ConfigurationFactory.getInstance().getBoolean(
                // 获取该配置中的 metrics.enabled，默认值为 false
                ConfigurationKeys.METRICS_PREFIX + ConfigurationKeys.METRICS_ENABLED, false);

`file.conf` 中 `metrics` 部分配置如下：

## metrics settings
    metrics {
      # 默认不启用
      enabled = false
      registry-type = "compact"
      # multi exporters use comma divided
      exporter-list = "prometheus"
      exporter-prometheus-port = 9898
    }

启用后，会读取 `metrics. registry-type`，目前仅支持 `compact`，然后通过下面方法：

EnhancedServiceLoader.load(Registry.class, Objects.requireNonNull(registryType).name());

获取对应的实现返回，在这里的 `load` 方法中也是单例形式，只会初始化一次。

然后在读取 `metrics.exporter-list`（逗号隔开） 获取所有支持的 `Exporter`（目前也只支持 `prometheus`）。

`prometheus` 对应的 `PrometheusExporter` 实现中会读取上面的 `exporter-prometheus-port` 获取 HTTP 服务的端口号。

## 2. 解析命令行参数 ##

在进行了配置的初始化后，开始处理 main 方法的命令行参数：

// 解析参数使用了 http://www.jcommander.org/
    //initialize the parameter parser
    ParameterParser parameterParser = new ParameterParser(args);
    // 获取存储方式("--storeMode", "-m")，默认为 file，可选 db
    System.setProperty(ConfigurationKeys.STORE_MODE, parameterParser.getStoreMode());
    
    RpcServer rpcServer = new RpcServer(WORKING_THREADS);
    // 获取端口号("--port", "-p")，默认为 8091
    rpcServer.setListenPort(parameterParser.getPort());
    // 获取服务节点ID("--serverNode", "-n")，默认 1
    UUIDGenerator.init(parameterParser.getServerNode());
    // 初始化存储
    SessionHolder.init(parameterParser.getStoreMode());//3

## 3. 初始化 `RpcServer` ##

代码如下：

/** * Instantiates a new Abstract rpc server. * * @param messageExecutor the message executor */
    public RpcServer(ThreadPoolExecutor messageExecutor) { 
        super(new NettyServerConfig(), messageExecutor);
    }

这里用到了 `NettyServerConfig`，这个类中存在大量类似下面的方法：

/** * Get boss thread prefix string. * * @return the string */
    public String getBossThreadPrefix() { 
        return CONFIG.getConfig("transport.thread-factory.boss-thread-prefix", DEFAULT_BOSS_THREAD_PREFIX);
    }

这里用到了下面部分的配置：

transport {
      # tcp udt unix-domain-socket
      type = "TCP"
      #NIO NATIVE
      server = "NIO"
      #enable heartbeat
      heartbeat = true
      #thread factory for netty
      thread-factory {
        boss-thread-prefix = "NettyBoss"
        worker-thread-prefix = "NettyServerNIOWorker"
        server-executor-thread-prefix = "NettyServerBizHandler"
        share-boss-worker = false
        client-selector-thread-prefix = "NettyClientSelector"
        client-selector-thread-size = 1
        client-worker-thread-prefix = "NettyClientWorkerThread"
        # netty boss thread size,will not be used for UDT
        boss-thread-size = 1
        #auto default pin or 8
        worker-thread-size = 8
      }
      shutdown {
        # when destroy server, wait seconds
        wait = 3
      }
      serialization = "seata"
      compressor = "none"
    }

这里是对 netty 的各种详细配置。

继续往下深入看 `UUIDGenerator.init`

## 4. `UUIDGenerator.init` ##

这里根据节点ID定义了服务器 UUID 值的范围，避免了节点间的冲突。

**有没有人觉得这段代码存在疑问？**

public static long generateUUID() { 
        //假设在超过最大值，进入下面同步前获取了10个id
        //在上面第一个 id 经过 UUID.set(id); 后又获取了 10 个新id
        //这10个新id 会不会和上面的10个- UUID_INTERNAL 后出现部分重复？
        long id = UUID.incrementAndGet();
        if (id >= UUID_INTERNAL * (serverNodeId + 1)) { 
            synchronized (UUID) { 
                if (UUID.get() >= id) { 
                    id -= UUID_INTERNAL;
                    UUID.set(id);
                }
            }
        }
        return id;
    }

## 5. `SessionHolder.init` ##

首先参数中的 `storeMode` 可选，如果没有设置会按下面方式获取：

if (StringUtils.isBlank(mode)) { 
        //use default
        mode = CONFIG.getConfig(ConfigurationKeys.STORE_MODE);
    }

也就是前面 `file.conf` 中的 `store.mode`，这部分默认配置如下：

## transaction log store
    store {
      ## store mode: file、db
      mode = "file"
    
      ## file store
      file {
        dir = "sessionStore"
    
        # branch session size , if exceeded first try compress lockkey, still exceeded throws exceptions
        max-branch-session-size = 16384
        # globe session size , if exceeded throws exceptions
        max-global-session-size = 512
        # file buffer size , if exceeded allocate new buffer
        file-write-buffer-cache-size = 16384
        # when recover batch read size
        session.reload.read_size = 100
        # async, sync
        flush-disk-mode = async
      }
    
      ## database store
      db {
        ## the implement of javax.sql.DataSource, such as DruidDataSource(druid)/BasicDataSource(dbcp) etc.
        datasource = "dbcp"
        ## mysql/oracle/h2/oceanbase etc.
        db-type = "mysql"
        driver-class-name = "com.mysql.jdbc.Driver"
        url = "jdbc:mysql://127.0.0.1:3306/seata"
        user = "mysql"
        password = "mysql"
        min-conn = 1
        max-conn = 3
        global.table = "global_table"
        branch.table = "branch_table"
        lock-table = "lock_table"
        query-limit = 100
      }
    }

上面的默认值也是 `file`，如果使用 `db` 参考上面配置修改即可。

再往下就是针对 db 和 file 的两种处理策略。

### 5.1 db 方式 ###

ROOT_SESSION_MANAGER = EnhancedServiceLoader.load(SessionManager.class, StoreMode.DB.name());

这里额外指定了最后一个参数 `activateName`，这会在默认的 `META-INF/services/` 和 `META-INF/seata/` 基础上额外去 `META-INF/seata/db` 目录加载资源，并且该资源会在队列的最后一个位置。然后判断所有的资源是否有 `@LoadLevel` 并且和 `activateName` 匹配，如果存在多个匹配的值，就会使用最后一个。如果没有匹配的值，会使用所有实现的最后一个。

这里的 db 类定义如下：

@LoadLevel(name = "db")
    public class DataBaseSessionManager extends AbstractSessionManager
        implements SessionManager, SessionLifecycleListener, Initialize {

`ASYNC_COMMITTING_SESSION_MANAGER` 等3个只是增加了额外的参数，在初始化的时候会使用相应的构造方法进行创建。

### 5.2 file 方式 ###

这里首先读取 `store.file.dir` 获取存储 file 的路径，然后创建下面的实现：

@LoadLevel(name = "file")
    public class FileBasedSessionManager extends DefaultSessionManager implements Reloadable {

在 db 和 file 中还涉及了 `TransactionStoreManager` 的实例化，这里不再深入。

在 `SessionHolder.init` 最后还有一个针对 file 方式的 `reload`。

回到 main 方法继续。

## 6. `coordinator.init()` ##

// 创建协调者
    DefaultCoordinator coordinator = new DefaultCoordinator(rpcServer);
    // 初始化
    coordinator.init();
    rpcServer.setHandler(coordinator);
    // register ShutdownHook
    ShutdownHook.getInstance().addDisposable(coordinator);

在 `DefaultCoordinator` 实现中，使用下面部分的配置：

recovery {
      #schedule committing retry period in milliseconds
      committing-retry-period = 1000
      #schedule asyn committing retry period in milliseconds
      asyn-committing-retry-period = 1000
      #schedule rollbacking retry period in milliseconds
      rollbacking-retry-period = 1000
      #schedule timeout retry period in milliseconds
      timeout-retry-period = 1000
    }
    
    transaction {
      undo.data.validation = true
      undo.log.serialization = "jackson"
      undo.log.save.days = 7
      #schedule delete expired undo_log in milliseconds
      undo.log.delete.period = 86400000
      undo.log.table = "undo_log"
    }
    
    service {
      #vgroup->rgroup
      vgroup_mapping.my_test_tx_group = "default"
      #only support single node
      default.grouplist = "127.0.0.1:8091"
      #degrade current not support
      enableDegrade = false
      #disable
      disable = false
      #unit ms,s,m,h,d represents milliseconds, seconds, minutes, hours, days, default permanent
      max.commit.retry.timeout = "-1"
      max.rollback.retry.timeout = "-1"
    }

## 7. 其他 ##

// 服务绑定 IP("--host", "-h")
    // 127.0.0.1 and 0.0.0.0 are not valid here.
    if (NetUtil.isValidIp(parameterParser.getHost(), false)) { 
        XID.setIpAddress(parameterParser.getHost());
    } else { 
        XID.setIpAddress(NetUtil.getLocalIp());
    }
    XID.setPort(rpcServer.getListenPort());
    // 启动服务，通过 netty hold 住
    rpcServer.init();
    // 退出
    System.exit(0);

这一篇只是记录下各种参数在何时使用，便于初始配置时理解和使用，后续看情况补充 seata 相关内容。