如何用源碼分析canal的deployer模塊

發(fā)布時間：2021-10-27 16:24:32 來源：億速云閱讀：157 作者：柒染欄目：大數據

這期內容當中小編將會給大家?guī)碛嘘P如何用源碼分析canal的deployer模塊，文章內容豐富且以專業(yè)的角度為大家分析和敘述，閱讀完這篇文章希望大家可以有所收獲。

CanalLauncher是啟動入口類

獲取canal.properties配置文件
如果canal.properties配置文件中屬性root.admin.manager有值，那么構造PlainCanalConfigClient，調用PlainCanalConfigClient的findServer獲取PlainCanal，調用PlainCanal的getProperties方法獲取properties
通過properties構造 CanalStarter并調用其start方法

CanalStarter是啟動類

public synchronized void start() throws Throwable {
        //首先根據canal.serverMode構造CanalMQProducer，如果是kafka，構造的是CanalKafkaProducer
        String serverMode = CanalController.getProperty(properties, CanalConstants.CANAL_SERVER_MODE);
        if (serverMode.equalsIgnoreCase("kafka")) {
            canalMQProducer = new CanalKafkaProducer();
        } else if (serverMode.equalsIgnoreCase("rocketmq")) {
            canalMQProducer = new CanalRocketMQProducer();
        }

        if (canalMQProducer != null) {
            // disable netty
            System.setProperty(CanalConstants.CANAL_WITHOUT_NETTY, "true");
            // 設置為raw避免ByteString->Entry的二次解析
            System.setProperty("canal.instance.memory.rawEntry", "false");
        }
        //接下來構造CanalController并調用其start方法
        logger.info("## start the canal server.");
        controller = new CanalController(properties);
        controller.start();
        logger.info("## the canal server is running now ......");
        ...
        //構造CanalMQStarter并調用其start方法，同時設置為CanalController的屬性
        if (canalMQProducer != null) {
            canalMQStarter = new CanalMQStarter(canalMQProducer);
            MQProperties mqProperties = buildMQProperties(properties);
            String destinations = CanalController.getProperty(properties, CanalConstants.CANAL_DESTINATIONS);
            canalMQStarter.start(mqProperties, destinations);
            controller.setCanalMQStarter(canalMQStarter);
        }
        ...
        running = true;
    }

CanalController是實例調度控制器

public CanalController(final Properties properties){
        // 初始化managerClients用于請求admin
        managerClients = MigrateMap.makeComputingMap(new Function<String, PlainCanalConfigClient>() {

            public PlainCanalConfigClient apply(String managerAddress) {
                return getManagerClient(managerAddress);
            }
        });

        // 初始化全局參數設置,包含了全局mode、lazy、managerAddress、springXml,初始化instanceGenerator用于創(chuàng)建instance，其根據InstanceConfig的mode值使用PlainCanalInstanceGenerator或者SpringCanalInstanceGenerator創(chuàng)建CanalInstance
        globalInstanceConfig = initGlobalConfig(properties);
        instanceConfigs = new MapMaker().makeMap();
        // 初始化instance config,包含了實例mode、lazy、managerAddress、springXml
        initInstanceConfig(properties);

        ...
        // 初始化CanalServerWithEmbedded，將instanceGenerator設置為CanalServerWithEmbedded的屬性
        embededCanalServer = CanalServerWithEmbedded.instance();
        embededCanalServer.setCanalInstanceGenerator(instanceGenerator);// 設置自定義的instanceGenerator
        int metricsPort = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_METRICS_PULL_PORT, "11112"));
        embededCanalServer.setMetricsPort(metricsPort);

        this.adminUser = getProperty(properties, CanalConstants.CANAL_ADMIN_USER);
        this.adminPasswd = getProperty(properties, CanalConstants.CANAL_ADMIN_PASSWD);
        embededCanalServer.setUser(getProperty(properties, CanalConstants.CANAL_USER));
        embededCanalServer.setPasswd(getProperty(properties, CanalConstants.CANAL_PASSWD));

        ...
        final String zkServers = getProperty(properties, CanalConstants.CANAL_ZKSERVERS);
        //初始化ZkClientx用于canal集群部署，創(chuàng)建/otteradmin/canal/destinations節(jié)點和/otteradmin/canal/cluster節(jié)點
        if (StringUtils.isNotEmpty(zkServers)) {
            zkclientx = ZkClientx.getZkClient(zkServers);
            // 初始化系統(tǒng)目錄
            zkclientx.createPersistent(ZookeeperPathUtils.DESTINATION_ROOT_NODE, true);
            zkclientx.createPersistent(ZookeeperPathUtils.CANAL_CLUSTER_ROOT_NODE, true);
        }
        // 初始化ServerRunningMonitors的ServerRunningMonitor，用于啟動、關閉實例
        final ServerRunningData serverData = new ServerRunningData(registerIp + ":" + port);
        ServerRunningMonitors.setServerData(serverData);
        ServerRunningMonitors.setRunningMonitors(MigrateMap.makeComputingMap(new Function<String, ServerRunningMonitor>() {
            ...
        }));

        // 初始化InstanceAction，用于啟動和關閉實例
        autoScan = BooleanUtils.toBoolean(getProperty(properties, CanalConstants.CANAL_AUTO_SCAN));
        if (autoScan) {
            defaultAction = new InstanceAction() {
                ...
            };
            // 初始化instanceConfigMonitors，用于獲取所有instanceConfig并啟動所有instance
            instanceConfigMonitors = MigrateMap.makeComputingMap(new Function<InstanceMode, InstanceConfigMonitor>() {

                public InstanceConfigMonitor apply(InstanceMode mode) {
                    ...
                }
            });
        }
    }

ManagerInstanceConfigMonitor是實例掃描器

public void start() {
        super.start();
        //啟動定時任務，定時掃描所有instance
        executor.scheduleWithFixedDelay(new Runnable() {

            public void run() {
                try {
                    scan();
                    if (isFirst) {
                        isFirst = false;
                    }
                } catch (Throwable e) {
                    logger.error("scan failed", e);
                }
            }

        }, 0, scanIntervalInSecond, TimeUnit.SECONDS);
    }

private void scan() {
        //緩存了所有instance的配置，如果發(fā)現有新的instance則啟動或者修改了instance則重啟
        String instances = configClient.findInstances(null);
        final List<String> is = Lists.newArrayList(StringUtils.split(instances, ','));
        List<String> start = Lists.newArrayList();
        List<String> stop = Lists.newArrayList();
        List<String> restart = Lists.newArrayList();
        for (String instance : is) {
            if (!configs.containsKey(instance)) {
                PlainCanal newPlainCanal = configClient.findInstance(instance, null);
                if (newPlainCanal != null) {
                    configs.put(instance, newPlainCanal);
                    start.add(instance);
                }
            } else {
                PlainCanal plainCanal = configs.get(instance);
                PlainCanal newPlainCanal = configClient.findInstance(instance, plainCanal.getMd5());
                if (newPlainCanal != null) {
                    // 配置有變化
                    restart.add(instance);
                    configs.put(instance, newPlainCanal);
                }
            }
        }

        configs.forEach((instance, plainCanal) -> {
            if (!is.contains(instance)) {
                stop.add(instance);
            }
        });

        stop.forEach(instance -> {
            notifyStop(instance);
        });

        restart.forEach(instance -> {
            notifyReload(instance);
        });

        start.forEach(instance -> {
            notifyStart(instance);
        });

    }

private void notifyStart(String destination) {
        try {
            //啟動instance調用InstanceAction啟動實例，最后是調用ServerRunningMonitor啟動實例
            defaultAction.start(destination);
            actions.put(destination, defaultAction);
            // 啟動成功后記錄配置文件信息
        } catch (Throwable e) {
            logger.error(String.format("scan add found[%s] but start failed", destination), e);
        }
    }

ServerRunningMonitor是針對server的running實例控制

public ServerRunningMonitor(){
        // 創(chuàng)建父節(jié)點
        dataListener = new IZkDataListener() {

            public void handleDataChange(String dataPath, Object data) throws Exception {
                MDC.put("destination", destination);
                ServerRunningData runningData = JsonUtils.unmarshalFromByte((byte[]) data, ServerRunningData.class);
                if (!isMine(runningData.getAddress())) {
                    mutex.set(false);
                }

                if (!runningData.isActive() && isMine(runningData.getAddress())) { // 說明出現了主動釋放的操作，并且本機之前是active
                    releaseRunning();// 徹底釋放mainstem
                }

                activeData = (ServerRunningData) runningData;
            }

            public void handleDataDeleted(String dataPath) throws Exception {
                MDC.put("destination", destination);
                mutex.set(false);
                if (!release && activeData != null && isMine(activeData.getAddress())) {
                    // 如果上一次active的狀態(tài)就是本機，則即時觸發(fā)一下active搶占
                    initRunning();
                } else {
                    // 否則就是等待delayTime，避免因網絡瞬端或者zk異常，導致出現頻繁的切換操作
                    delayExector.schedule(new Runnable() {

                        public void run() {
                            initRunning();
                        }
                    }, delayTime, TimeUnit.SECONDS);
                }
            }

        };

    }

public synchronized void start() {
        super.start();
        try {
            //首先調用listener的processStart方法
            processStart();
            if (zkClient != null) {
                // 如果需要盡可能釋放instance資源，不需要監(jiān)聽running節(jié)點，不然即使stop了這臺機器，另一臺機器立馬會start
                // 監(jiān)視/otteradmin/canal/destinations/{0}/running節(jié)點變化
                String path = ZookeeperPathUtils.getDestinationServerRunning(destination);
                zkClient.subscribeDataChanges(path, dataListener);
                
                initRunning();
            } else {
                processActiveEnter();// 沒有zk，直接啟動
            }
        } catch (Exception e) {
            logger.error("start failed", e);
            // 沒有正常啟動，重置一下狀態(tài)，避免干擾下一次start
            stop();
        }

    }
private void processStart() {
        if (listener != null) {
            try {
                //processStart方法中創(chuàng)建/otteradmin/canal/destinations/{0}/cluster/{1}節(jié)點，0是實例名稱，1是當前節(jié)點ip:port
                listener.processStart();
            } catch (Exception e) {
                logger.error("processStart failed", e);
            }
        }
    }

private void initRunning() {
        if (!isStart()) {
            return;
        }

        String path = ZookeeperPathUtils.getDestinationServerRunning(destination);
        // 序列化
        byte[] bytes = JsonUtils.marshalToByte(serverData);
        try {
            mutex.set(false);
            //嘗試創(chuàng)建/otteradmin/canal/destinations/{0}/running節(jié)點
            zkClient.create(path, bytes, CreateMode.EPHEMERAL);
            activeData = serverData;
            //如果成功則調用listener的processEnter方法,processEnter方法中調用CanalServerWithEmbedded的start方法啟動實例和CanalMQStarter的start方法啟動實例
            processActiveEnter();// 觸發(fā)一下事件
            mutex.set(true);
            release = false;
        } catch (ZkNodeExistsException e) {
            bytes = zkClient.readData(path, true);
            if (bytes == null) {// 如果不存在節(jié)點，立即嘗試一次
                initRunning();
            } else {
                activeData = JsonUtils.unmarshalFromByte(bytes, ServerRunningData.class);
            }
        } catch (ZkNoNodeException e) {
            zkClient.createPersistent(ZookeeperPathUtils.getDestinationPath(destination), true); // 嘗試創(chuàng)建父節(jié)點
            initRunning();
        }
    }

canal.properties配置

canal.register.ip =
canal.admin.manager = 127.0.0.1:8089
canal.admin.port = 11110
canal.admin.user = admin
canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
canal.admin.register.auto = true
canal.admin.register.cluster =

上述就是小編為大家分享的如何用源碼分析canal的deployer模塊了，如果剛好有類似的疑惑，不妨參照上述分析進行理解。如果想知道更多相關知識，歡迎關注億速云行業(yè)資訊頻道。

向AI問一下細節(jié)

如何用源碼分析canal的deployer模塊

猜你喜歡

最新資訊

相關推薦

相關標簽