public void init() {
        loader = ExtensionLoader.getExtensionLoader(OuterAdapter.class);

        String canalServerHost = this.canalClientConfig.getCanalServerHost();
        SocketAddress sa = null;
        if (canalServerHost != null) {
            String[] ipPort = canalServerHost.split(":");
            sa = new InetSocketAddress(ipPort[0], Integer.parseInt(ipPort[1]));
        }
        String zkHosts = this.canalClientConfig.getZookeeperHosts();

        if ("tcp".equalsIgnoreCase(canalClientConfig.getMode())) {
            // 初始化canal-client的适配器
            for (CanalClientConfig.CanalAdapter canalAdapter : canalClientConfig.getCanalAdapters()) {
                List<List<OuterAdapter>> canalOuterAdapterGroups = new ArrayList<>();

                for (CanalClientConfig.Group connectorGroup : canalAdapter.getGroups()) {
                    List<OuterAdapter> canalOutConnectors = new ArrayList<>();
                    for (OuterAdapterConfig c : connectorGroup.getOuterAdapters()) {
                        loadAdapter(c, canalOutConnectors);
                    }
                    canalOuterAdapterGroups.add(canalOutConnectors);
                }
                CanalAdapterWorker worker;
                if (sa != null) {
                    worker = new CanalAdapterWorker(canalClientConfig,
                        canalAdapter.getInstance(),
                        sa,
                        canalOuterAdapterGroups);
                } else if (zkHosts != null) {
                    worker = new CanalAdapterWorker(canalClientConfig,
                        canalAdapter.getInstance(),
                        zkHosts,
                        canalOuterAdapterGroups);
                } else {
                    throw new RuntimeException("No canal server connector found");
                }
                canalWorkers.put(canalAdapter.getInstance(), worker);
                worker.start();
                logger.info("Start adapter for canal instance: {} succeed", canalAdapter.getInstance());
            }
        } else if ("kafka".equalsIgnoreCase(canalClientConfig.getMode())) {
            // 初始化canal-client-kafka的适配器
            for (CanalClientConfig.CanalAdapter canalAdapter : canalClientConfig.getCanalAdapters()) {
                for (CanalClientConfig.Group group : canalAdapter.getGroups()) {
                    List<List<OuterAdapter>> canalOuterAdapterGroups = new ArrayList<>();
                    List<OuterAdapter> canalOuterAdapters = new ArrayList<>();
                    for (OuterAdapterConfig config : group.getOuterAdapters()) {
                        loadAdapter(config, canalOuterAdapters);
                    }
                    canalOuterAdapterGroups.add(canalOuterAdapters);

                    CanalAdapterKafkaWorker canalKafkaWorker = new CanalAdapterKafkaWorker(canalClientConfig,
                        canalClientConfig.getMqServers(),
                        canalAdapter.getInstance(),
                        group.getGroupId(),
                        canalOuterAdapterGroups,
                        canalClientConfig.getFlatMessage());
                    canalMQWorker.put(canalAdapter.getInstance() + "-kafka-" + group.getGroupId(), canalKafkaWorker);
                    canalKafkaWorker.start();
                    logger.info("Start adapter for canal-client mq topic: {} succeed",
                        canalAdapter.getInstance() + "-" + group.getGroupId());
                }
            }
        } else if ("rocketMQ".equalsIgnoreCase(canalClientConfig.getMode())) {
            // 初始化canal-client-rocketMQ的适配器
            for (CanalClientConfig.CanalAdapter canalAdapter : canalClientConfig.getCanalAdapters()) {
                for (CanalClientConfig.Group group : canalAdapter.getGroups()) {
                    List<List<OuterAdapter>> canalOuterAdapterGroups = new ArrayList<>();
                    List<OuterAdapter> canalOuterAdapters = new ArrayList<>();
                    for (OuterAdapterConfig config : group.getOuterAdapters()) {
                        loadAdapter(config, canalOuterAdapters);
                    }
                    canalOuterAdapterGroups.add(canalOuterAdapters);
                    CanalAdapterRocketMQWorker rocketMQWorker = new CanalAdapterRocketMQWorker(canalClientConfig,
                        canalClientConfig.getMqServers(),
                        canalAdapter.getInstance(),
                        group.getGroupId(),
                        canalOuterAdapterGroups,
                        canalClientConfig.getAccessKey(),
                        canalClientConfig.getSecretKey(),
                        canalClientConfig.getFlatMessage());
                    canalMQWorker.put(canalAdapter.getInstance() + "-rocketmq-" + group.getGroupId(), rocketMQWorker);
                    rocketMQWorker.start();

                    logger.info("Start adapter for canal-client mq topic: {} succeed",
                        canalAdapter.getInstance() + "-" + group.getGroupId());
                }
            }
        }
    }
public void init ( ) { loader = ExtensionLoader . getExtensionLoader ( OuterAdapter . class ) ; String canalServerHost = this . canalClientConfig . getCanalServerHost ( ) ; SocketAddress sa = null ; if ( canalServerHost != null ) { String [ ] ipPort = canalServerHost . split ( ":" ) ; sa = new InetSocketAddress ( ipPort [ 0 ] , Integer . parseInt ( ipPort [ 1 ] ) ) ; } String zkHosts = this . canalClientConfig . getZookeeperHosts ( ) ; if ( "tcp" . equalsIgnoreCase ( canalClientConfig . getMode ( ) ) ) { for ( CanalClientConfig . CanalAdapter canalAdapter : canalClientConfig . getCanalAdapters ( ) ) { List < List < OuterAdapter > > canalOuterAdapterGroups = new ArrayList < > ( ) ; for ( CanalClientConfig . Group connectorGroup : canalAdapter . getGroups ( ) ) { List < OuterAdapter > canalOutConnectors = new ArrayList < > ( ) ; for ( OuterAdapterConfig c : connectorGroup . getOuterAdapters ( ) ) { loadAdapter ( c , canalOutConnectors ) ; } canalOuterAdapterGroups . add ( canalOutConnectors ) ; } CanalAdapterWorker worker ; if ( sa != null ) { worker = new CanalAdapterWorker ( canalClientConfig , canalAdapter . getInstance ( ) , sa , canalOuterAdapterGroups ) ; } else if ( zkHosts != null ) { worker = new CanalAdapterWorker ( canalClientConfig , canalAdapter . getInstance ( ) , zkHosts , canalOuterAdapterGroups ) ; } else { throw new RuntimeException ( "No canal server connector found" ) ; } canalWorkers . put ( canalAdapter . getInstance ( ) , worker ) ; worker . start ( ) ; logger . info ( "Start adapter for canal instance: {} succeed" , canalAdapter . getInstance ( ) ) ; } } else if ( "kafka" . equalsIgnoreCase ( canalClientConfig . getMode ( ) ) ) { for ( CanalClientConfig . CanalAdapter canalAdapter : canalClientConfig . getCanalAdapters ( ) ) { for ( CanalClientConfig . Group group : canalAdapter . getGroups ( ) ) { List < List < OuterAdapter > > canalOuterAdapterGroups = new ArrayList < > ( ) ; List < OuterAdapter > canalOuterAdapters = new ArrayList < > ( ) ; for ( OuterAdapterConfig config : group . getOuterAdapters ( ) ) { loadAdapter ( config , canalOuterAdapters ) ; } canalOuterAdapterGroups . add ( canalOuterAdapters ) ; CanalAdapterKafkaWorker canalKafkaWorker = new CanalAdapterKafkaWorker ( canalClientConfig , canalClientConfig . getMqServers ( ) , canalAdapter . getInstance ( ) , group . getGroupId ( ) , canalOuterAdapterGroups , canalClientConfig . getFlatMessage ( ) ) ; canalMQWorker . put ( canalAdapter . getInstance ( ) + "-kafka-" + group . getGroupId ( ) , canalKafkaWorker ) ; canalKafkaWorker . start ( ) ; logger . info ( "Start adapter for canal-client mq topic: {} succeed" , canalAdapter . getInstance ( ) + "-" + group . getGroupId ( ) ) ; } } } else if ( "rocketMQ" . equalsIgnoreCase ( canalClientConfig . getMode ( ) ) ) { for ( CanalClientConfig . CanalAdapter canalAdapter : canalClientConfig . getCanalAdapters ( ) ) { for ( CanalClientConfig . Group group : canalAdapter . getGroups ( ) ) { List < List < OuterAdapter > > canalOuterAdapterGroups = new ArrayList < > ( ) ; List < OuterAdapter > canalOuterAdapters = new ArrayList < > ( ) ; for ( OuterAdapterConfig config : group . getOuterAdapters ( ) ) { loadAdapter ( config , canalOuterAdapters ) ; } canalOuterAdapterGroups . add ( canalOuterAdapters ) ; CanalAdapterRocketMQWorker rocketMQWorker = new CanalAdapterRocketMQWorker ( canalClientConfig , canalClientConfig . getMqServers ( ) , canalAdapter . getInstance ( ) , group . getGroupId ( ) , canalOuterAdapterGroups , canalClientConfig . getAccessKey ( ) , canalClientConfig . getSecretKey ( ) , canalClientConfig . getFlatMessage ( ) ) ; canalMQWorker . put ( canalAdapter . getInstance ( ) + "-rocketmq-" + group . getGroupId ( ) , rocketMQWorker ) ; rocketMQWorker . start ( ) ; logger . info ( "Start adapter for canal-client mq topic: {} succeed" , canalAdapter . getInstance ( ) + "-" + group . getGroupId ( ) ) ; } } } }
初始化canal-client