`

转发：2、canal源码分析指 deployer模块

 
阅读更多

转发：http://www.tianshouzhi.com/api/tutorials/canal/381

 

canal有两种使用方式：1、独立部署 2、内嵌到应用中。 deployer模块主要用于独立部署canal server。关于这两种方式的区别，请参见server模块源码分析。deployer模块源码目录结构如下所示：

Image.png

在独立部署canal时，需要首先对canal的源码进行打包

  1. mvn clean install -Dmaven.test.skip -Denv=release

   以本教程使用1.0.24版本为例，打包后会在target目录生成一个以下两个文件：

Image.png

其中canal.deployer-1.0.24.tar.gz就是canal的独立部署包。解压缩后，目录如下所示。其中bin目录和conf目录(包括子目录spring)中的所有文件，都来自于deployer模块。

  1. canal
  2. ├── bin
  3. │   ├── startup.bat
  4. │   ├── startup.sh
  5. │   └── stop.sh
  6. ├── conf
  7. │   ├── canal.properties
  8. │   ├── example
  9. │   │   └── instance.properties
  10. │   ├── logback.xml
  11. │   └── spring
  12. │       ├── default-instance.xml
  13. │       ├── file-instance.xml
  14. │       ├── group-instance.xml
  15. │       ├── local-instance.xml
  16. │       └── memory-instance.xml
  17. ├── lib
  18. │   └── ....依赖的各种jar
  19. └── logs

 

deployer模块主要完成以下功能：

1、读取canal,properties配置文件

2、启动canal server，监听canal client的请求

3、启动canal instance，连接mysql数据库，伪装成slave，解析binlog

4、在canal的运行过程中，监听配置文件的变化

1、启动和停止脚本

bin目录中包含了canal的启动和停止脚本startup.sh和stop.sh，当我们要启动canal时，只需要输入以下命令即可

  1. sh bin/startup.sh

       在windows环境下，可以直接双击startup.bat。

在startup.sh脚本内，会调用com.alibaba.otter.canal.deployer.CanalLauncher类来进行启动，这是分析Canal源码的入口类，如下图所示：

Image.png

同时，startup.sh还会在bin目录下生成一个canal.pid文件，用于存储canal的进程id。当停止canal的时候

  1. sh bin/stop.sh

会根据canal.pid文件中记录的进程id，kill掉canal进程，并且删除这个文件。

2、CannalLauncher

CanalLauncher是整个源码分析的入口类，代码相当简单。步骤是：

1、读取canal.properties文件中的配置

2、利用读取的配置构造一个CanalController实例，将所有的启动操作都委派给CanalController进行处理。

3、最后注册一个钩子函数，在JVM停止时同时也停止canal server。

com.alibaba.otter.canal.deployer.CanalLauncher

  1. public class CanalLauncher {
  2.  
  3.     private static final String CLASSPATH_URL_PREFIX = "classpath:";
  4.     private static final Logger logger               = LoggerFactory.getLogger(CanalLauncher.class);
  5.  
  6.     public static void main(String[] args) throws Throwable {
  7.         try {
  8.             //1、读取canal.properties文件中配置，默认读取classpath下的canal.properties
  9.             String conf = System.getProperty("canal.conf", "classpath:canal.properties");
  10.             Properties properties = new Properties();
  11.             if (conf.startsWith(CLASSPATH_URL_PREFIX)) {
  12.                 conf = StringUtils.substringAfter(conf, CLASSPATH_URL_PREFIX);
  13.                 properties.load(CanalLauncher.class.getClassLoader().getResourceAsStream(conf));
  14.             } else {
  15.                 properties.load(new FileInputStream(conf));
  16.             }
  17.             //2、启动canal，首先将properties对象传递给CanalController，然后调用其start方法启动
  18.             logger.info("## start the canal server.");
  19.             final CanalController controller = new CanalController(properties);
  20.             controller.start();
  21.             logger.info("## the canal server is running now ......");
  22.             //3、关闭canal，通过添加JVM的钩子，JVM停止前会回调run方法，其内部调用controller.stop()方法进行停止
  23.             Runtime.getRuntime().addShutdownHook(new Thread() {
  24.  
  25.                 public void run() {
  26.                     try {
  27.                         logger.info("## stop the canal server");
  28.                         controller.stop();
  29.                     } catch (Throwable e) {
  30.                         logger.warn("##something goes wrong when stopping canal Server:\n{}",
  31.                             ExceptionUtils.getFullStackTrace(e));
  32.                     } finally {
  33.                         logger.info("## canal server is down.");
  34.                     }
  35.                 }
  36.  
  37.             });
  38.         } catch (Throwable e) {
  39.             logger.error("## Something goes wrong when starting up the canal Server:\n{}",
  40.                 ExceptionUtils.getFullStackTrace(e));
  41.             System.exit(0);
  42.         }
  43.     }
  44. }

可以看到，CanalLauncher实际上只是负责读取canal.properties配置文件，然后构造CanalController对象，并通过其start和stop方法来开启和停止canal。因此，如果说CanalLauncher是canal源码分析的入口类，那么CanalController就是canal源码分析的核心类。

3、CanalController

在CanalController的构造方法中，会对配置文件内容解析，初始化相关成员变量，做好canal server的启动前的准备工作，之后在CanalLauncher中调用CanalController.start方法来启动。

CanalController中定义的相关字段和构造方法，如下所示：

  1. public class CanalController {
  2.  
  3.     private static final Logger  logger   = LoggerFactory.getLogger(CanalController.class);
  4.     private Long                                     cid;
  5.     private String                                   ip; 
  6.     private int                                  port;
  7.     // 默认使用spring的方式载入    
  8.     private Map<String, InstanceConfig>              instanceConfigs;
  9.     private InstanceConfig                           globalInstanceConfig;
  10.     private Map<String, CanalConfigClient>           managerClients;
  11.     // 监听instance config的变化
  12.     private boolean                             autoScan = true;
  13.     private InstanceAction                           defaultAction;
  14.     private Map<InstanceMode, InstanceConfigMonitor> instanceConfigMonitors;
  15.     private CanalServerWithEmbedded                  embededCanalServer;
  16.     private CanalServerWithNetty                     canalServer;
  17.  
  18.     private CanalInstanceGenerator                   instanceGenerator;
  19.     private ZkClientx                                zkclientx;
  20.  
  21.     public CanalController(){
  22.         this(System.getProperties());
  23.     }
  24.  
  25.     public CanalController(final Properties properties){
  26.         managerClients = MigrateMap.makeComputingMap(new Function<String, CanalConfigClient>() {
  27.  
  28.             public CanalConfigClient apply(String managerAddress) {
  29.                 return getManagerClient(managerAddress);
  30.             }
  31.         });
  32.          //1、配置解析    
  33.        globalInstanceConfig = initGlobalConfig(properties);
  34.         instanceConfigs = new MapMaker().makeMap();      
  35.        initInstanceConfig(properties);
  36.  
  37.         // 2、准备canal server
  38.         cid = Long.valueOf(getProperty(properties, CanalConstants.CANAL_ID));
  39.         ip = getProperty(properties, CanalConstants.CANAL_IP);
  40.         port = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_PORT));
  41.         embededCanalServer = CanalServerWithEmbedded.instance();
  42.         embededCanalServer.setCanalInstanceGenerator(instanceGenerator);// 设置自定义的instanceGenerator       
  43.        canalServer = CanalServerWithNetty.instance();
  44.         canalServer.setIp(ip);
  45.         canalServer.setPort(port);
  46.         
  47.          //3、初始化zk相关代码 
  48.         // 处理下ip为空，默认使用hostIp暴露到zk中       
  49.        if (StringUtils.isEmpty(ip)) {
  50.             ip = AddressUtils.getHostIp();
  51.         }
  52.         final String zkServers = getProperty(properties, CanalConstants.CANAL_ZKSERVERS);
  53.         if (StringUtils.isNotEmpty(zkServers)) {
  54.             zkclientx = ZkClientx.getZkClient(zkServers);
  55.             // 初始化系统目录           
  56.           zkclientx.createPersistent(ZookeeperPathUtils.DESTINATION_ROOT_NODE, true);
  57.             zkclientx.createPersistent(ZookeeperPathUtils.CANAL_CLUSTER_ROOT_NODE, true);
  58.         }
  59.         //4 CanalInstance运行状态监控
  60.         final ServerRunningData serverData = new ServerRunningData(cid, ip + ":" + port);
  61.         ServerRunningMonitors.setServerData(serverData);
  62.         ServerRunningMonitors.setRunningMonitors(//...);
  63.  
  64.         //5、autoScan机制相关代码    
  65.        autoScan = BooleanUtils.toBoolean(getProperty(properties, CanalConstants.CANAL_AUTO_SCAN));
  66.         if (autoScan) {
  67.             defaultAction = new InstanceAction() {//....};
  68.  
  69.             instanceConfigMonitors = //....
  70.         }
  71.     }
  72. ....
  73. }

为了读者能够尽量容易的看出CanalController的构造方法中都做了什么，上面代码片段中省略了部分代码。这样，我们可以很明显的看出来， ，在CanalController构造方法中的代码分划分为了固定的几个处理步骤，下面按照几个步骤的划分，逐一进行讲解，并详细的介绍CanalController中定义的各个字段的作用。

3.1 配置解析相关代码

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

3.1.1 globalInstanceConfig字段

表示canal instance的全局配置，类型为InstanceConfig，通过initGlobalConfig方法进行初始化。主要用于解析canal.properties以下几个配置项：

  • canal.instance.global.mode：确定canal instance配置加载方式，取值有manager|spring两种方式

  • canal.instance.global.lazy：确定canal instance是否延迟初始化

  • canal.instance.global.manager.address：配置中心地址。如果canal.instance.global.mode=manager，需要提供此配置项

  • canal.instance.global.spring.xml：spring配置文件路径。如果canal.instance.global.mode=spring，需要提供此配置项

 

initGlobalConfig源码如下所示：

  1. private InstanceConfig initGlobalConfig(Properties properties) {
  2.     InstanceConfig globalConfig = new InstanceConfig();
  3.     //读取canal.instance.global.mode
  4.     String modeStr = getProperty(properties, CanalConstants.getInstanceModeKey(CanalConstants.GLOBAL_NAME));
  5.     if (StringUtils.isNotEmpty(modeStr)) {
  6.         //将modelStr转成枚举InstanceMode，这是一个枚举类，只有2个取值，SPRING\MANAGER，对应两种配置方式
  7.         globalConfig.setMode(InstanceMode.valueOf(StringUtils.upperCase(modeStr)));
  8.     }
  9.     //读取canal.instance.global.lazy
  10.     String lazyStr = getProperty(properties, CanalConstants.getInstancLazyKey(CanalConstants.GLOBAL_NAME));
  11.     if (StringUtils.isNotEmpty(lazyStr)) {
  12.         globalConfig.setLazy(Boolean.valueOf(lazyStr));
  13.     }
  14.    //读取canal.instance.global.manager.address
  15.     String managerAddress = getProperty(properties,
  16.         CanalConstants.getInstanceManagerAddressKey(CanalConstants.GLOBAL_NAME));
  17.     if (StringUtils.isNotEmpty(managerAddress)) {
  18.         globalConfig.setManagerAddress(managerAddress);
  19.     }
  20.     //读取canal.instance.global.spring.xml
  21.     String springXml = getProperty(properties, CanalConstants.getInstancSpringXmlKey(CanalConstants.GLOBAL_NAME));
  22.     if (StringUtils.isNotEmpty(springXml)) {
  23.         globalConfig.setSpringXml(springXml);
  24.     }
  25.  
  26.     instanceGenerator = //...初始化instanceGenerator 
  27.  
  28.     return globalConfig;
  29. }

其中canal.instance.global.mode用于确定canal instance的全局配置加载方式，其取值范围有2个：spring、manager。我们知道一个canal server中可以启动多个canal instance，每个instance都有各自的配置。instance的配置也可以放在本地，也可以放在远程配置中心里。我们可以自定义每个canal instance配置文件存储的位置，如果所有canal instance的配置都在本地或者远程，此时我们就可以通过canal.instance.global.mode这个配置项，来统一的指定配置文件的位置，避免为每个canal instance单独指定。

其中：

spring方式：

表示所有的canal instance的配置文件位于本地。此时，我们必须提供配置项canal.instance.global.spring.xml指定spring配置文件的路径。canal提供了多个spring配置文件：file-instance.xml、default-instance.xml、memory-instance.xml、local-instance.xml、group-instance.xml。这么多配置文件主要是为了支持canal instance不同的工作方式。我们在稍后将会讲解各个配置文件的区别。而在这些配置文件的开头，我们无一例外的可以看到以下配置：

  1. <bean class="com.alibaba.otter.canal.instance.spring.support.PropertyPlaceholderConfigurer" lazy-init="false">
  2.         <property name="ignoreResourceNotFound" value="true" />
  3.         <property name="systemPropertiesModeName" value="SYSTEM_PROPERTIES_MODE_OVERRIDE"/><!-- 允许system覆盖 -->
  4.         <property name="locationNames">
  5.             <list>
  6.                 <value>classpath:canal.properties</value>
  7.                 <value>classpath:${canal.instance.destination:}/instance.properties</value>
  8.             </list>
  9.         </property>
  10.     </bean>

这里我们可以看到，所谓通过spring方式加载canal instance配置，无非就是通过spring提供的PropertyPlaceholderConfigurer来加载canal instance的配置文件instance.properties。

这里instance.properties的文件完整路径是${canal.instance.destination:}/instance.properties，其中${canal.instance.destination}是一个变量。这是因为我们可以在一个canal server中配置多个canal instance，每个canal instance配置文件的名称都是instance.properties，因此我们需要通过目录进行区分。例如我们通过配置项canal.destinations指定多个canal instance的名字

  1. canal.destinations= example1,example2

此时我们就要conf目录下，新建两个子目录example1和example2，每个目录下各自放置一个instance.properties。

canal在初始化时就会分别使用example1和example2来替换${canal.instance.destination:}，从而分别根据example1/instance.properties和example2/instance.properties创建2个canal instance。

manager方式：

表示所有的canal instance的配置文件位于远程配置中心，此时我们必须提供配置项 canal.instance.global.manager.address来指定远程配置中心的地址。目前alibaba内部配置使用这种方式。开发者可以自己实现CanalConfigClient，连接各自的管理系统，完成接入。

3.1.2 instanceGenerator字段

类型为CanalInstanceGenerator。在initGlobalConfig方法中，除了创建了globalInstanceConfig实例，同时还为字段instanceGenerator字段进行了赋值。

顾名思义，这个字段用于创建CanalInstance实例。这是instance模块中的类，其作用就是为canal.properties文件中canal.destinations配置项列出的每个destination，创建一个CanalInstance实例。CanalInstanceGenerator是一个接口，定义如下所示：

  1. public interface CanalInstanceGenerator {
  2.  
  3.     /**
  4.      * 通过 destination 产生特定的 {@link CanalInstance}
  5.      */
  6.     CanalInstance generate(String destination);
  7. }

针对spring和manager两种instance配置的加载方式，CanalInstanceGenerator提供了两个对应的实现类，如下所示：

Image.png

instanceGenerator字段通过一个匿名内部类进行初始化。其内部会判断配置的各个destination的配置加载方式，spring 或者manager。

  1. instanceGenerator = new CanalInstanceGenerator() {
  2.  
  3.         public CanalInstance generate(String destination) {
  4.            //1、根据destination从instanceConfigs获取对应的InstanceConfig对象
  5.             InstanceConfig config = instanceConfigs.get(destination);
  6.             if (config == null) {
  7.                 throw new CanalServerException("can't find destination:{}");
  8.             }
  9.           //2、如果destination对应的InstanceConfig的mode是manager方式，使用ManagerCanalInstanceGenerator
  10.             if (config.getMode().isManager()) {
  11.                 ManagerCanalInstanceGenerator instanceGenerator = new ManagerCanalInstanceGenerator();
  12.                 instanceGenerator.setCanalConfigClient(managerClients.get(config.getManagerAddress()));
  13.                 return instanceGenerator.generate(destination);
  14.             } else if (config.getMode().isSpring()) {
  15.           //3、如果destination对应的InstanceConfig的mode是spring方式，使用SpringCanalInstanceGenerator
  16.                 SpringCanalInstanceGenerator instanceGenerator = new SpringCanalInstanceGenerator();
  17.                 synchronized (this) {
  18.                     try {
  19.                         // 设置当前正在加载的通道，加载spring查找文件时会用到该变量                        
  20.                         System.setProperty(CanalConstants.CANAL_DESTINATION_PROPERTY, destination);
  21.                         instanceGenerator.setBeanFactory(getBeanFactory(config.getSpringXml()));
  22.                         return instanceGenerator.generate(destination);
  23.                     } catch (Throwable e) {
  24.                         logger.error("generator instance failed.", e);
  25.                         throw new CanalException(e);
  26.                     } finally {
  27.                         System.setProperty(CanalConstants.CANAL_DESTINATION_PROPERTY, "");
  28.                     }
  29.                 }
  30.             } else {
  31.                 throw new UnsupportedOperationException("unknow mode :" + config.getMode());
  32.             }
  33.  
  34.         }
  35.  
  36.     };

上述代码中的第1步比较变态，从instanceConfigs中根据destination作为参数，获得对应的InstanceConfig。而instanceConfigs目前还没有被初始化，这个字段是在稍后将后将要讲解的initInstanceConfig方法初始化的，不过由于这是一个引用类型，当initInstanceConfig方法被执行后，instanceConfigs字段中也就有值了。目前，我们姑且认为， instanceConfigs这个Map<String, InstanceConfig>类型的字段已经被初始化好了。  

2、3两步用于确定是instance的配置加载方式是spring还是manager，如果是spring，就使用SpringCanalInstanceGenerator创建CanalInstance实例，如果是manager，就使用ManagerCanalInstanceGenerator创建CanalInstance实例。

由于目前manager方式的源码并未开源，因此，我们只分析SpringCanalInstanceGenerator相关代码。

上述代码中，首先创建了一个SpringCanalInstanceGenerator实例，然后往里面设置了一个BeanFactory。

  1. instanceGenerator.setBeanFactory(getBeanFactory(config.getSpringXml()));

其中config.getSpringXml()返回的就是我们在canal.properties中通过canal.instance.global.spring.xml配置项指定了spring配置文件路径。getBeanFactory方法源码如下所示：

  1. private BeanFactory getBeanFactory(String springXml) {
  2.         ApplicationContext applicationContext = new ClassPathXmlApplicationContext(springXml);
  3.         return applicationContext;
  4.     }

 

往SpringCanalInstanceGenerator设置了BeanFactory之后，就可以通过其的generate方法获得CanalInstance实例。

SpringCanalInstanceGenerator的源码如下所示：

  1. public class SpringCanalInstanceGenerator implements CanalInstanceGenerator, BeanFactoryAware {
  2.  
  3.     private String      defaultName = "instance";
  4.     private BeanFactory beanFactory;
  5.  
  6.     public CanalInstance generate(String destination) {
  7.         String beanName = destination;
  8.         //首先判断beanFactory是否包含以destination为id的bean
  9.         if (!beanFactory.containsBean(beanName)) {
  10.             beanName = defaultName;//如果没有，设置要获取的bean的id为instance。
  11.         }
  12.         //以默认的bean的id值"instance"来获取CanalInstance实例
  13.         return (CanalInstance) beanFactory.getBean(beanName);
  14.     }
  15.  
  16.     public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
  17.         this.beanFactory = beanFactory;
  18.     }
  19.  
  20. }

首先尝试以传入的参数destination来获取CanalInstance实例，如果没有，就以默认的bean的id值"instance"来获取CanalInstance实例。事实上，如果你没有修改spring配置文件，那么默认的名字就是instance。事实上，在canal提供的各个spring配置文件xxx-instance.xml中，都有类似以下配置：

  1.  <bean id="instance" class="com.alibaba.otter.canal.instance.spring.CanalInstanceWithSpring">
  2.       <property name="destination" value="${canal.instance.destination}" />
  3.       <property name="eventParser">
  4.          <ref local="eventParser" />
  5.       </property>
  6.       <property name="eventSink">
  7.          <ref local="eventSink" />
  8.       </property>
  9.       <property name="eventStore">
  10.          <ref local="eventStore" />
  11.       </property>
  12.       <property name="metaManager">
  13.          <ref local="metaManager" />
  14.       </property>
  15.       <property name="alarmHandler">
  16.          <ref local="alarmHandler" />
  17.       </property>
  18.    </bean>

上面的代码片段中，我们看到的确有一个bean的名字是instance，其类型是CanalInstanceWithSpring，这是CanalInstance接口的实现类。类似的，我们可以想到在manager配置方式下，获取的CanalInstance实现类是CanalInstanceWithManager。事实上，你想的没错，CanalInstance的类图继承关系如下所示：

Image.png

需要注意的是，到目前为止，我们只是创建好了CanalInstanceGenerator，而CanalInstance尚未创建。在CanalController的start方法被调用时，CanalInstance才会被真正的创建，相关源码将在稍后分析。

3.1.3 instanceConfigs字段

类型为Map<String, InstanceConfig>。前面提到初始化instanceGenerator后，当其generate方法被调用时，会尝试从instanceConfigs根据一个destination获取对应的InstanceConfig，现在分析instanceConfigs的相关初始化代码。

我们知道globalInstanceConfig定义全局的配置加载方式。如果需要把部分CanalInstance配置放于本地，另外一部分CanalIntance配置放于远程配置中心，则只通过全局方式配置，无法达到这个要求。虽然这种情况很少见，但是为了提供最大的灵活性，canal支持每个CanalIntance自己来定义自己的加载方式，来覆盖默认的全局配置加载方式。而每个destination对应的InstanceConfig配置就存放于instanceConfigs字段中。

举例来说：

  1. //当前server上部署的instance列表
  2. canal.destinations=instance1,instance2 
  3.  
  4. //instance配置全局加载方式
  5. canal.instance.global.mode = spring
  6. canal.instance.global.lazy = false
  7. canal.instance.global.spring.xml = classpath:spring/file-instance.xml
  8.  
  9. //instance1覆盖全局加载方式
  10. canal.instance.instance1.mode = manager
  11. canal.instance.instance1.manager.address = 127.0.0.1:1099
  12. canal.instance.instance1.lazy = tue

这段配置中，设置了instance的全局加载方式为spring，instance1覆盖了全局配置，使用manager方式加载配置。而instance2没有覆盖配置，因此默认使用spring加载方式。

instanceConfigs字段通过initInstanceConfig方法进行初始化

  1. instanceConfigs = new MapMaker().makeMap();//这里利用Google Guava框架的MapMaker创建Map实例并赋值给instanceConfigs
  2. // 初始化instance config
  3. initInstanceConfig(properties);

initInstanceConfig方法源码如下：

  1. private void initInstanceConfig(Properties properties) {
  2.     //读取配置项canal.destinations
  3.     String destinationStr = getProperty(properties, CanalConstants.CANAL_DESTINATIONS);
  4.     //以","分割canal.destinations，得到一个数组形式的destination
  5.     String[] destinations = StringUtils.split(destinationStr, CanalConstants.CANAL_DESTINATION_SPLIT);
  6.     for (String destination : destinations) {
  7.         //为每一个destination生成一个InstanceConfig实例
  8.         InstanceConfig config = parseInstanceConfig(properties, destination);
  9.         //将destination对应的InstanceConfig放入instanceConfigs中
  10.         InstanceConfig oldConfig = instanceConfigs.put(destination, config);
  11.  
  12.         if (oldConfig != null) {
  13.             logger.warn("destination:{} old config:{} has replace by new config:{}", new Object[] { destination,
  14.                     oldConfig, config });
  15.         }
  16.     }
  17. }

上面代码片段中，首先解析canal.destinations配置项，可以理解一个destination就对应要初始化一个canal instance。针对每个destination会创建各自的InstanceConfig，最终都会放到instanceConfigs这个Map中。

各个destination对应的InstanceConfig都是通过parseInstanceConfig方法来解析

  1. private InstanceConfig parseInstanceConfig(Properties properties, String destination) {
  2.     //每个destination对应的InstanceConfig都引用了全局的globalInstanceConfig
  3.     InstanceConfig config = new InstanceConfig(globalInstanceConfig);
  4.     //...其他几个配置项与获取globalInstanceConfig类似，不再赘述，唯一注意的的是配置项的key部分中的global变成传递进来的destination
  5.     return config;
  6. }

此时我们可以看一下InstanceConfig类的源码：

  1. public class InstanceConfig {
  2.  
  3.     private InstanceConfig globalConfig;
  4.     private InstanceMode   mode;
  5.     private Boolean        lazy;
  6.     private String         managerAddress;
  7.     private String         springXml;
  8.  
  9.     public InstanceConfig(){
  10.  
  11.     }
  12.  
  13.     public InstanceConfig(InstanceConfig globalConfig){
  14.         this.globalConfig = globalConfig;
  15.     }
  16.  
  17.     public static enum InstanceMode {
  18.         SPRING, MANAGER;
  19.  
  20.         public boolean isSpring() {
  21.             return this == InstanceMode.SPRING;
  22.         }
  23.  
  24.         public boolean isManager() {
  25.             return this == InstanceMode.MANAGER;
  26.         }
  27.     }
  28.  
  29.     public Boolean getLazy() {
  30.         if (lazy == null && globalConfig != null) {
  31.             return globalConfig.getLazy();
  32.         } else {
  33.             return lazy;
  34.         }
  35.     }
  36.  
  37.     public void setLazy(Boolean lazy) {
  38.         this.lazy = lazy;
  39.     }
  40.  
  41.     public InstanceMode getMode() {
  42.         if (mode == null && globalConfig != null) {
  43.             return globalConfig.getMode();
  44.         } else {
  45.             return mode;
  46.         }
  47.     }
  48.  
  49.     public void setMode(InstanceMode mode) {
  50.         this.mode = mode;
  51.     }
  52.  
  53.     public String getManagerAddress() {
  54.         if (managerAddress == null && globalConfig != null) {
  55.             return globalConfig.getManagerAddress();
  56.         } else {
  57.             return managerAddress;
  58.         }
  59.     }
  60.  
  61.     public void setManagerAddress(String managerAddress) {
  62.         this.managerAddress = managerAddress;
  63.     }
  64.  
  65.     public String getSpringXml() {
  66.         if (springXml == null && globalConfig != null) {
  67.             return globalConfig.getSpringXml();
  68.         } else {
  69.             return springXml;
  70.         }
  71.     }
  72.  
  73.     public void setSpringXml(String springXml) {
  74.         this.springXml = springXml;
  75.     }
  76.  
  77.     public String toString() {
  78.         return ToStringBuilder.reflectionToString(this, CanalToStringStyle.DEFAULT_STYLE);
  79.     }
  80.  
  81. }

  可以看到，InstanceConfig类中维护了一个globalConfig字段，其类型也是InstanceConfig。而其相关get方法在执行时，会按照以下逻辑进行判断：如果没有自身没有这个配置，则返回全局配置，如果有，则返回自身的配置。通过这种方式实现对全局配置的覆盖。

 3.2 准备canal server相关代码 

  1. cid = Long.valueOf(getProperty(properties, CanalConstants.CANAL_ID));
  2. ip = getProperty(properties, CanalConstants.CANAL_IP);
  3. port = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_PORT));
  4.  
  5. embededCanalServer = CanalServerWithEmbedded.instance();
  6. embededCanalServer.setCanalInstanceGenerator(instanceGenerator);// 设置自定义的instanceGenerator
  7. canalServer = CanalServerWithNetty.instance();
  8. canalServer.setIp(ip);
  9. canalServer.setPort(port);

上述代码中，首先解析了cid、ip、port字段，其中：

cid：Long，对应canal.properties文件中的canal.id，目前无实际用途

ip：String，对应canal.properties文件中的canal.ip，canal server监听的ip。

port：int，对应canal.properties文件中的canal.port，canal server监听的端口

 

之后分别为以下两个字段赋值：

embededCanalServer：类型为CanalServerWithEmbedded 

canalServer：类型为CanalServerWithNetty

CanalServerWithEmbedded 和 CanalServerWithNetty都实现了CanalServer接口，且都实现了单例模式，通过静态方法instance获取实例。

关于这两种类型的实现，canal官方文档有以下描述：

Image.png

说白了，就是我们可以不必独立部署canal server。在应用直接使用CanalServerWithEmbedded直连mysql数据库。如果觉得自己的技术hold不住相关代码，就独立部署一个canal server，使用canal提供的客户端，连接canal server获取binlog解析后数据。而CanalServerWithNetty是在CanalServerWithEmbedded的基础上做的一层封装，用于与客户端通信。   

在独立部署canal server时，Canal客户端发送的所有请求都交给CanalServerWithNetty处理解析，解析完成之后委派给了交给CanalServerWithEmbedded进行处理。因此CanalServerWithNetty就是一个马甲而已。CanalServerWithEmbedded才是核心。

因此，在上述代码中，我们看到，用于生成CanalInstance实例的instanceGenerator被设置到了CanalServerWithEmbedded中，而ip和port被设置到CanalServerWithNetty中。

关于CanalServerWithNetty如何将客户端的请求委派给CanalServerWithEmbedded进行处理，我们将在server模块源码分析中进行讲解。

3.3 初始化zk相关代码

  1.    //读取canal.properties中的配置项canal.zkServers，如果没有这个配置，则表示项目不使用zk
  2. final String zkServers = getProperty(properties, CanalConstants.CANAL_ZKSERVERS);
  3. if (StringUtils.isNotEmpty(zkServers)) {
  4.     //创建zk实例
  5.     zkclientx = ZkClientx.getZkClient(zkServers);
  6.     // 初始化系统目录
  7.     //destination列表，路径为/otter/canal/destinations
  8.     zkclientx.createPersistent(ZookeeperPathUtils.DESTINATION_ROOT_NODE, true);
  9.     //整个canal server的集群列表，路径为/otter/canal/cluster
  10.     zkclientx.createPersistent(ZookeeperPathUtils.CANAL_CLUSTER_ROOT_NODE, true);
  11. }

canal支持利用了zk来完成HA机制、以及将当前消费到到的mysql的binlog位置记录到zk中。ZkClientx是canal对ZkClient进行了一层简单的封装。

显然，当我们没有配置canal.zkServers，那么zkclientx不会被初始化。

关于Canal如何利用ZK做HA，我们将在稍后的代码中进行分。而利用zk记录binlog的消费进度，将在之后的章节进行分析。

3.4 CanalInstance运行状态监控相关代码

由于这段代码比较长且恶心，这里笔者暂时对部分代码进行省略，以便读者看清楚整各脉络

  1. final ServerRunningData serverData = new ServerRunningData(cid, ip + ":" + port);
  2.         ServerRunningMonitors.setServerData(serverData);
  3.         ServerRunningMonitors.setRunningMonitors(MigrateMap.makeComputingMap(new Function<String, ServerRunningMonitor>() {
  4.             public ServerRunningMonitor apply(final String destination) {
  5.                 ServerRunningMonitor runningMonitor = new ServerRunningMonitor(serverData);
  6.                 runningMonitor.setDestination(destination);
  7.                 runningMonitor.setListener(new ServerRunningListener() {....});//省略ServerRunningListener的具体实现
  8.                 if (zkclientx != null) {
  9.                     runningMonitor.setZkClient(zkclientx);
  10.                 }
  11.                 // 触发创建一下cid节点
  12.                 runningMonitor.init();
  13.                 return runningMonitor;
  14.             }
  15.         }));

上述代码中，ServerRunningMonitors是ServerRunningMonitor对象的容器，而ServerRunningMonitor用于监控CanalInstance。

    canal会为每一个destination创建一个CanalInstance，每个CanalInstance都会由一个ServerRunningMonitor来进行监控。而ServerRunningMonitor统一由ServerRunningMonitors进行管理。

    除了CanalInstance需要监控，CanalServer本身也需要监控。因此我们在代码一开始，就看到往ServerRunningMonitors设置了一个ServerRunningData对象，封装了canal server监听的ip和端口等信息。

ServerRunningMonitors源码如下所示：

  1. public class ServerRunningMonitors {
  2.     private static ServerRunningData serverData;
  3.     private static Map               runningMonitors; // <String,ServerRunningMonitor>
  4.     public static ServerRunningData getServerData() {
  5.         return serverData;
  6.     }
  7.     public static Map<String, ServerRunningMonitor> getRunningMonitors() {
  8.         return runningMonitors;
  9.     }
  10.     public static ServerRunningMonitor getRunningMonitor(String destination) {
  11.         return (ServerRunningMonitor) runningMonitors.get(destination);
  12.     }
  13.     public static void setServerData(ServerRunningData serverData) {
  14.         ServerRunningMonitors.serverData = serverData;
  15.     }
  16.     public static void setRunningMonitors(Map runningMonitors) {
  17.         ServerRunningMonitors.runningMonitors = runningMonitors;
  18.     }
  19. }

ServerRunningMonitors的setRunningMonitors方法接收的参数是一个Map，其中Map的key是destination，value是ServerRunningMonitor，也就是说针对每一个destination都有一个ServerRunningMonitor来监控。

上述代码中，在往ServerRunningMonitors设置Map时，是通过MigrateMap.makeComputingMap方法来创建的，其接受一个Function类型的参数，这是guava中定义的接口，其声明了apply抽象方法。其工作原理可以通过下面代码片段进行介绍：

  1. Map<String, User> map = MigrateMap.makeComputingMap(new Function<String, User>() {
  2.             @Override
  3.             public User apply(String name) {
  4.                 return new User(name);
  5.             }
  6.         });
  7. User user = map.get("tianshouzhi");//第一次获取时会创建
  8. assert user != null;
  9. assert user == map.get("tianshouzhi");//之后获取，总是返回之前已经创建的对象

这段代码中，我们利用MigrateMap.makeComputingMap创建了一个Map，其中key为String类型，value为User类型。当我们调用map.get("tianshouzhi")方法，最开始这个Map中并没有任何key/value的，于是其就会回调Function的apply方法，利用参数"tianshouzhi"创建一个User对象并返回。之后当我们再以"tianshouzhi"为key从Map中获取User对象时，会直接将前面创建的对象返回。不会回调apply方法，也就是说，只有在第一次尝试获取时，才会回调apply方法。

    而在上述代码中，实际上就利用了这个特性，只不过是根据destination获取ServerRunningMonitor对象，如果不存在就创建。

在创建ServerRunningMonitor对象时，首先根据ServerRunningData创建ServerRunningMonitor实例，之后设置了destination和ServerRunningListener对象，接着，判断如果zkClientx字段如果不为空，也设置到ServerRunningMonitor中，最后调用init方法进行初始化。

  1. ServerRunningMonitor runningMonitor = new ServerRunningMonitor(serverData);
  2. runningMonitor.setDestination(destination);
  3. runningMonitor.setListener(new ServerRunningListener(){...})//省略ServerRunningListener具体代码
  4. if (zkclientx != null) {
  5. runningMonitor.setZkClient(zkclientx);
  6. }
  7. // 触发创建一下cid节点
  8. runningMonitor.init();
  9. return runningMonitor;

ServerRunningListener的实现如下：

  1. new ServerRunningListener() {
  2.     /*内部调用了embededCanalServer的start(destination)方法。
  3.     此处需要划重点，说明每个destination对应的CanalInstance是通过embededCanalServer的start方法启动的，
  4.     这与我们之前分析将instanceGenerator设置到embededCanalServer中可以对应上。
  5.     embededCanalServer负责调用instanceGenerator生成CanalInstance实例，并负责其启动。*/
  6.      public void processActiveEnter() {
  7.          try {
  8.              MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
  9.              embededCanalServer.start(destination);
  10.          } finally {
  11.              MDC.remove(CanalConstants.MDC_DESTINATION);
  12.          }
  13.      }
  14.   //内部调用embededCanalServer的stop(destination)方法。与上start方法类似，只不过是停止CanalInstance。
  15.      public void processActiveExit() {
  16.          try {
  17.              MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
  18.              embededCanalServer.stop(destination);
  19.          } finally {
  20.              MDC.remove(CanalConstants.MDC_DESTINATION);
  21.          }
  22.      }
  23.      /*处理存在zk的情况下，在Canalinstance启动之前，在zk中创建节点。
  24.      路径为：/otter/canal/destinations/{0}/cluster/{1}，其0会被destination替换，1会被ip:port替换。
  25.      此方法会在processActiveEnter()之前被调用*/
  26.      public void processStart() {
  27.          try {
  28.              if (zkclientx != null) {
  29.                  final String path = ZookeeperPathUtils.getDestinationClusterNode(destination, ip + ":" + port);
  30.                  initCid(path);
  31.                  zkclientx.subscribeStateChanges(new IZkStateListener() {
  32.                      public void handleStateChanged(KeeperState state) throws Exception {
  33.                      }
  34.                      public void handleNewSession() throws Exception {
  35.                          initCid(path);
  36.                      }
  37.                  });
  38.              }
  39.          } finally {
  40.              MDC.remove(CanalConstants.MDC_DESTINATION);
  41.          }
  42.      }
  43. //处理存在zk的情况下，在Canalinstance停止前，释放zk节点，路径为/otter/canal/destinations/{0}/cluster/{1}，
  44. //其0会被destination替换，1会被ip:port替换。此方法会在processActiveExit()之前被调用
  45.      public void processStop() {
  46.          try {
  47.              MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
  48.              if (zkclientx != null) {
  49.                  final String path = ZookeeperPathUtils.getDestinationClusterNode(destination, ip + ":" + port);
  50.                  releaseCid(path);
  51.              }
  52.          } finally {
  53.              MDC.remove(CanalConstants.MDC_DESTINATION);
  54.          }
  55.      }
  56. }

上述代码中，我们可以看到启动一个CanalInstance实际上是在ServerRunningListener的processActiveEnter方法中，通过调用embededCanalServer的start(destination)方法进行的，对于停止也是类似。

那么ServerRunningListener中的相关方法到底是在哪里回调的呢？我们可以在ServerRunningMonitor的start和stop方法中找到答案，这里只列出start方法。

  1. public class ServerRunningMonitor extends AbstractCanalLifeCycle {
  2.  
  3. ...
  4. public void start() {
  5.     super.start();
  6.     processStart();//其内部会调用ServerRunningListener的processStart()方法
  7.     if (zkClient != null) {//存在zk，以HA方式启动
  8.         // 如果需要尽可能释放instance资源，不需要监听running节点，不然即使stop了这台机器，另一台机器立马会start
  9.         String path = ZookeeperPathUtils.getDestinationServerRunning(destination);
  10.         zkClient.subscribeDataChanges(path, dataListener);
  11.  
  12.         initRunning();
  13.     } else {//没有zk，直接启动
  14.         processActiveEnter();
  15.     }
  16. }
  17.  
  18. //...stop方法逻辑类似，相关代码省略
  19. }

当ServerRunningMonitor的start方法被调用时，其首先会直接调用processStart方法，这个方法内部直接调了ServerRunningListener的processStart()方法，源码如下所示。通过前面的分析，我们已经知道在存在zkClient!=null的情况，会往zk中创建一个节点。

  1. private void processStart() {
  2.     if (listener != null) {
  3.         try {
  4.             listener.processStart();
  5.         } catch (Exception e) {
  6.             logger.error("processStart failed", e);
  7.         }
  8.     }
  9. }

之后会判断是否存在zkClient，如果不存在，则以本地方式启动，如果存在，则以HA方式启动。我们知道，canal server可以部署成两种方式：集群方式或者独立部署。其中集群方式是利用zk来做HA，独立部署则可以直接进行启动。我们先来看比较简单的直接启动。

直接启动：

不存在zk的情况下，会进入else代码块，调用processActiveEnter方法，其内部调用了listener的processActiveEnter，启动相应destination对应的CanalInstance。

  1. private void processActiveEnter() {
  2.     if (listener != null) {
  3.         try {
  4.             listener.processActiveEnter();
  5.         } catch (Exception e) {
  6.             logger.error("processActiveEnter failed", e);
  7.         }
  8.     }
  9. }

 

HA方式启动：

存在zk，说明canal server可能做了集群，因为canal就是利用zk来做HA的。首先根据destination构造一个zk的节点路径，然后进行监听。

  1. /*构建临时节点的路径：/otter/canal/destinations/{0}/running，其中占位符{0}会被destination替换。
  2. 在集群模式下，可能会有多个canal server共同处理同一个destination，
  3. 在某一时刻，只能由一个canal server进行处理，处理这个destination的canal server进入running状态，其他canal server进入standby状态。*/
  4. String path = ZookeeperPathUtils.getDestinationServerRunning(destination);
  5.  
  6. /*对destination对应的running节点进行监听，一旦发生了变化，则说明可能其他处理相同destination的canal server可能出现了异常，
  7. 此时需要尝试自己进入running状态。*/
  8. zkClient.subscribeDataChanges(path, dataListener);

上述只是监听代码，之后尝试调用initRunning方法通过HA的方式来启动CanalInstance。

  1. private void initRunning() {
  2.     if (!isStart()) {
  3.         return;
  4.     }
  5.     //构建临时节点的路径：/otter/canal/destinations/{0}/running，其中占位符{0}会被destination替换
  6.     String path = ZookeeperPathUtils.getDestinationServerRunning(destination);
  7.     // 序列化
  8.     //构建临时节点的数据，标记当前destination由哪一个canal server处理
  9.     byte[] bytes = JsonUtils.marshalToByte(serverData);
  10.     try {
  11.         mutex.set(false);
  12.         //尝试创建临时节点。如果节点已经存在，说明是其他的canal server已经启动了这个canal instance。
  13.         //此时会抛出ZkNodeExistsException，进入catch代码块。
  14.         zkClient.create(path, bytes, CreateMode.EPHEMERAL);
  15.         activeData = serverData;
  16.         processActiveEnter();//如果创建成功，触发一下事件，内部调用ServerRunningListener的processActiveEnter方法
  17.         mutex.set(true);
  18.     } catch (ZkNodeExistsException e) {
  19.       //创建节点失败，则根据path从zk中获取当前是哪一个canal server创建了当前canal instance的相关信息。
  20.       //第二个参数true，表示的是，如果这个path不存在，则返回null。
  21.         bytes = zkClient.readData(path, true);
  22.         if (bytes == null) {// 如果不存在节点，立即尝试一次            
  23.             initRunning();
  24.         } else {
  25.         //如果的确存在，则将创建该canal instance实例信息存入activeData中。
  26.             activeData = JsonUtils.unmarshalFromByte(bytes, ServerRunningData.class);
  27.         }
  28.     } catch (ZkNoNodeException e) {//如果/otter/canal/destinations/{0}/节点不存在，进行创建其中占位符{0}会被destination替换
  29.         zkClient.createPersistent(ZookeeperPathUtils.getDestinationPath(destination), true); 
  30.        // 尝试创建父节点        
  31.         initRunning();
  32.     }
  33. }

可以看到，initRunning方法内部只有在尝试在zk中创建节点成功后，才会去调用listener的processActiveEnter方法来真正启动destination对应的canal instance，这是canal HA方式启动的核心。canal官方文档中介绍了CanalServer HA机制启动的流程，如下：

Image.png

事实上，这个说明的前两步，都是在initRunning方法中实现的。从上面的代码中，我们可以看出，在HA机启动的情况下，initRunning方法不一定能走到processActiveEnter()方法，因为创建临时节点可能会出错。

此外，根据官方文档说明，如果出错，那么当前canal instance则进入standBy状态。也就是另外一个canal instance出现异常时，当前canal instance顶上去。那么相关源码在什么地方呢？在HA方式启动最开始的2行代码的监听逻辑中：

  1. String path = ZookeeperPathUtils.getDestinationServerRunning(destination);
  2. zkClient.subscribeDataChanges(path, dataListener);

其中dataListener类型是IZkDataListener，这是zkclient客户端提供的接口，定义如下：

  1. public interface IZkDataListener {
  2.     public void handleDataChange(String dataPath, Object data) throws Exception;
  3.     public void handleDataDeleted(String dataPath) throws Exception;
  4. }

当zk节点中的数据发生变更时，会自动回调这两个方法，很明显，一个是用于处理节点数据发生变化，一个是用于处理节点数据被删除。

而dataListener是在ServerRunningMonitor的构造方法中初始化的，如下：

  1. public ServerRunningMonitor(){
  2.     // 创建父节点
  3.     dataListener = new IZkDataListener() {
  4.         //！！！目前看来，好像并没有存在修改running节点数据的代码，为什么这个方法不是空实现？
  5.         public void handleDataChange(String dataPath, Object data) throws Exception {
  6.             MDC.put("destination", destination);
  7.             ServerRunningData runningData = JsonUtils.unmarshalFromByte((byte[]) data, ServerRunningData.class);
  8.             if (!isMine(runningData.getAddress())) {
  9.                 mutex.set(false);
  10.             }
  11.  
  12.             if (!runningData.isActive() && isMine(runningData.getAddress())) { // 说明出现了主动释放的操作，并且本机之前是active                                   
  13.               release = true;
  14.                 releaseRunning();// 彻底释放mainstem            }
  15.  
  16.             activeData = (ServerRunningData) runningData;
  17.         }
  18.         //当其他canal instance出现异常，临时节点数据被删除时，会自动回调这个方法，此时当前canal instance要顶上去
  19.         public void handleDataDeleted(String dataPath) throws Exception {
  20.             MDC.put("destination", destination);
  21.             mutex.set(false);
  22.             if (!release && activeData != null && isMine(activeData.getAddress())) {
  23.                 // 如果上一次active的状态就是本机，则即时触发一下active抢占                
  24.                 initRunning();
  25.             } else {
  26.                 // 否则就是等待delayTime，避免因网络瞬端或者zk异常，导致出现频繁的切换操作                
  27.                 delayExector.schedule(new Runnable() {
  28.  
  29.                     public void run() {
  30.                         initRunning();//尝试自己进入running状态
  31.                     }
  32.                 }, delayTime, TimeUnit.SECONDS);
  33.             }
  34.         }
  35.  
  36.     };
  37.  
  38. }

那么现在问题来了？ServerRunningMonitor的start方法又是在哪里被调用的， 这个方法被调用了，才能真正的启动canal instance。这部分代码我们放到后面的CanalController中的start方法进行讲解。

下面分析最后一部分代码，autoScan机制相关代码。

3.5 autoScan机制相关代码

关于autoscan，官方文档有以下介绍：

Image.png

结合autoscan机制的相关源码：

  1. //   
  2. autoScan = BooleanUtils.toBoolean(getProperty(properties, CanalConstants.CANAL_AUTO_SCAN));
  3.         if (autoScan) {
  4.             defaultAction = new InstanceAction() {//....};
  5.  
  6.             instanceConfigMonitors = //....
  7.         }

可以看到，autoScan是否需要自动扫描的开关，只有当autoScan为true时，才会初始化defaultAction字段和instanceConfigMonitors字段。其中：

其中：

    defaultAction：其作用是如果配置发生了变更，默认应该采取什么样的操作。其实现了InstanceAction接口定义的三个抽象方法：start、stop和reload。当新增一个destination配置时，需要调用start方法来启动；当移除一个destination配置时，需要调用stop方法来停止；当某个destination配置发生变更时，需要调用reload方法来进行重启。

    instanceConfigMonitors：类型为Map<InstanceMode, InstanceConfigMonitor>。defaultAction字段只是定义了配置发生变化默认应该采取的操作，那么总该有一个类来监听配置是否发生了变化，这就是InstanceConfigMonitor的作用。官方文档中，只提到了对canal.conf.dir配置项指定的目录的监听，这指的是通过spring方式加载配置。显然的，通过manager方式加载配置，配置中心的内容也是可能发生变化的，也需要进行监听。此时可以理解为什么instanceConfigMonitors的类型是一个Map，key为InstanceMode，就是为了对这两种方式的配置加载方式都进行监听。

defaultAction字段初始化源码如下所示：

  1. defaultAction = new InstanceAction() {
  2.  
  3.     public void start(String destination) {
  4.         InstanceConfig config = instanceConfigs.get(destination);
  5.         if (config == null) {
  6.             // 重新读取一下instance config
  7.             config = parseInstanceConfig(properties, destination);
  8.             instanceConfigs.put(destination, config);
  9.         }
  10.  
  11.         if (!embededCanalServer.isStart(destination)) {
  12.             // HA机制启动
  13.             ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
  14.             if (!config.getLazy() && !runningMonitor.isStart()) {
  15.                 runningMonitor.start();
  16.             }
  17.         }
  18.     }
  19.  
  20.     public void stop(String destination) {
  21.         // 此处的stop，代表强制退出，非HA机制，所以需要退出HA的monitor和配置信息
  22.         InstanceConfig config = instanceConfigs.remove(destination);
  23.         if (config != null) {
  24.             embededCanalServer.stop(destination);
  25.             ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
  26.             if (runningMonitor.isStart()) {
  27.                 runningMonitor.stop();
  28.             }
  29.         }
  30.     }
  31.  
  32.     public void reload(String destination) {
  33.         // 目前任何配置变化，直接重启，简单处理
  34.         stop(destination);
  35.         start(destination);
  36.     }
  37. };

instanceConfigMonitors字段初始化源码如下所示：

  1. instanceConfigMonitors = MigrateMap.makeComputingMap(new Function<InstanceMode, InstanceConfigMonitor>() {
  2.        public InstanceConfigMonitor apply(InstanceMode mode) {
  3.            int scanInterval = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_AUTO_SCAN_INTERVAL));
  4.            if (mode.isSpring()) {//如果加载方式是spring，返回SpringInstanceConfigMonitor
  5.                SpringInstanceConfigMonitor monitor = new SpringInstanceConfigMonitor();
  6.                monitor.setScanIntervalInSecond(scanInterval);
  7.                monitor.setDefaultAction(defaultAction);
  8.                // 设置conf目录，默认是user.dir + conf目录组成
  9.                String rootDir = getProperty(properties, CanalConstants.CANAL_CONF_DIR);
  10.                if (StringUtils.isEmpty(rootDir)) {
  11.                    rootDir = "../conf";
  12.                }
  13.                if (StringUtils.equals("otter-canal", System.getProperty("appName"))) {
  14.                    monitor.setRootConf(rootDir);
  15.                } else {
  16.                    // eclipse debug模式
  17.                    monitor.setRootConf("src/main/resources/");
  18.                }
  19.                return monitor;
  20.            } else if (mode.isManager()) {//如果加载方式是manager，返回ManagerInstanceConfigMonitor
  21.                return new ManagerInstanceConfigMonitor();
  22.            } else {
  23.                throw new UnsupportedOperationException("unknow mode :" + mode + " for monitor");
  24.            }
  25.        }
  26.    });

可以看到instanceConfigMonitors也是根据mode属性，来采取不同的监控实现类SpringInstanceConfigMonitor 或者ManagerInstanceConfigMonitor，二者都实现了InstanceConfigMonitor接口。

  1. public interface InstanceConfigMonitor extends CanalLifeCycle {
  2.     void register(String destination, InstanceAction action);
  3.     void unregister(String destination);
  4. }

当需要对一个destination进行监听时，调用register方法

当取消对一个destination监听时，调用unregister方法。

事实上，unregister方法在canal 内部并没有有任何地方被调用，也就是说，某个destination如果开启了autoScan=true，那么你是无法在运行时停止对其进行监控的。如果要停止，你可以选择将对应的目录删除。

InstanceConfigMonitor本身并不知道哪些canal instance需要进行监控，因为不同的canal instance，有的可能设置autoScan为true，另外一些可能设置为false。

在CanalConroller的start方法中，对于autoScan为true的destination，会调用InstanceConfigMonitor的register方法进行注册，此时InstanceConfigMonitor才会真正的对这个destination配置进行扫描监听。对于那些autoScan为false的destination，则不会进行监听。

目前SpringInstanceConfigMonitor对这两个方法都进行了实现，而ManagerInstanceConfigMonitor目前对这两个方法实现的都是空，需要开发者自己来实现。

在实现ManagerInstanceConfigMonitor时，可以参考SpringInstanceConfigMonitor。

此处不打算再继续进行分析SpringInstanceConfigMonitor的源码，因为逻辑很简单，感兴趣的读者可以自行查看SpringInstanceConfigMonitor 的scan方法，内部在什么情况下会回调defualtAction的start、stop、reload方法 。

4 CanalController的start方法

而ServerRunningMonitor的start方法，是在CanalController中的start方法中被调用的，CanalController中的start方法是在CanalLauncher中被调用的。

com.alibaba.otter.canal.deployer.CanalController#start

  1. public void start() throws Throwable {
  2.         logger.info("## start the canal server[{}:{}]", ip, port);
  3.         // 创建整个canal的工作节点 :/otter/canal/cluster/{0}
  4.         final String path = ZookeeperPathUtils.getCanalClusterNode(ip + ":" + port);
  5.         initCid(path);
  6.         if (zkclientx != null) {
  7.             this.zkclientx.subscribeStateChanges(new IZkStateListener() {
  8.                 public void handleStateChanged(KeeperState state) throws Exception {
  9.                 }
  10.                 public void handleNewSession() throws Exception {
  11.                     initCid(path);
  12.                 }
  13.             });
  14.         }
  15.         // 优先启动embeded服务
  16.         embededCanalServer.start();
  17.         //启动不是lazy模式的CanalInstance，通过迭代instanceConfigs，根据destination获取对应的ServerRunningMonitor，然后逐一启动
  18.         for (Map.Entry<String, InstanceConfig> entry : instanceConfigs.entrySet()) {
  19.             final String destination = entry.getKey();
  20.             InstanceConfig config = entry.getValue();
  21.             // 如果destination对应的CanalInstance没有启动，则进行启动
  22.             if (!embededCanalServer.isStart(destination)) {
  23.                 ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
  24.                 //如果不是lazy，lazy模式需要等到第一次有客户端请求才会启动
  25.                 if (!config.getLazy() && !runningMonitor.isStart()) {
  26.                     runningMonitor.start();
  27.                 }
  28.             }
  29.             if (autoScan) {
  30.                 instanceConfigMonitors.get(config.getMode()).register(destination, defaultAction);
  31.             }
  32.         }
  33.         if (autoScan) {//启动配置文件自动检测机制
  34.             instanceConfigMonitors.get(globalInstanceConfig.getMode()).start();
  35.             for (InstanceConfigMonitor monitor : instanceConfigMonitors.values()) {
  36.                 if (!monitor.isStart()) {
  37.                     monitor.start();//启动monitor
  38.                 }
  39.             }
  40.         }
  41.         // 启动网络接口，监听客户端请求
  42.         canalServer.start();
  43.     }

5 总结

deployer模块的主要作用：

1、读取canal.properties，确定canal instance的配置加载方式

2、确定canal instance的启动方式：独立启动或者集群方式启动

3、监听canal instance的配置的变化，动态停止、启动或新增

4、启动canal server，监听客户端请求

 

分享到：
| 转发：1、canal源码分析简介
评论
发表评论

您还没有登录,请您登录后再发表评论

相关推荐

Magicbox
Global site tag (gtag.js) - Google Analytics