canal调度控制器CanalController源码分析

canal调度控制器初始化：

public CanalController(final Properties properties)

1.? 初始化instance公共全局配置： canal.instance.global.(mode、lazy、manager.address以及spring.xml，并放入内存 InstanceConfig globalInstanceConfig;

canal.instance.global.mode:?用于确定canal instance的全局配置加载方式，主要有两种方式一种是spring本地模式，一种是admin管理端配置， 如果指定了manager地址（canal.admin.manager）,则强制使用manager模式，spring模式的是SpringCanalInstanceGenerator，通过参数化将instance名称带入路径classpath:${canal.instance.destination:}/instance.properties，获取本地resource目录下配置文件

 private InstanceConfig initGlobalConfig(Properties properties) {
   .......
}

2. 初始化每个instance独有的配置，通过destinations配置多个instance，并放入内存Map<String, InstanceConfig> instanceConfigs;

    private void initInstanceConfig(Properties properties) {
        // 对instance进行逗号分割 存在多个instance的情况下
        String destinationStr = getDestinations(properties);
        String[] destinations = StringUtils.split(destinationStr, CanalConstants.CANAL_DESTINATION_SPLIT);

        for (String destination : destinations) {
            // 可针对instance进行单独配置 mode 、Spring、lazy
            InstanceConfig config = parseInstanceConfig(properties, destination);
            InstanceConfig oldConfig = instanceConfigs.put(destination, config);

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

3. 初始化canal server，server主要有两种方式，官方描述如下：

我们这里使用的是独立部署，会使用两种，均是单利模式，CanalServerWithNetty会将客户端请求派给CanalServerWithEmbeded 进行真正的处理，CanalServerWithNetty接收来自canal client的请求。

        // 嵌入式服务实例化
        embeddedCanalServer = CanalServerWithEmbedded.instance();
        embeddedCanalServer.setCanalInstanceGenerator(instanceGenerator);// 设置自定义的instanceGenerator
        .....此处代码省略
        // canal.withoutNetty 配置
        String canalWithoutNetty = getProperty(properties, CanalConstants.CANAL_WITHOUT_NETTY);
        if (canalWithoutNetty == null || "false".equals(canalWithoutNetty)) {
            // server 初始化
            canalServer = CanalServerWithNetty.instance();
            canalServer.setIp(ip);
            canalServer.setPort(port);
        }

4. 初始化canal server集群模式，根据canal.zkServers配置的zk地址，是否走HA模式，?并进行初始化目录，创建永久节点（集群节点以及instance节点）

        // zk 集群地址
        final String zkServers = getProperty(properties, CanalConstants.CANAL_ZKSERVERS);
        if (StringUtils.isNotEmpty(zkServers)) {
            zkclientx = ZkClientx.getZkClient(zkServers);
            // 初始化系统目录 /otter/canal/destinations
            zkclientx.createPersistent(ZookeeperPathUtils.DESTINATION_ROOT_NODE, true);
            //  /otter/canal/cluster  整个集群信息
            zkclientx.createPersistent(ZookeeperPathUtils.CANAL_CLUSTER_ROOT_NODE, true);
        }

每个canal-server对每个instance的管理是交给ServerRunningMonitor类，监控运行状态，有变更的时候会进行相应的变更处理。只有当后面开启自动化扫描，才会进行初始化，每个instance对应一个ServerRunningMonitor。

 
//ServerRunningMonitors.getRunningMonitor被调用的时候，先对map进行查找，没有的话进行以下
//代码生成 ServerRunningMonitors.setRunningMonitors(MigrateMap.makeComputingMap((Function<String, ServerRunningMonitor>) destination -> {
            ServerRunningMonitor runningMonitor = new ServerRunningMonitor(serverData);
            runningMonitor.setDestination(destination);
            // 回调所做的事情 启动前以及启动后
            runningMonitor.setListener(new ServerRunningListener() {

                public void processActiveEnter() {...}

                public void processActiveExit() {...}

                public void processStart() {...}

                public void processStop() {...}

            });
            if (zkclientx != null) {
                runningMonitor.setZkClient(zkclientx);
            }
            // 触发创建一下cid节点
            runningMonitor.init();
            return runningMonitor;
        }));

5.? 初始化instance自动化扫描，通过配置canal.auto.scan=true 进行开启自动化扫描。

defaultAction：其作用是如果配置发生了变更，默认应该采取什么样的操作。实现了InstanceAction接口定义的三个抽象方法：start、stop和reload。当新增一个destination配置时，需要调用start方法来启动一个新instance，当移除一个destination配置时，需要调用stop方法来停止当前instance；当某个destination配置发生变更时，需要调用reload方法来进行重启。instanceConfigMonitors： 监听配置变更，Spring模式定时扫描默认是user.dir + conf目录组成，manager模式通过PlainCanalConfigClient http 方式获取admin端管理的配置。

//当instance发现变更的时候，默认应该采取什么样的操作        
autoScan = BooleanUtils.toBoolean(getProperty(properties, CanalConstants.CANAL_AUTO_SCAN));
        if (autoScan) {
            defaultAction = new InstanceAction() {
                public void start(String destination) {...}

                public void stop(String destination) {...}

                public void reload(String destination) {...}

                @Overrid
                public void release(String destination) {....}   
              }                               
            };
            // 监听配置
            instanceConfigMonitors = MigrateMap.makeComputingMap(mode -> {
//扫描间隔时间                
int scanInterval = Integer.valueOf(getProperty(properties,
                    CanalConstants.CANAL_AUTO_SCAN_INTERVAL,
                    "5"));

                     // spring 模式
                     if (mode.isSpring()) {
                    SpringInstanceConfigMonitor monitor = new SpringInstanceConfigMonitor();
                    monitor.setScanIntervalInSecond(scanInterval);
                    monitor.setDefaultAction(defaultAction);
                    // 设置conf目录，默认是user.dir + conf目录组成
                    String rootDir = getProperty(properties, CanalConstants.CANAL_CONF_DIR);
                    if (StringUtils.isEmpty(rootDir)) {
                        rootDir = "../conf";
                    }

                    if (StringUtils.equals("otter-canal", System.getProperty("appName"))) {
                        monitor.setRootConf(rootDir);
                    } else {
                        // eclipse debug模式
                        monitor.setRootConf("src/main/resources/");
                    }
                    return monitor;
                 // admin 模式
                } else if (mode.isManager()) {
                    ManagerInstanceConfigMonitor monitor = new ManagerInstanceConfigMonitor();
                    monitor.setScanIntervalInSecond(scanInterval);
                    monitor.setDefaultAction(defaultAction);
                    String managerAddress = getProperty(properties, CanalConstants.CANAL_ADMIN_MANAGER);
                    monitor.setConfigClient(getManagerClient(managerAddress));
                    return monitor;
                } else {
                    throw new UnsupportedOperationException("unknow mode :" + mode + " for monitor");
                }
                }
            });
        }

canal调度控制器start()：

  public void start() throws Throwable {
        logger.info("## start the canal server[{}({}):{}]", ip, registerIp, port);
        // 创建整个canal的工作节点
        // /otter/canal/cluster/ip:port
        final String path = ZookeeperPathUtils.getCanalClusterNode(registerIp + ":" + port);
        // 创建临时节点
        initCid(path);
        if (zkclientx != null) {
            this.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);
                }
            });
        }
        // 优先启动embedded服务
        embeddedCanalServer.start();
        // 尝试启动一下非lazy状态的通道
        for (Map.Entry<String, InstanceConfig> entry : instanceConfigs.entrySet()) {
            final String destination = entry.getKey();
            InstanceConfig config = entry.getValue();
            // 创建destination的工作节点
            if (!embeddedCanalServer.isStart(destination)) {
                // HA机制启动  创建监听，有就获取，没有就新增
                ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
                if (!config.getLazy() && !runningMonitor.isStart()) {
                    runningMonitor.start();
                }
            }

            if (autoScan) {
                instanceConfigMonitors.get(config.getMode()).register(destination, defaultAction);
            }
        }
        //启动配置文件自动检测机制
        if (autoScan) {
            instanceConfigMonitors.get(globalInstanceConfig.getMode()).start();
            for (InstanceConfigMonitor monitor : instanceConfigMonitors.values()) {
                if (!monitor.isStart()) {
                    // 启动monitor
                    monitor.start();
                }
            }
        }

        // 启动网络接口
        if (canalServer != null) {
            // 启动网络接口，监听客户端请求
            canalServer.start();
        }
    }