Alibaba Canal Manager Model 配置管理实现

Alibaba Canal 用于增量订阅消费 mysql 数据库 binlog 日志，详细介绍请见 alibaba/canal。
其中 Server 端配置有两种管理方式: Spring 和 Manager。其中 Spring 方式是基于spring xml + properties 进行定义构建 spring 配置, Manager 方式则可以对接 Web console/manager 系统。本文主要记录一下 Manager 方式的对接逻辑，源码在 canal-deployer 模块，相对比较简单。

源码入口

版本：canal-1.0.24
查看 Canal Server 端的脚本 ./bin/startup.sh，可以找到启动入口类是 CanalLauncher。该类 main 方法首先加载了配置文件到内存用于启动参数，./conf/canal.properties 文件。将参数传递给 final 类 CanalController，所以我们主要查看 CanalController 类。

配置 Manager 方式

要使用 Manager 方式，需要对启动参数进行设置。由前文可知启动时会先读取 canal.properties 文件，所以先需要在该文件增加以下配置

# 配置方式
canal.instance.global.mode=manager
# 是否开启自动扫描
canal.auto.scan=true
# 自动扫描间隔，单位秒
canal.auto.scan.interval=5
# 全局的manager配置方式的链接信息，用于标识该 Server
canal.instance.global.manager.address = 127.0.0.1:1099

这里需要简单说明几个概念
Server: 表示由 ./bin/startup.sh 脚本启动的程序，即一个 JVM;
Instance: 对应一个 Mysql 实例，代码为 CanalInstance 接口;
Destination: 字符串类型，对应一个 Instance;

CanalServerWithEmbedded 类: 连接 mysql master，管理多个 CanalInstance;
CanalServerWithNetty 类: 基于 netty 网络服务的 server 实现，用于与 Client 通讯;
CanalConfigClient 类: 存放配置相关信息;
CanalInstanceGenerator 接口: canal 实例生产者，根据 destination 以及 InstanceConfig 生产 CanalInstance;

其 Server 架构如下图

启动一个 Server，里面可有多个 Instance，一个 Instance 读取一个 Mysql 实例的binlog 日志，Destination 则是对一个 Instance 实例的描述字符串，在该 Server 中唯一。

Manager 实现

Canal-Server 配置加载方式

配置初始化

查看 CanalController 类
在其构造 函数可以看到 CanalInstanceGenerator 实例作为 CanalServerWithEmbedded 实例的组件。

public CanalController(final Properties properties) {
        managerClients = MigrateMap.makeComputingMap(new Function<String, CanalConfigClient>() {

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

        // 初始化全局参数设置
        globalInstanceConfig = initGlobalConfig(properties);
        instanceConfigs = new MapMaker().makeMap();
        // 初始化instance config
        initInstanceConfig(properties);

        // 准备canal server
        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);
        ......
}

查看 CanalInstanceGenerator 实例的具体实现，在初始化全局配置 initGlobalConfig 方法中

instanceGenerator = new CanalInstanceGenerator() {

    public CanalInstance generate(String destination) {
        InstanceConfig config = instanceConfigs.get(destination);
        if (config == null) {
            throw new CanalServerException("can't find destination:{}");
        }

        logger.info("CanalInstanceGenerator generate mode[{}]", config.getMode());
        if (config.getMode().isManager()) {
            ManagerCanalInstanceGenerator instanceGenerator = new ManagerCanalInstanceGenerator();
            instanceGenerator.setCanalConfigClient(managerClients.get(config.getManagerAddress()));
            return instanceGenerator.generate(destination);
        } else if (config.getMode().isSpring()) {
            SpringCanalInstanceGenerator instanceGenerator = new SpringCanalInstanceGenerator();
            synchronized (this) {
                try {
                    // 设置当前正在加载的通道，加载spring查找文件时会用到该变量
                    System.setProperty(CanalConstants.CANAL_DESTINATION_PROPERTY, destination);
                    instanceGenerator.setBeanFactory(getBeanFactory(config.getSpringXml()));
                    return instanceGenerator.generate(destination);
                } catch (Throwable e) {
                    logger.error("generator instance failed.", e);
                    throw new CanalException(e);
                } finally {
                    System.setProperty(CanalConstants.CANAL_DESTINATION_PROPERTY, "");
                }
            }
        } else {
            throw new UnsupportedOperationException("unknow mode :" + config.getMode());
        }

    }
};

可以看到 Manger 方式的配置主要是 CanalConfigClient，从 managerClients 集合中获取。该集合实现主要在构造器中，调用 getManagerClient 方法。所以我们主要通过该方法，构造 CanalConfigClient 实例返回即可。查看 CanalConfigClient 类主要有两个方法 findCanal, findFilter，可以确定 CanalConfigClient 表示整个 Server 的配置。getManagerClient 方法参数 managerAddress 则是在配置文件 canal.properties 中 destination 对应的 ip 地址，若没有配置则使用本机ip。这里比较奇怪的是，既然是通过 destination 获取的 ip，然后构建 CanalConfigClient 实例，为啥不让 CanalConfigClient 实例表示一个 destination 的配置而却是整个 Sever 的配置。

实现 CanalConfigClient 类，主要是一些参数的配置(如 Mysql 连接用户名、密码等)， Canal 和字符串 Filter 的构建。Canal 实例的构建可以参考一下 CanalInstanceWithManager 类所用到的一些参数。

        Canal canal = new Canal();

        CanalParameter canalParameter = new CanalParameter();
        canalParameter.setSlaveId(dc.getSlaveId());
        canalParameter.setDbUsername(dc.getUsername());
        canalParameter.setDbPassword(dc.getPassword());
        canalParameter.setIndexMode(CanalParameter.IndexMode.MEMORY);
        List<InetSocketAddress> dbAddresses = new ArrayList<>();
        dbAddresses.add(new InetSocketAddress(dc.getHost(), dc.getPort()));
        canalParameter.setDbAddresses(dbAddresses);

        canal.setCanalParameter(canalParameter);
        canal.setName(destination);

到此可以说已经完成 Manger 方式配置的初始化，那如何更新配置呢，主要是用到了 ManagerInstanceConfigMonitor 类，该类在 CanalController 构造器中初始化。

配置刷新

CanalContoller 构造器中

......
    // 初始化monitor机制
        autoScan = BooleanUtils.toBoolean(getProperty(properties, CanalConstants.CANAL_AUTO_SCAN));
        if (autoScan) {
            defaultAction = new InstanceAction() {

                public void start(String destination) {
                    InstanceConfig config = instanceConfigs.get(destination);
                    if (config == null) {
                        // 重新读取一下instance config
                        config = parseInstanceConfig(properties, destination);
                        instanceConfigs.put(destination, config);
                    }

                    if (!embededCanalServer.isStart(destination)) {
                        // HA机制启动
                        ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
                        if (!config.getLazy() && !runningMonitor.isStart()) {
                            runningMonitor.start();
                        }
                    }
                }

                public void stop(String destination) {
                    // 此处的stop，代表强制退出，非HA机制，所以需要退出HA的monitor和配置信息
                    InstanceConfig config = instanceConfigs.remove(destination);
                    if (config != null) {
                        embededCanalServer.stop(destination);
                        ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
                        if (runningMonitor.isStart()) {
                            runningMonitor.stop();
                        }
                    }
                }

                public void reload(String destination) {
                    // 目前任何配置变化，直接重启，简单处理
                    stop(destination);
                    start(destination);
                }
            };

            instanceConfigMonitors = MigrateMap.makeComputingMap(new Function<InstanceMode, InstanceConfigMonitor>() {

                public InstanceConfigMonitor apply(InstanceMode mode) {
                    int scanInterval = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_AUTO_SCAN_INTERVAL));

                    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;
                    } else if (mode.isManager()) {
                        // 配置更新，实现 ManagerInstanceConfigMonitor 参考 SpringInstanceConfigMonitor， 使用上面的 defaultAction 即可
                        ManagerInstanceConfigMonitor monitor = new ManagerInstanceConfigMonitor();
                        monitor.setScanIntervalInSecond(scanInterval);
                        monitor.setDefaultAction(defaultAction);
                        monitor.setIp(ip);
                        return monitor;
                    } else {
                        throw new UnsupportedOperationException("unknow mode :" + mode + " for monitor");
                    }
                }
            });
        }
.....

查看 ManagerInstanceConfigMonitor 类，主要是实现 scan 方法，若配置有更新的话，回调 InstanceAction。这里还有个地方比较奇怪，可以看到 InstanceAction 回调方法参数是字符串 destination？我们在 scan 方法中调用后端服务器接口，已经获取到新的配置，却只能回调字符串 destination，然后在 CanalController.getManagerClient 方法根据 destination 再去调服务器获取具体配置？这里多了一次调用服务器接口，感觉比较奇怪。

总结

Manager 模式实现步骤

1. 在 canal.properties 配置文件设置 `canal.instance.global.mode=manager`等;
2. 在 CanalController 类构建 CanalConfigClient 实例，根据 ManagerAddress 从 managerClients 获取 CanalConfigClient (根据destination 获取 Canal, Filter);
3. 实现 ManagerInstanceConfigMonitor，启用定时线程刷新配置，使用 InstanceAction 当作回调与 CanalServerWithEmbedded 通信，

配置的过滤 Filter 传递到 EventParser 中的 BinlogParser （实现类 LogEventConvert），所以感觉这里的 Filter 没什么意义，倒不如让 Client 消费所有数据，然后下发再做过滤。