聊聊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

原创声明,本文系作者授权云+社区发表,未经许可,不得转载。

如有侵权,请联系 yunjia_community@tencent.com 删除。

编辑于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏IT那个小笔记

Hibernate核心API

Configuration相当于是Hibernate的配置对象,通过这个对象去加载核心配置文件,把里面的数据内容或得到,才能进行后面的工作

10520
来自专栏Java技术栈

牛逼哄哄的 "零拷贝" 是什么?

缓冲区是所有I/O的基础,I/O讲的无非就是把数据移进或移出缓冲区;进程执行I/O操作,就是向操作系统发出请求,让它要么把缓冲区的数据排干(写),要么填充缓冲区...

8610
来自专栏智能计算时代

「集成架构」ETL工具大比拼:Talend vs Pentaho

数据总是巨大的,任何行业都必须存储这些“数据”,因为它带有巨大的信息,从而导致他们的战略规划。正如人们需要房子感到安全一样,数据也必须得到保障。这个数据主页在技...

28720
来自专栏Java程序猿部落

ArrayList、LinkedList和Vector的源码解析,带你走近List的世界

java.util.List接口是Java Collections Framework的一个重要组成部分,List接口的架构图如下:

7620
来自专栏程序猿杂货铺

业务代码的救星——Java 对象转换框架 MapStruct 妙用

在业务项目的开发中,我们经常需要将 Java 对象进行转换,比如从将外部微服务得到的对象转换为本域的业务对象 domainobject,将 domainobje...

14540
来自专栏Java研发军团

挑逗 Java 程序员的那些 Scala 绝技

昨天,看到一篇介绍 Scala 技巧的文章,作者的语言很风趣,从 val,字符串,集合,链式调用等多个角度来探讨这门语言的优雅之处,使得我们更容易接受它,并愿意...

8420
来自专栏Java研发军团

手把手教你写一个网关服务

如图所示,在不使用网关的情况下,我们的服务是直接暴露给服务调用方。当调用方增多,势必需要添加定制化访问权限、校验等逻辑。当添加 API 网关后,再第三方调用端和...

13930
来自专栏Java技术栈

一个 Java 字符串到底有多少个字符?

依照Java的文档, Java中的字符内部是以UTF-16编码方式表示的,最小值是 \u0000 (0),最大值是\uffff(65535), 也就是一个字符以...

12710
来自专栏java技术大本营

JVM|04垃圾回收

程序的运行必然需要申请内存资源,无效的对象资源如果不及时处理就会一直占有内存资源,最终将导致内存溢出,所以对内存资源的管理是非常重要了。

10920
来自专栏java技术大本营

Netty|01 入门学习

1、Neetty 是由 JBOSS 提供的一个 Java 开源框架。Netty 提供异步的、基于事件驱动的网络 应用程序框架,用以快速开发高性能、高可靠性的网络...

11530

扫码关注云+社区

领取腾讯云代金券

年度创作总结 领取年终奖励