cannal源码赏析 - deployer模块

// 1. 读取配置文件properties

// 2. 创建CanalStarter
final CanalStarter canalStater = new CanalStarter(properties);

// 3. 启动
canalStater.start();

public synchronized void start() throws Throwable {
    String serverMode = CanalController.getProperty(properties, CanalConstants.CANAL_SERVER_MODE);
    if (!"tcp".equalsIgnoreCase(serverMode)) {
        ExtensionLoader<CanalMQProducer> loader = ExtensionLoader.getExtensionLoader(CanalMQProducer.class);
        canalMQProducer = loader
            .getExtension(serverMode.toLowerCase(), CONNECTOR_SPI_DIR, CONNECTOR_STANDBY_SPI_DIR);
        if (canalMQProducer != null) {
            canalMQProducer =  new ProxyCanalMQProducer(canalMQProducer);
            canalMQProducer.init(properties);
        }
    }

    if (canalMQProducer != null) {
        MQProperties mqProperties = canalMQProducer.getMqProperties();
        // disable netty
        System.setProperty(CanalConstants.CANAL_WITHOUT_NETTY, "true");
        if (mqProperties.isFlatMessage()) {
            // 设置为raw避免ByteString->Entry的二次解析
            System.setProperty("canal.instance.memory.rawEntry", "false");
        }
    }

    logger.info("## start the canal server.");
    controller = new CanalController(properties);
    controller.start();
    logger.info("## the canal server is running now ......");
    shutdownThread = new Thread(() -> {
        try {
            logger.info("## stop the canal server");
            controller.stop();
            CanalLauncher.runningLatch.countDown();
        } catch (Throwable e) {
            logger.warn("##something goes wrong when stopping canal Server:", e);
        } finally {
            logger.info("## canal server is down.");
        }
    });
    Runtime.getRuntime().addShutdownHook(shutdownThread);

    if (canalMQProducer != null) {
        canalMQStarter = new CanalMQStarter(canalMQProducer);
        String destinations = CanalController.getDestinations(properties);
        canalMQStarter.start(destinations);
        controller.setCanalMQStarter(canalMQStarter);
    }

    // start canalAdmin
    String port = CanalController.getProperty(properties, CanalConstants.CANAL_ADMIN_PORT);
    if (canalAdmin == null && StringUtils.isNotEmpty(port)) {
        String user = CanalController.getProperty(properties, CanalConstants.CANAL_ADMIN_USER);
        String passwd = CanalController.getProperty(properties, CanalConstants.CANAL_ADMIN_PASSWD);
        CanalAdminController canalAdmin = new CanalAdminController(this);
        canalAdmin.setUser(user);
        canalAdmin.setPasswd(passwd);
        String ip = CanalController.getProperty(properties, CanalConstants.CANAL_IP);

        logger.debug("canal admin port:{}, canal admin user:{}, canal admin password: {}, canal ip:{}",
            port,
            user,
            passwd,
            ip);

        CanalAdminWithNetty canalAdminWithNetty = CanalAdminWithNetty.instance();
        canalAdminWithNetty.setCanalAdmin(canalAdmin);
        canalAdminWithNetty.setPort(Integer.parseInt(port));
        canalAdminWithNetty.setIp(ip);
        canalAdminWithNetty.start();
        this.canalAdmin = canalAdminWithNetty;
    }

    running = true;
}