private FileWatcher watcher = new FileWatcher() {
    @Override
    public void onChange(FileChangeEvent event) {
      readClusterConfFromDisk();
    }

    @Override
    public boolean interest(String context) {
      return StringUtils.contains(context, "cluster.conf");
    }
};

@Override
public void start() throws NacosException {
  readClusterConfFromDisk();

  if (memberManager.getServerList().isEmpty()) {
    throw new NacosException(NacosException.SERVER_ERROR,
          "Failed to initialize the member node, is empty");
  }

  // Use the inotify mechanism to monitor file changes and automatically
  // trigger the reading of cluster.conf
  try {
    WatchFileCenter.registerWatcher(ApplicationUtils.getConfFilePath(), watcher);
  }
  catch (Throwable e) {
    Loggers.CLUSTER.error("An exception occurred in the launch file monitor : {}", e);
  }
}

@Override
public void start() throws NacosException {
  if (start.compareAndSet(false, true)) {
    this.maxFailCount = Integer.parseInt(ApplicationUtils.getProperty("maxHealthCheckFailCount", "12"));
    initAddressSys();
    run();
  }
}

private void initAddressSys() {
  String envDomainName = System.getenv("address_server_domain");
  if (StringUtils.isBlank(envDomainName)) {
    domainName = System.getProperty("address.server.domain", "jmenv.tbsite.net");
  } else {
    domainName = envDomainName;
  }
  String envAddressPort = System.getenv("address_server_port");
  if (StringUtils.isBlank(envAddressPort)) {
    addressPort = System.getProperty("address.server.port", "8080");
  } else {
    addressPort = envAddressPort;
  }
  addressUrl = System.getProperty("address.server.url",
        ApplicationUtils.getContextPath() + "/" + "serverlist");
  addressServerUrl = "http://" + domainName + ":" + addressPort + addressUrl;
  envIdUrl = "http://" + domainName + ":" + addressPort + "/env";

  Loggers.CORE.info("ServerListService address-server port:" + addressPort);
  Loggers.CORE.info("ADDRESS_SERVER_URL:" + addressServerUrl);
}

@SuppressWarnings("PMD.UndefineMagicConstantRule")
private void run() throws NacosException {
  // With the address server, you need to perform a synchronous member node pull at startup
  // Repeat three times, successfully jump out
  boolean success = false;
  Throwable ex = null;
  int maxRetry = ApplicationUtils.getProperty("nacos.core.address-server.retry", Integer.class, 5);
  for (int i = 0; i < maxRetry; i ++) {
    try {
      syncFromAddressUrl();
      success = true;
      break;
    } catch (Throwable e) {
      ex = e;
      Loggers.CLUSTER.error("[serverlist] exception, error : {}", ExceptionUtil.getAllExceptionMsg(ex));
    }
  }
  if (!success) {
    throw new NacosException(NacosException.SERVER_ERROR, ex);
  }

  GlobalExecutor.scheduleByCommon(new AddressServerSyncTask(), 5_000L);
}

public interface ConsistencyProtocol<T extends Config, P extends LogProcessor> extends CommandOperations {

    /**
     * Consistency protocol initialization: perform initialization operations based on the incoming Config
     * 一致性协议初始化，根据 Config 实现类
     *
     * @param config {@link Config}
     */
    void init(T config);

    /**
     * Add a log handler
     *
     * @param processors {@link LogProcessor}
     */
    void addLogProcessors(Collection<P> processors);

    /**
     * Copy of metadata information for this consensus protocol
     * 该一致性协议的元数据信息
     *
     * @return metaData {@link ProtocolMetaData}
     */
    ProtocolMetaData protocolMetaData();

    /**
     * Obtain data according to the request
     *
     * @param request request
     * @return data {@link Response}
     * @throws Exception
     */
    Response getData(GetRequest request) throws Exception;

    /**
     * Get data asynchronously
     *
     * @param request request
     * @return data {@link CompletableFuture<Response>}
     */
    CompletableFuture<Response> aGetData(GetRequest request);

    /**
     * Data operation, returning submission results synchronously
     * 同步数据提交，在 Datum 中已携带相应的数据操作信息
     *
     * @param data {@link Log}
     * @return submit operation result {@link Response}
     * @throws Exception
     */
    Response submit(Log data) throws Exception;

    /**
     * Data submission operation, returning submission results asynchronously
     * 异步数据提交，在 Datum 中已携带相应的数据操作信息，返回一个Future，自行操作，提交发生的异常会在CompleteFuture中
     *
     * @param data {@link Log}
     * @return {@link CompletableFuture<Response>} submit result
     * @throws Exception when submit throw Exception
     */
    CompletableFuture<Response> submitAsync(Log data);

    /**
     * New member list
     * 新的成员节点列表，一致性协议自行处理相应的成员节点是加入还是离开
     *
     * @param addresses [ip:port, ip:port, ...]
     */
    void memberChange(Set<String> addresses);

    /**
     * Consistency agreement service shut down
     * 一致性协议服务关闭
     */
    void shutdown();

}

public interface CPProtocol<C extends Config> extends ConsistencyProtocol<C> {

  /**
   * Returns whether this node is a leader node
   *
   * @param group business module info
   * @return is leader
   * @throws Exception
   */
  boolean isLeader(String group) throws Exception;

}

message Log {
  // 功能模块分组信息
    string group = 1;
    // 摘要或者标识
    string key = 2;
    // 具体请求数据
    bytes data = 3;
    // 数据类型
    string type = 4;
    // 更为具体的数据操作
    string operation = 5;
    // 额外信息
    map<string, string> extendInfo = 6;
}

message GetRequest {
  // 功能模块分组信息
    string group = 1;
    // 具体请求数据
    bytes data = 2;
    // 额外信息
    map<string, string> extendInfo = 3;
}

public abstract class LogProcessor {

    /**
     * get data by key
     *
     * @param request request {@link GetRequest}
     * @return target type data
     */
    public abstract Response onRequest(GetRequest request);

    /**
     * Process Submitted Log
     *
     * @param log {@link Log}
     * @return {@link boolean}
     */
    public abstract Response onApply(Log log);

    /**
     * Irremediable errors that need to trigger business price cuts
     *
     * @param error {@link Throwable}
     */
    public void onError(Throwable error) {
    }

    /**
     * In order for the state machine that handles the transaction to be able to route
     * the Log to the correct LogProcessor, the LogProcessor needs to have an identity
     * information
     *
     * @return Business unique identification name
     */
    public abstract String group();

}

public abstract class LogProcessor4CP extends LogProcessor {

    /**
     * Discovery snapshot handler
     * It is up to LogProcessor to decide which SnapshotOperate should be loaded and saved by itself
     *
     * @return {@link List <SnapshotOperate>}
     */
    public List<SnapshotOperation> loadSnapshotOperate() {
        return Collections.emptyList();
    }

}

/**
 * A concrete implementation of CP protocol: JRaft
 *
 * <pre>
 *                                           ┌──────────────────────┐
 *                                           │                      │
 *            ┌──────────────────────┐       │                      ▼
 *            │   ProtocolManager    │       │        ┌───────────────────────────┐
 *            └──────────────────────┘       │        │for p in [LogProcessor4CP] │
 *                        │                  │        └───────────────────────────┘
 *                        ▼                  │                      │
 *      ┌──────────────────────────────────┐ │                      ▼
 *      │    discovery LogProcessor4CP     │ │             ┌─────────────────┐
 *      └──────────────────────────────────┘ │             │  get p.group()  │
 *                        │                  │             └─────────────────┘
 *                        ▼                  │                      │
 *                 ┌─────────────┐           │                      │
 *                 │ RaftConfig  │           │                      ▼
 *                 └─────────────┘           │      ┌──────────────────────────────┐
 *                        │                  │      │  create raft group service   │
 *                        ▼                  │      └──────────────────────────────┘
 *              ┌──────────────────┐         │
 *              │  JRaftProtocol   │         │
 *              └──────────────────┘         │
 *                        │                  │
 *                     init()                │
 *                        │                  │
 *                        ▼                  │
 *               ┌─────────────────┐         │
 *               │   JRaftServer   │         │
 *               └─────────────────┘         │
 *                        │                  │
 *                        │                  │
 *                        ▼                  │
 *             ┌────────────────────┐        │
 *             │JRaftServer.start() │        │
 *             └────────────────────┘        │
 *                        │                  │
 *                        └──────────────────┘
 * </pre>
 *
 * @author <a href="mailto:liaochuntao@live.com">liaochuntao</a>
 */

synchronized void createMultiRaftGroup(Collection<LogProcessor4CP> processors) {
  // There is no reason why the LogProcessor cannot be processed because of the synchronization
  if (!this.isStarted) {
    this.processors.addAll(processors);
    return;
  }

  final String parentPath = Paths
        .get(ApplicationUtils.getNacosHome(), "data/protocol/raft").toString();

  for (LogProcessor4CP processor : processors) {
    final String groupName = processor.group();
    if (multiRaftGroup.containsKey(groupName)) {
      throw new DuplicateRaftGroupException(groupName);
    }

    // Ensure that each Raft Group has its own configuration and NodeOptions
    Configuration configuration = conf.copy();
    NodeOptions copy = nodeOptions.copy();
    JRaftUtils.initDirectory(parentPath, groupName, copy);

    // Here, the LogProcessor is passed into StateMachine, and when the StateMachine
    // triggers onApply, the onApply of the LogProcessor is actually called
    NacosStateMachine machine = new NacosStateMachine(this, processor);

    copy.setFsm(machine);
    copy.setInitialConf(configuration);

    // Set snapshot interval, default 1800 seconds
    int doSnapshotInterval = ConvertUtils.toInt(raftConfig
              .getVal(RaftSysConstants.RAFT_SNAPSHOT_INTERVAL_SECS),
          RaftSysConstants.DEFAULT_RAFT_SNAPSHOT_INTERVAL_SECS);

    // If the business module does not implement a snapshot processor, cancel the snapshot
    doSnapshotInterval = CollectionUtils
          .isEmpty(processor.loadSnapshotOperate()) ? 0 : doSnapshotInterval;

    copy.setSnapshotIntervalSecs(doSnapshotInterval);
    Loggers.RAFT.info("create raft group : {}", groupName);
    RaftGroupService raftGroupService = new RaftGroupService(groupName,
          localPeerId, copy, rpcServer, true);

    // Because RpcServer has been started before, it is not allowed to start again here
    Node node = raftGroupService.start(false);
    machine.setNode(node);
    RouteTable.getInstance().updateConfiguration(groupName, configuration);
    RaftExecutor.executeByCommon(() -> registerSelfToCluster(groupName, localPeerId, configuration));

    // Turn on the leader auto refresh for this group
    Random random = new Random();
    long period = nodeOptions.getElectionTimeoutMs() + random.nextInt(5 * 1000);
    RaftExecutor.scheduleRaftMemberRefreshJob(() -> refreshRouteTable(groupName),
          nodeOptions.getElectionTimeoutMs(), period, TimeUnit.MILLISECONDS);
    multiRaftGroup.put(groupName,
          new RaftGroupTuple(node, processor, raftGroupService, machine));
  }
}

POST /nacos/v1/core/ops/raft
{
    "groupId": "xxx",
    "command": "transferLeader"
    "value": "ip:{raft_port} or ip:{raft_port},ip:{raft_port},ip:{raft_port}"
}

POST /nacos/v1/core/ops/raft
{
    "groupId": "xxx",
    "command": "resetRaftCluster",
    "value": "ip:{raft_port},ip:{raft_port},ip:{raft_port},ip:{raft_port}"
}

POST /nacos/v1/core/ops/raft
{
    "groupId": "xxx",
    "command": "doSnapshot",
    "value": "ip:{raft_port}"
}

POST /nacos/v1/core/ops/raft
{
    "groupId": "xxx",
    "command": "removePeer",
    "value": "ip:{raft_port}"
}

POST /nacos/v1/core/ops/raft
{
    "groupId": "xxx",
    "command": "removePeers",
    "value": "ip:{raft_port},ip:{raft_port},ip:{raft_port},..."
}

### Sets the Raft cluster election timeout, default value is 5 second
nacos.core.protocol.raft.data.election_timeout_ms=5000
### Sets the amount of time the Raft snapshot will execute periodically, default is 30 minute
nacos.core.protocol.raft.data.snapshot_interval_secs=30
### Requested retries, default value is 1
nacos.core.protocol.raft.data.request_failoverRetries=1
### raft internal worker threads
nacos.core.protocol.raft.data.core_thread_num=8
### Number of threads required for raft business request processing
nacos.core.protocol.raft.data.cli_service_thread_num=4
### raft linear read strategy, defaults to index
nacos.core.protocol.raft.data.read_index_type=ReadOnlySafe
### rpc request timeout, default 5 seconds
nacos.core.protocol.raft.data.rpc_request_timeout_ms=5000

InetAddress address;
try {
  address = InetAddress.getLocalHost();
} catch (final UnknownHostException e) {
  throw new IllegalStateException(
            "Cannot get LocalHost InetAddress, please check your network!");
}
byte[] ipAddressByteArray = address.getAddress();
workerId = (((ipAddressByteArray[ipAddressByteArray.length - 2] & 0B11)
          << Byte.SIZE) + (ipAddressByteArray[ipAddressByteArray.length - 1]
          & 0xFF));

### set the dataCenterID manually
# nacos.core.snowflake.data-center=
### set the WorkerID manually
# nacos.core.snowflake.worker-id=

private void registerSubscribe() {
  NotifyCenter.registerSubscribe(new SmartSubscribe() {

    @Override
    public void onEvent(Event event) {
      if (event instanceof RaftDBErrorRecoverEvent) {
        downgrading = false;
        return;
      }
      if (event instanceof RaftDBErrorEvent) {
        downgrading = true;
      }
    }

    @Override
    public boolean canNotify(Event event) {
      return (event instanceof RaftDBErrorEvent) || (event instanceof RaftDBErrorRecoverEvent);
    }
  });
}

public class SelectRequest implements Serializable {

    private static final long serialVersionUID = 2212052574976898602L;
    // 查询类别，因为目前使用的是JdbcTemplate，查询单个、查询多个，是否使用RowMapper转为对象
    private byte queryType;
    // sql语句
    // select * from config_info where
    private String sql;
    private Object[] args;
    private String className;
}

public class ModifyRequest implements Serializable {

    private static final long serialVersionUID = 4548851816596520564L;

    private int executeNo;
    private String sql;
    private Object[] args;
}

public void addConfigInfo(final String srcIp,
      final String srcUser, final ConfigInfo configInfo, final Timestamp time,
      final Map<String, Object> configAdvanceInfo, final boolean notify) {

  try {
    final String tenantTmp = StringUtils.isBlank(configInfo.getTenant()) ?
          StringUtils.EMPTY :
          configInfo.getTenant();
    configInfo.setTenant(tenantTmp);

        // 通过雪花ID算法获取数据库主键
    long configId = idGeneratorManager.nextId(RESOURCE_CONFIG_INFO_ID);
    long hisId = idGeneratorManager.nextId(RESOURCE_CONFIG_HISTORY_ID);

    addConfigInfoAtomic(configId, srcIp, srcUser, configInfo, time,
          configAdvanceInfo);
    String configTags = configAdvanceInfo == null ?
          null :
          (String) configAdvanceInfo.get("config_tags");

    addConfigTagsRelation(configId, configTags, configInfo.getDataId(),
          configInfo.getGroup(), configInfo.getTenant());
    insertConfigHistoryAtomic(hisId, configInfo, srcIp, srcUser, time, "I");
    EmbeddedStorageContextUtils.onModifyConfigInfo(configInfo, srcIp, time);
    databaseOperate.blockUpdate();
  }
  finally {
    EmbeddedStorageContextUtils.cleanAllContext();
  }
}

public long addConfigInfoAtomic(final long id, final String srcIp,
      final String srcUser, final ConfigInfo configInfo, final Timestamp time,
      Map<String, Object> configAdvanceInfo) {
  ...
    // 参数处理
    ...
  final String sql =
        "INSERT INTO config_info(id, data_id, group_id, tenant_id, app_name, content, md5, src_ip, src_user, gmt_create,"
            + "gmt_modified, c_desc, c_use, effect, type, c_schema) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)";
  final Object[] args = new Object[] { id, configInfo.getDataId(),
        configInfo.getGroup(), tenantTmp, appNameTmp, configInfo.getContent(),
        md5Tmp, srcIp, srcUser, time, time, desc, use, effect, type, schema, };
  SqlContextUtils.addSqlContext(sql, args);
  return id;
}

public void addConfigTagRelationAtomic(long configId, String tagName, String dataId,
      String group, String tenant) {
  final String sql =
        "INSERT INTO config_tags_relation(id,tag_name,tag_type,data_id,group_id,tenant_id) "
            + "VALUES(?,?,?,?,?,?)";
  final Object[] args = new Object[] { configId, tagName, null, dataId, group,
        tenant };
  SqlContextUtils.addSqlContext(sql, args);
}

public void insertConfigHistoryAtomic(long configHistoryId, ConfigInfo configInfo,
      String srcIp, String srcUser, final Timestamp time, String ops) {
  ...
    // 参数处理
    ...
  final String sql =
        "INSERT INTO his_config_info (id,data_id,group_id,tenant_id,app_name,content,md5,"
            + "src_ip,src_user,gmt_modified,op_type) VALUES(?,?,?,?,?,?,?,?,?,?,?)";
  final Object[] args = new Object[] { configHistoryId, configInfo.getDataId(),
        configInfo.getGroup(), tenantTmp, appNameTmp, configInfo.getContent(),
        md5Tmp, srcIp, srcUser, time, ops };

  SqlContextUtils.addSqlContext(sql, args);
}

/**
 * Temporarily saves all insert, update, and delete statements under
 * a transaction in the order in which they occur
 *
 * @author <a href="mailto:liaochuntao@live.com">liaochuntao</a>
 */
public class SqlContextUtils {

    private static final ThreadLocal<ArrayList<ModifyRequest>> SQL_CONTEXT =
            ThreadLocal.withInitial(ArrayList::new);

    public static void addSqlContext(String sql, Object... args) {
        ArrayList<ModifyRequest> requests = SQL_CONTEXT.get();
        ModifyRequest context = new ModifyRequest();
        context.setExecuteNo(requests.size());
        context.setSql(sql);
        context.setArgs(args);
        requests.add(context);
        SQL_CONTEXT.set(requests);
    }

    public static List<ModifyRequest> getCurrentSqlContext() {
        return SQL_CONTEXT.get();
    }

    public static void cleanCurrentSqlContext() {
        SQL_CONTEXT.remove();
    }

}

GET /nacos/v1/cs/ops/derby?sql=select * from config_info

return List<Map<String, Object>>