聊聊nacos的MysqlHealthCheckProcessor

序

本文主要研究一下 nacos 的 MysqlHealthCheckProcessor

MysqlHealthCheckProcessor

nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/healthcheck/MysqlHealthCheckProcessor.java

@Component
public class MysqlHealthCheckProcessor implements HealthCheckProcessor {

    @Autowired
    private HealthCheckCommon healthCheckCommon;

    @Autowired
    private SwitchDomain switchDomain;

    public static final int CONNECT_TIMEOUT_MS = 500;

    private static final String CHECK_MYSQL_MASTER_SQL = "show global variables where variable_name='read_only'";
    private static final String MYSQL_SLAVE_READONLY = "ON";

    private static ConcurrentMap<String, Connection> CONNECTION_POOL
            = new ConcurrentHashMap<String, Connection>();

    private static ExecutorService EXECUTOR;

    static {int processorCount = Runtime.getRuntime().availableProcessors();
        EXECUTOR
                = Executors.newFixedThreadPool(processorCount <= 1 ? 1 : processorCount / 2,
                new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {Thread thread = new Thread(r);
                        thread.setDaemon(true);
                        thread.setName("com.nacos.mysql.checker");
                        return thread;
                    }
                }
        );
    }

    public MysqlHealthCheckProcessor() {}

    @Override
    public String getType() {return "MYSQL";}

    @Override
    public void process(HealthCheckTask task) {List<Instance> ips = task.getCluster().allIPs(false);

        SRV_LOG.debug("mysql check, ips:" + ips);
        if (CollectionUtils.isEmpty(ips)) {return;}

        for (Instance ip : ips) {
            try {if (ip.isMarked()) {if (SRV_LOG.isDebugEnabled()) {SRV_LOG.debug("mysql check, ip is marked as to skip health check, ip: {}", ip.getIp());
                    }
                    continue;
                }

                if (!ip.markChecking()) {SRV_LOG.warn("mysql check started before last one finished, service: {}:{}:{}",
                        task.getCluster().getService().getName(), task.getCluster().getName(), ip.getIp());

                    healthCheckCommon.reEvaluateCheckRT(task.getCheckRTNormalized() * 2, task, switchDomain.getMysqlHealthParams());
                    continue;
                }

                EXECUTOR.execute(new MysqlCheckTask(ip, task));
                MetricsMonitor.getMysqlHealthCheckMonitor().incrementAndGet();
            } catch (Exception e) {ip.setCheckRT(switchDomain.getMysqlHealthParams().getMax());
                healthCheckCommon.checkFail(ip, task, "mysql:error:" + e.getMessage());
                healthCheckCommon.reEvaluateCheckRT(switchDomain.getMysqlHealthParams().getMax(), task, switchDomain.getMysqlHealthParams());
            }
        }
    }

    //......
}

• MysqlHealthCheckProcessor 实现了 HealthCheckProcessor 接口，其 getType 方法返回的是 MYSQL；其 process 方法会遍历 instances，对于非 markChecking 的会执行 healthCheckCommon.reEvaluateCheckRT，对于 marked 的直接跳过，对于 markChecking 的会创建 MysqlCheckTask 提交给 EXECUTOR 执行

MysqlCheckTask

nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/healthcheck/MysqlHealthCheckProcessor.java

    private class MysqlCheckTask implements Runnable {
        private Instance ip;
        private HealthCheckTask task;
        private long startTime = System.currentTimeMillis();

        public MysqlCheckTask(Instance ip, HealthCheckTask task) {
            this.ip = ip;
            this.task = task;
        }

        @Override
        public void run() {

            Statement statement = null;
            ResultSet resultSet = null;

            try {Cluster cluster = task.getCluster();
                String key = cluster.getService().getName() + ":" + cluster.getName() + ":" + ip.getIp() + ":" + ip.getPort();
                Connection connection = CONNECTION_POOL.get(key);
                AbstractHealthChecker.Mysql config = (AbstractHealthChecker.Mysql) cluster.getHealthChecker();

                if (connection == null || connection.isClosed()) {MysqlDataSource dataSource = new MysqlDataSource();
                    dataSource.setConnectTimeout(CONNECT_TIMEOUT_MS);
                    dataSource.setSocketTimeout(CONNECT_TIMEOUT_MS);
                    dataSource.setUser(config.getUser());
                    dataSource.setPassword(config.getPwd());
                    dataSource.setLoginTimeout(1);

                    dataSource.setServerName(ip.getIp());
                    dataSource.setPort(ip.getPort());

                    connection = dataSource.getConnection();
                    CONNECTION_POOL.put(key, connection);
                }

                statement = connection.createStatement();
                statement.setQueryTimeout(1);

                resultSet = statement.executeQuery(config.getCmd());
                int resultColumnIndex = 2;

                if (CHECK_MYSQL_MASTER_SQL.equals(config.getCmd())) {resultSet.next();
                    if (MYSQL_SLAVE_READONLY.equals(resultSet.getString(resultColumnIndex))) {throw new IllegalStateException("current node is slave!");
                    }
                }

                healthCheckCommon.checkOK(ip, task, "mysql:+ok");
                healthCheckCommon.reEvaluateCheckRT(System.currentTimeMillis() - startTime, task, switchDomain.getMysqlHealthParams());
            } catch (SQLException e) {
                // fail immediately
                healthCheckCommon.checkFailNow(ip, task, "mysql:" + e.getMessage());
                healthCheckCommon.reEvaluateCheckRT(switchDomain.getHttpHealthParams().getMax(), task, switchDomain.getMysqlHealthParams());
            } catch (Throwable t) {
                Throwable cause = t;
                int maxStackDepth = 50;
                for (int deepth = 0; deepth < maxStackDepth && cause != null; deepth++) {
                    if (cause instanceof SocketTimeoutException
                            || cause instanceof ConnectTimeoutException
                            || cause instanceof TimeoutException
                            || cause.getCause() instanceof TimeoutException) {healthCheckCommon.checkFail(ip, task, "mysql:timeout:" + cause.getMessage());
                        healthCheckCommon.reEvaluateCheckRT(task.getCheckRTNormalized() * 2, task, switchDomain.getMysqlHealthParams());
                        return;
                    }

                    cause = cause.getCause();}

                // connection error, probably not reachable
                healthCheckCommon.checkFail(ip, task, "mysql:error:" + t.getMessage());
                healthCheckCommon.reEvaluateCheckRT(switchDomain.getMysqlHealthParams().getMax(), task, switchDomain.getMysqlHealthParams());
            } finally {ip.setCheckRT(System.currentTimeMillis() - startTime);
                if (statement != null) {
                    try {statement.close();
                    } catch (SQLException e) {Loggers.SRV_LOG.error("[MYSQL-CHECK] failed to close statement:" + statement, e);
                    }
                }
                if (resultSet != null) {
                    try {resultSet.close();
                    } catch (SQLException e) {Loggers.SRV_LOG.error("[MYSQL-CHECK] failed to close resultSet:" + resultSet, e);
                    }
                }
            }
        }
    }

• MysqlCheckTask 实现了 Runnable 方法，其 run 方法会从 CONNECTION_POOL 获取指定实例的 connection，如果 connection 为 null 或者是 closed 的则重新创建 MysqlDataSource 并 getConnection() 放入到 CONNECTION_POOL；获取到 connection 之后会执行 config.getCmd() 指定的查询，然后根据返回结果或异常情况执行 healthCheckCommon.checkOK 或者 healthCheckCommon.checkFailNow、healthCheckCommon.checkFail 操作

小结

MysqlHealthCheckProcessor 实现了 HealthCheckProcessor 接口，其 getType 方法返回的是 MYSQL；其 process 方法会遍历 instances，对于非 markChecking 的会执行 healthCheckCommon.reEvaluateCheckRT，对于 marked 的直接跳过，对于 markChecking 的会创建 MysqlCheckTask 提交给 EXECUTOR 执行

doc

• MysqlHealthCheckProcessor