项目介绍

Canal核心各组件介绍及最佳部署实践

本文期望以精炼的语言以及相对清晰的架构图来让你对Canal整体上有一个全面的了解，以及让你知道部署一套Canal的最佳实践

https://blog.zhuchao941.world/canal-core-introduction-and-deployment-best-practice

canal-client-adapter，也属于canal-client，可以把它理解为官方提供的一些常见的异构数据同步通道的实现，比如同步到elasticsearch、hbase、mongodb、mysql、redis、kafka等。所以和canal-client一样，它支持TCP和MQ的消费模式。 当然如果预设的满足不了，那就还是需要自定义了，可以选择基于canal-client-adapter的框架自定义扩展，也可以自己消费对应的MQ或者store去实现。

项目结构

1. launcher——启动模块，里面主要是启动类加一些核心配置

2. common——公共模块

3. 其他——都是具体的adapter实现模块，比如es7x，就是具体用es7x的客户端同步到es里

启动流程分析

1. BootstrapConfiguration

2. CanalAdapterService

3. ApplicationConfigMonitor

4. CanalAdapterApplication

CanalAdapterApplication

@SpringBootApplication
public class CanalAdapterApplication {

    public static void main(String[] args) {
        SpringApplication application = new SpringApplication(CanalAdapterApplication.class);
        application.setBannerMode(Banner.Mode.OFF);
        application.run(args);
    }
}

BootstrapConfiguration

1. 从mysql读取配置

2. 从配置文件读取配置

public class BootstrapConfiguration {

    @Autowired
    private Environment        env;

    private RemoteConfigLoader remoteConfigLoader = null;

    @PostConstruct
    public void loadRemoteConfig() {
        remoteConfigLoader = RemoteConfigLoaderFactory.getRemoteConfigLoader(env);
        if (remoteConfigLoader != null) {
            remoteConfigLoader.loadRemoteConfig();
            remoteConfigLoader.loadRemoteAdapterConfigs();
            remoteConfigLoader.startMonitor(); // 启动监听
        }
    }

    @PreDestroy
    public synchronized void destroy() {
        if (remoteConfigLoader != null) {
            remoteConfigLoader.destroy();
        }
    }
}

	// com.alibaba.otter.canal.adapter.launcher.monitor.remote.RemoteConfigLoaderFactory#getRemoteConfigLoader
    public static RemoteConfigLoader getRemoteConfigLoader(Environment env) {
        try {
            String jdbcUrl = env.getProperty("canal.manager.jdbc.url");
            if (!StringUtils.isEmpty(jdbcUrl)) {
                // load remote config
                String driverName = env.getProperty("canal.manager.jdbc.driverName");
                String jdbcUsername = env.getProperty("canal.manager.jdbc.username");
                String jdbcPassword = env.getProperty("canal.manager.jdbc.password");
                return new DbRemoteConfigLoader(driverName, jdbcUrl, jdbcUsername, jdbcPassword);
            }
            // 可扩展其它远程配置加载器
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
        return null;
    }

1. loadRemoteConfig——从mysql下载application.yml 主配置文件到本地磁盘

2. loadRemoteAdapterConfigs——从mysql下载adapter相关配置文件到本地磁盘

3. startMonitor——启动监听，定时触发loadRemoteConfig和loadRemoteAdapterConfigs

    public void startMonitor() {
        // 监听application.yml变化
        executor.scheduleWithFixedDelay(() -> {
            try {
                loadRemoteConfig();
            } catch (Throwable e) {
                logger.error("scan remote application.yml failed", e);
            }
        }, 10, 3, TimeUnit.SECONDS);

        // 监听adapter变化
        executor.scheduleWithFixedDelay(() -> {
            try {
                loadRemoteAdapterConfigs();
            } catch (Throwable e) {
                logger.error("scan remote adapter configs failed", e);
            }
        }, 10, 3, TimeUnit.SECONDS);
    }

loadRemoteConfig

select name, content, modified_time from canal_config where id=2

loadRemoteAdapterConfigs

select id, category, name, modified_time from canal_adapter_config

select id, category, name, content, modified_time from canal_adapter_config  where id in ()

application.yml

1. 消费模式（直连canal还是通过消息队列）

2. 订阅关系（topic、消费组、消费方式（或者叫异构模式））

3. 数据源定义（主要是消费消息时用来做一些数据查询转换以及etl时的数据源）

adapter.yml

org.springframework.cloud.bootstrap.BootstrapConfiguration=\\
  com.alibaba.otter.canal.adapter.launcher.config.BootstrapConfiguration

ApplicationConfigMonitor

    @PostConstruct
    public void init() {
        File confDir = Util.getConfDirPath();
        try {
            FileAlterationObserver observer = new FileAlterationObserver(confDir,
                FileFilterUtils.and(FileFilterUtils.fileFileFilter(),
                    FileFilterUtils.prefixFileFilter("application"),
                    FileFilterUtils.suffixFileFilter("yml")));
            FileListener listener = new FileListener();
            observer.addListener(listener);
            fileMonitor = new FileAlterationMonitor(3000, observer);
            fileMonitor.start();
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }

1. RemoteConfigLoader.startMonitor()是负责把mysql里的配置下载到本地磁盘上

2. ApplicationConfigMonitor是负责监听本地磁盘上文件的变化，进而触发应用内的相关对象的变化

CanalAdapterService

    // com.alibaba.otter.canal.adapter.launcher.loader.CanalAdapterService#init
	@PostConstruct
    public synchronized void init() {
        if (running) {
            return;
        }
        adapterLoader = new CanalAdapterLoader(adapterCanalConfig);
        // 初始化CanalAdapterLoader
        adapterLoader.init();
        running = true;
    }

	// com.alibaba.otter.canal.adapter.launcher.loader.CanalAdapterLoader#init
    public void init() {
        loader = ExtensionLoader.getExtensionLoader(OuterAdapter.class);

		// 根据application.yml里的配置来创建AdapterProcessor
        for (CanalClientConfig.CanalAdapter canalAdapter : canalClientConfig.getCanalAdapters()) {
            for (CanalClientConfig.Group group : canalAdapter.getGroups()) {
                List<List<OuterAdapter>> canalOuterAdapterGroups = new CopyOnWriteArrayList<>();
                List<OuterAdapter> canalOuterAdapters = new CopyOnWriteArrayList<>();
                // 对于每个配置的outerAdapter都会去load对应的Adapter，并加入到canalOuterAdapters
                for (OuterAdapterConfig config : group.getOuterAdapters()) {
                    loadAdapter(config, canalOuterAdapters);
                }
                // 这个list有点匪夷所思，因为不管怎么样，这个list的长度永远都是1...
                canalOuterAdapterGroups.add(canalOuterAdapters);

                // AdapterProcessor 是个核心类，后面我们分析
                AdapterProcessor adapterProcessor = canalAdapterProcessors.computeIfAbsent(canalAdapter.getInstance()
                                                                                           + "|"
                                                                                           + StringUtils.trimToEmpty(group.getGroupId()),
                    f -> new AdapterProcessor(canalClientConfig,
                        canalAdapter.getInstance(),
                        group.getGroupId(),
                        canalOuterAdapterGroups));
                // 启动 AdapterProcessor
                adapterProcessor.start();                                                                                   + group.getGroupId());
            }
        }
    }

   private void loadAdapter(OuterAdapterConfig config, List<OuterAdapter> canalOutConnectors) {
       // 简化后的代码
       OuterAdapter adapter = loader.getExtension(config.getName(), StringUtils.trimToEmpty(config.getKey()));
       adapter.init(config, evnProperties);
       canalOutConnectors.add(adapter);
    }

1. 首先，因为核心逻辑主要围绕application.yml这个配置文件，所以你需要对这个配置文件的结构有一定的了解，主要了解canalAdapters这个配置项 参考链接

2. 看完配置项之后我再大概解释一下，其实canalAdapters有点像配置MQ的消费组一样，对于每个canalAdapter，你需要配置
1. 1个需要监听的topic
2. 1-n个消费组标识groupId
3. 每个消费组内可以配置1-n个串行消费动作outerAdapters

3. 同一个topic可以配置多个消费组，每个消费组可以配置多个消费动作

4. 每一个消费组会生成一个AdapterProcessor，会调用每个AdapterProcessor的start方法

AdapterProcessor

    public AdapterProcessor(CanalClientConfig canalClientConfig, String destination, String groupId,
                            List<List<OuterAdapter>> canalOuterAdapters){
        // application.yml配置
        this.canalClientConfig = canalClientConfig;
        // 监听topic
        this.canalDestination = destination;
        // 消费组标识
        this.groupId = groupId;
        // 消费动作
        this.canalOuterAdapters = canalOuterAdapters;

        // 永远都是size为1的线程池
        this.groupInnerExecutorService = Util.newFixedThreadPool(canalOuterAdapters.size(), 5000L);
        syncSwitch = (SyncSwitch) SpringContext.getBean(SyncSwitch.class);

        // 加载canalMsgConsumer，这个可以理解是MQ的代理对象
        ExtensionLoader<CanalMsgConsumer> loader = new ExtensionLoader<>(CanalMsgConsumer.class);
        canalMsgConsumer = loader
            .getExtension(canalClientConfig.getMode().toLowerCase(), CONNECTOR_SPI_DIR, CONNECTOR_STANDBY_SPI_DIR);

        Properties properties = canalClientConfig.getConsumerProperties();
        properties.put(CanalConstants.CANAL_MQ_FLAT_MESSAGE, canalClientConfig.getFlatMessage());
        properties.put(CanalConstants.CANAL_ALIYUN_ACCESS_KEY, canalClientConfig.getAccessKey());
        properties.put(CanalConstants.CANAL_ALIYUN_SECRET_KEY, canalClientConfig.getSecretKey());
        ClassLoader cl = Thread.currentThread().getContextClassLoader();
        Thread.currentThread().setContextClassLoader(canalMsgConsumer.getClass().getClassLoader());
        // 初始化 CanalMsgConsumer
        canalMsgConsumer.init(properties, canalDestination, groupId);
        Thread.currentThread().setContextClassLoader(cl);
    }

public interface CanalMsgConsumer {

    void init(Properties properties, String topic, String groupId);

    void connect();

    List<CommonMessage> getMessage(Long timeout, TimeUnit unit);

    void ack();

    void rollback();

    void disconnect();
}

    public void start() {
        if (!running) {
            thread = new Thread(this::process);
            thread.setUncaughtExceptionHandler(handler);
            thread.start();
            running = true;
        }
    }

   private void process() {
       	// 这里为什么需要2个while 不是很懂
        while (!running) { // waiting until running == true
            while (!running) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                }
            }
        }

       	// 消息重试次数
        int retry = canalClientConfig.getRetries() == null
                    || canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries();
        if (retry == -1) {
            retry = Integer.MAX_VALUE;
        }

        while (running) {
            try {
                // 连接MQ，理论上应该是订阅topic，一会看具体代码
                canalMsgConsumer.connect();
                while (running) {
                    if (!running) {
                        break;
                    }

                    for (int i = 0; i < retry; i++) {
                        if (!running) {
                            break;
                        }
                        try {
                            // pull消息
                            List<CommonMessage> commonMessages = canalMsgConsumer
                                .getMessage(this.canalClientConfig.getTimeout(), TimeUnit.MILLISECONDS);
                            // 如果超时没有pull到消息，理论上这里最好加个非空判断
                            // 消费逻辑
                            writeOut(commonMessages);
                            // 消费成功之后ack
                            canalMsgConsumer.ack();
                            break;
                        } catch (Exception e) {
                            // 如果消费重试retry次依旧报错，那么就直接ack掉了（注意：这样会丢消息）
                            if (i != retry - 1) {
                                canalMsgConsumer.rollback();
                            } else {
                                canalMsgConsumer.ack();
                            }
                            Thread.sleep(500);
                        }
                    }
                }
                // 断开与MQ的连接
                canalMsgConsumer.disconnect();
            } catch (Throwable e) {
                logger.error("process error!", e);
            }

            if (running) { // is reconnect
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    // ignore
                }
            }
        }
    }

1. 向mq订阅对应的topic

2. 去mq里拉取消息

3. 处理消息

4. 向mq ack消息

5. 重复第2步

    public void writeOut(final List<CommonMessage> commonMessages) {
        List<Future<Boolean>> futures = new ArrayList<>();
        // 这个canalOuterAdapters size永远都是1，其实不需要foreach
        canalOuterAdapters.forEach(outerAdapters -> {
            futures.add(groupInnerExecutorService.submit(() -> {
                try {
                    // 组内适配器穿行运行，尽量不要配置组内适配器
                    outerAdapters.forEach(adapter -> {
                        List<Dml> dmls = MessageUtil.flatMessage2Dml(canalDestination, groupId, commonMessages);
                        batchSync(dmls, adapter);
                    });
                    return true;
                } catch (Exception e) {
                    logger.error(e.getMessage(), e);
                    return false;
                }
            }));

            RuntimeException exception = null;
            for (Future<Boolean> future : futures) {
                try {
                    if (!future.get()) {
                        exception = new RuntimeException("Outer adapter sync failed! ");
                    }
                } catch (Exception e) {
                    exception = new RuntimeException(e);
                }
            }
            if (exception != null) {
                throw exception;
            }
        });
    }

    public void writeOut(final List<CommonMessage> commonMessages) {
        canalOuterAdapters.get(0).forEach(adapter -> {
            // 消息类型转换
            List<Dml> dmls = MessageUtil.flatMessage2Dml(canalDestination, groupId, commonMessages);
            batchSync(dmls, adapter);
        });
    }

    private void batchSync(List<Dml> dmls, OuterAdapter adapter) {
        // 分批同步
        if (dmls.size() <= canalClientConfig.getSyncBatchSize()) {
            adapter.sync(dmls);
        } else {
            int len = 0;
            List<Dml> dmlsBatch = new ArrayList<>();
            for (Dml dml : dmls) {
                dmlsBatch.add(dml);
                if (dml.getData() == null || dml.getData().isEmpty()) {
                    len += 1;
                } else {
                    len += dml.getData().size();
                }
                if (len >= canalClientConfig.getSyncBatchSize()) {
                    adapter.sync(dmlsBatch);
                    dmlsBatch.clear();
                    len = 0;
                }
            }
            if (!dmlsBatch.isEmpty()) {
                adapter.sync(dmlsBatch);
            }
        }
    }

1. 如果总消息数都没达到1个批次的数量，那么直接同步

2. 如果总消息数超过了1个批次的数量，那么就1批批的做同步
1. 如果最终还剩余不到1批次的数量，那么再同步一次

总结

1. 连接mysql加载主配置文件和adapter相关配置文件

2. 启动定时器轮询配置文件，发现有变化则下载到本地

3. 启动定时器监听本地配置文件，发现有变化则触发应用读取文件并reload相应组件

4. 针对于每一个topic / instance，可配置多个消费组，每个消费组可以配置一组消费单元，组内消费单元串行执行

5. 每个消费组只有一个线程去拉取消息，但是可以通过多实例来控制线程拉取的topic的队列数来达到水平扩展的效果