【Canal源码分析】Canal Server的启动和停止过程

本文主要解析下canal server的启动过程，希望能有所收获。

一、序列图

1.1 启动

1.2 停止

二、源码分析

整个server启动的过程比较复杂，看图难以理解，需要辅以文字说明。

首先程序的入口在CanalLauncher的main方法中。

2.1 加载配置文件

String conf = System.getProperty("canal.conf", "classpath:canal.properties");
Properties properties = new Properties();
if (conf.startsWith(CLASSPATH_URL_PREFIX)) {
    conf = StringUtils.substringAfter(conf, CLASSPATH_URL_PREFIX);
    properties.load(CanalLauncher.class.getClassLoader().getResourceAsStream(conf));
} else {
    properties.load(new FileInputStream(conf));
}

从canal.properties文件中load所有的配置信息，加载到上下文中。不再赘述。

2.2 构造CanalController

根据配置文件来构造CanalController，这块的代码比较多，主要分为七个步骤，具体如下。

2.2.1 初始化全局参数配置

调用initGlobalConfig方法，过程如下：

  • 判断运行模式，是从spring加载还是manager加载，目前开源版本建议使用spring
  • 获取是否懒加载
  • 如果是manager模式启动，获取manager的ip地址；如果是spring模式启动，获取spring xml的文件地址，加载到全部配置中
  • 构造一个实例构造器CanalInstanceGenerator，我们用到的就是在spring的beanFactory中加上destination的bean，这个destination就是canal instance的名称

这块逻辑在CanalController的initGlobalConfig方法中。

2.2.2 初始化实例配置

这块的逻辑是从instance.properties里面初始化实例。

private void initInstanceConfig(Properties properties) {
    String destinationStr = getProperty(properties, CanalConstants.CANAL_DESTINATIONS);
    String[] destinations = StringUtils.split(destinationStr, CanalConstants.CANAL_DESTINATION_SPLIT);

    for (String destination : destinations) {
        InstanceConfig config = parseInstanceConfig(properties, destination);
        InstanceConfig oldConfig = instanceConfigs.put(destination, config);

        if (oldConfig != null) {
            logger.warn("destination:{} old config:{} has replace by new config:{}", new Object[] { destination,
                    oldConfig, config });
        }
    }
}

从这段代码中可以看出，我们在一个canal.properties文件中，可以配置多个destination，也就是可以配置多个instance，不同的instance以逗号隔开。这里主要看的是parseInstanceConfig()方法，里面的逻辑如下：

  • 获取启动模式，是manager还是spring，我们这边默认都是spring。
  • 获取懒加载字段
  • 获取spring xml配置文件地址

2.2.3 初始SocketChannel

从配置文件中获取canal.socketChannel字段，放到全局变量中。

2.2.4 准备canal server

从配置文件中分别获取canal.id、ip、port（对外提供socket服务的端口），获取一个内存级的server单例，同时也获取一个对外提供Netty服务的单例。

cid = Long.valueOf(getProperty(properties, CanalConstants.CANAL_ID));
ip = getProperty(properties, CanalConstants.CANAL_IP);
port = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_PORT));
embededCanalServer = CanalServerWithEmbedded.instance();
embededCanalServer.setCanalInstanceGenerator(instanceGenerator);// 设置自定义的instanceGenerator
canalServer = CanalServerWithNetty.instance();
canalServer.setIp(ip);
canalServer.setPort(port);

2.2.5 初始化系统目录

从配置文件中获取zk地址(canal.zkServers)，启动一个zk客户端，然后初始化两个系统目录，分别是：

  • /otter/canal/destinations
  • /otter/canal/cluster

2.2.6 初始化系统监控

根据destination构造运行时监控，其实就是根据instance名来构造ServerRunningMonitor。其实就是实现了ServerRunningListener中的一些方法。

public interface ServerRunningListener {

    /**
     * 启动时回调做点事情
     */
    public void processStart();

    /**
     * 关闭时回调做点事情
     */
    public void processStop();

    /**
     * 触发现在轮到自己做为active，需要载入上一个active的上下文数据
     */
    public void processActiveEnter();

    /**
     * 触发一下当前active模式失败
     */
    public void processActiveExit();

}

然后初始化一下ServerRunningMonitor。

runningMonitor.init();

这个init方法跟踪的结果，其实就是执行了ServerRunningListener中的processStart方法。

public void processStart() {
    try {
        if (zkclientx != null) {
            final String path = ZookeeperPathUtils.getDestinationClusterNode(destination, ip + ":" + port);
            initCid(path);
            zkclientx.subscribeStateChanges(new IZkStateListener() {

                public void handleStateChanged(KeeperState state) throws Exception {

                }

                public void handleNewSession() throws Exception {
                    initCid(path);
                }

                @Override
                public void handleSessionEstablishmentError(Throwable error) throws Exception {
                    logger.error("failed to connect to zookeeper", error);
                }
            });
        }
    } finally {
        MDC.remove(CanalConstants.MDC_DESTINATION);
    }
}

首先获取了/otter/canal/destinations/{destination}/cluster/ip:port的内容，其实就是server的地址，最后一个ip:port是个zk的临时节点。然后订阅一下节点事件，当节点有事件推送过来后，做一些动作。

2.2.7 初始化配置文件监控

如果canal.auto.scan配置为true（默认为true），首先定义一个InstanceAction，包含了启动、停止、重启instance的动作。

定义一个SpringInstanceConfigMonitor，配置定时扫描的事件为canal.auto.scan.interval，默认5s，扫描canal.conf.dir目录下的文件，与上面定义的InstanceAction结合起来。

2.3 启动CanalController

上面的构造方法其实就是定义一些必要的内容，真正的启动在这个方法中。

2.3.1 创建工作节点

创建临时节点/otter/canal/cluster/ip:port，同时启动监听器.

2.3.2 启动embeded服务

embededCanalServer.start();

这个start里面，一个是将当前server的running状态置为true，同时根据destination构建CanalInstance。

2.3.3 HA启动

遍历Map

ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
if (!config.getLazy() && !runningMonitor.isStart()) {
    runningMonitor.start();
}

public synchronized void start() {
    super.start();
    try {
        processStart();
        if (zkClient != null) {
            // 如果需要尽可能释放instance资源，不需要监听running节点，不然即使stop了这台机器，另一台机器立马会start
            String path = ZookeeperPathUtils.getDestinationServerRunning(destination);
            zkClient.subscribeDataChanges(path, dataListener);

            initRunning();
        } else {
            processActiveEnter();// 没有zk，直接启动
        }
    } catch (Exception e) {
        logger.error("start failed", e);
        // 没有正常启动，重置一下状态，避免干扰下一次start
        stop();
    }

}

这里面启动的内容我们来看看。

  • 首先调用super.start()把当前的running状态置为true。
  • 然后启动zk节点的监听（这边的processStart是否多余了？）。
  • 监听路径/otter/canal/destinations/{destination}/running节点的变化 
    zkClient.subscribeDataChanges(path, dataListener);
  • 这里的dataListener是ServerRunningMonitor构造函数中定义的，就是定义一些zk节点监听的动作。
    • 如果有数据变化，如果running节点中的内容ServerRunningData发生了变化，字段active变为了false，而且address就是本机，说明本机出现了主动释放，需要释放运行时状态。此时需要调用到processActiveExit方法，其实就是停止了本机的server中destination对应的instance。
    • 如果节点发生了删除动作，如果上一次active的状态就是本机，则即时触发一下active抢占，调用initRunning()方法，当然，如果启动失败，也不是立即切换，而是会等待5s，再尝试启动。这个启动方法中，主要调用的是processActiveEnter()方法，来启动了embededCanalServer.start(destination)。其实就是启动canalInstance，这块后续再分析。
  • 其实除了监听器，在本身的ServerRunningMonitor的start方法中，也有initRunning方法。这块启动canalInstance的方法，我们下一篇文章分析。

2.3.4 instance文件扫描启动

在扫描之前，把destination和InstanceAction绑定到缓存中。

instanceConfigMonitors.get(config.getMode()).register(destination, defaultAction);

首先启动一个全局扫描，然后再对应的destination配置文件的扫描。

if (autoScan) {
    instanceConfigMonitors.get(globalInstanceConfig.getMode()).start();
    for (InstanceConfigMonitor monitor : instanceConfigMonitors.values()) {
        if (!monitor.isStart()) {
            monitor.start();
        }
    }
}

这个start方法启动了一个定时器，默认5s扫描一次。扫描的内容就是配置文件路径下的内容，针对文件的新增、删除、修改，对应InstanceAction中的start，stop和reload方法。也就是说，我们在canal运行的过程中，通过动态修改配置文件，来实现动态调整运行时参数，主要可以用来进行重复消费，位点的迁移等等。

2.3.5 网络接口启动

CanalServerWithNetty的启动，首先需要启动CanalServerWithEmbedded，主要的业务逻辑在SessionHandler中。这块其实是暴露外部服务，给canal client进行调用。

2.4 增加关闭hook

Runtime.getRuntime().addShutdownHook(new Thread() {

    public void run() {
        try {
            logger.info("## stop the canal server");
            controller.stop();
        } catch (Throwable e) {
            logger.warn("##something goes wrong when stopping canal Server:", e);
        } finally {
            logger.info("## canal server is down.");
        }
    }

});

在server停止时，调用controller.stop()方法。

public void stop() throws Throwable {
    canalServer.stop();

    if (autoScan) {
        for (InstanceConfigMonitor monitor : instanceConfigMonitors.values()) {
            if (monitor.isStart()) {
                monitor.stop();
            }
        }
    }

    for (ServerRunningMonitor runningMonitor : ServerRunningMonitors.getRunningMonitors().values()) {
        if (runningMonitor.isStart()) {
            runningMonitor.stop();
        }
    }

    // 释放canal的工作节点
    releaseCid(ZookeeperPathUtils.getCanalClusterNode(ip + ":" + port));
    logger.info("## stop the canal server[{}:{}]", ip, port);

    if (zkclientx != null) {
        zkclientx.close();
    }
}

主要是停止controller，server相关的monitor，instance相关的monitor，然后释放zk节点，关闭zk连接。

原文地址：https://www.cnblogs.com/f-zhao/p/9083099.html

时间： 2024-08-04 05:51:38

【Canal源码分析】Canal Server的启动和停止过程的相关文章

Kubernetes 源码分析 -- API Server之编解码

--------------------- 作者:weixin_34037977 来源:CSDN 原文:https://blog.csdn.net/weixin_34037977/article/details/87058105 在Kubernetes源码分析-- API Server之API Install篇中,我们了解到K8S可以支持多版本的API,但是Rest API的不同版本中接口的输入输出参数的格式是有差别的,Kubernetes是怎么处理这个问题的呢?另外Kubernetes支持ya

Solr初始化源码分析-Solr初始化与启动

用solr做项目已经有一年有余,但都是使用层面,只是利用solr现有机制,修改参数,然后监控调优,从没有对solr进行源码级别的研究.但是,最近手头的一个项目,让我感觉必须把solrn内部原理和扩展机制弄熟,才能把这个项目做好.今天分享的就是:Solr是如何启动并且初始化的.大家知道,部署solr时,分两部分:一.solr的配置文件.二.solr相关的程序.插件.依赖lucene相关的jar包.日志方面的jar.因此,在研究solr也可以顺着这个思路:加载配置文件.初始化各个core.初始化各个

源码分析_Shinken-2.4.0001.启动脚本/etc/init.d/shinken源码分析?

简单介绍:说明: Shinken是一个网络监控平台,可以通过一系列直观的方式监控网络内的各种健康状况.Shinken脱胎于Nagios,其实Shinken这个项目本身就是一帮Nagios项目的人无法忍受Nagios,自己跳出来重新用纯Python重构了一下,甚至完全兼容Nagios的配置文件. 相关地址: 官网地址: http://www.shinken-monitoring.org/ 官网文档: http://shinken.readthedocs.io/en/latest/ 论坛地址: ht

kafka源码分析之一server启动分析

1. 分析kafka源码的目的 深入掌握kafka的内部原理 深入掌握scala运用 2. server的启动 如下所示(本来准备用时序图的,但感觉时序图没有思维图更能反映,故采用了思维图): 2.1 启动入口Kafka.scala 从上面的思维导图,可以看到Kafka的启动入口是Kafka.scala的main()函数: def main(args: Array[String]): Unit = { try { val serverProps = getPropsFromArgs(args)

Tomcat 源码分析（一）——启动与生命周期组件

写在前面的话:读Tomcat源码也有段时间了,大领悟谈不上.一些小心得记录下来,供大家参考相护学习. 一.启动流程 Tomcat启动首先需要熟悉的是它的启动流程.和初学者第一天开始写Hello World一样,Tomcat的启动也依赖main方法. 1 /* 2 * org.apache.catalina.startup.Bootstrap 3 */ 4 if (daemon == null) { 5 Bootstrap bootstrap = new Bootstrap(); // 实例对象

netty源码分析之服务端启动

ServerBootstrap与Bootstrap分别是netty中服务端与客户端的引导类,主要负责服务端与客户端初始化.配置及启动引导等工作,接下来我们就通过netty源码中的示例对ServerBootstrap与Bootstrap的源码进行一个简单的分析.首先我们知道这两个类都继承自AbstractBootstrap类 接下来我们就通过netty源码中ServerBootstrap的实例入手对其进行一个简单的分析. // Configure the server. EventLoopGrou

DroidPlugin源码分析处理Activity的启动211ew4dfr

正常情况下启动一个Activity,首先需要在AndroidManifest文件中声明,其次需要把该应用安装到手机系统中. 而插件apk是没有正在安装到手机系统中的,也就按照正常的启动流程插件Activity是不能启动的.另外插件apk的类需要加载进来是需要指定ClassLoader.前面的文章也大概讲过,当启动一个插件Activity时,先是用预定义的代理Activity替换目标Activity(及插件Activity)去启动,当AMS处理完回调到应用空间时(及回到运行Activity的进程空

DroidPlugin源码分析处理Activity的启动

正常情况下启动一个Activity,首先需要在AndroidManifest文件中声明,其次需要把该应用安装到手机系统中. 而插件apk是没有正在安装到手机系统中的,也就按照正常的启动流程插件Activity是不能启动的.另外插件apk的类需要加载进来是需要指定ClassLoader.前面的文章也大概讲过,当启动一个插件Activity时,先是用预定义的代理Activity替换目标Activity(及插件Activity)去启动,当AMS处理完回调到应用空间时(及回到运行Activity的进程空

SpringBoot源码分析之---SpringBoot项目启动类SpringApplication浅析

源码版本说明 本文源码采用版本为SpringBoot 2.1.0BUILD,对应的SpringFramework 5.1.0.RC1 注意:本文只是从整体上梳理流程,不做具体深入分析 SpringBoot入口类 @SpringBootApplication public class DemoApplication { public static void main(String[] args) { SpringApplication.run(DemoApplication.class, args