标签：版本 net change val ring nec failed com oid
本文主要解析下canal server的启动过程，希望能有所收获。
整个server启动的过程比较复杂，看图难以理解，需要辅以文字说明。
首先程序的入口在CanalLauncher的main方法中。
String conf = System.getProperty("canal.conf", "classpath:canal.properties");
Properties properties = new Properties();
if (conf.startsWith(CLASSPATH_URL_PREFIX)) {
conf = StringUtils.substringAfter(conf, CLASSPATH_URL_PREFIX);
properties.load(CanalLauncher.class.getClassLoader().getResourceAsStream(conf));
} else {
properties.load(new FileInputStream(conf));
}从canal.properties文件中load所有的配置信息，加载到上下文中。不再赘述。
根据配置文件来构造CanalController，这块的代码比较多，主要分为七个步骤，具体如下。
调用initGlobalConfig方法，过程如下：
这块逻辑在CanalController的initGlobalConfig方法中。
这块的逻辑是从instance.properties里面初始化实例。
private void initInstanceConfig(Properties properties) {
String destinationStr = getProperty(properties, CanalConstants.CANAL_DESTINATIONS);
String[] destinations = StringUtils.split(destinationStr, CanalConstants.CANAL_DESTINATION_SPLIT);
for (String destination : destinations) {
InstanceConfig config = parseInstanceConfig(properties, destination);
InstanceConfig oldConfig = instanceConfigs.put(destination, config);
if (oldConfig != null) {
logger.warn("destination:{} old config:{} has replace by new config:{}", new Object[] { destination,
oldConfig, config });
}
}
}从这段代码中可以看出，我们在一个canal.properties文件中，可以配置多个destination，也就是可以配置多个instance，不同的instance以逗号隔开。这里主要看的是parseInstanceConfig()方法，里面的逻辑如下：
从配置文件中获取canal.socketChannel字段，放到全局变量中。
从配置文件中分别获取canal.id、ip、port（对外提供socket服务的端口），获取一个内存级的server单例，同时也获取一个对外提供Netty服务的单例。
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);从配置文件中获取zk地址(canal.zkServers)，启动一个zk客户端，然后初始化两个系统目录，分别是：
根据destination构造运行时监控，其实就是根据instance名来构造ServerRunningMonitor。其实就是实现了ServerRunningListener中的一些方法。
public interface ServerRunningListener {
/**
* 启动时回调做点事情
*/
public void processStart();
/**
* 关闭时回调做点事情
*/
public void processStop();
/**
* 触发现在轮到自己做为active，需要载入上一个active的上下文数据
*/
public void processActiveEnter();
/**
* 触发一下当前active模式失败
*/
public void processActiveExit();
}然后初始化一下ServerRunningMonitor。
runningMonitor.init();这个init方法跟踪的结果，其实就是执行了ServerRunningListener中的processStart方法。
public void processStart() {
try {
if (zkclientx != null) {
final String path = ZookeeperPathUtils.getDestinationClusterNode(destination, ip + ":" + port);
initCid(path);
zkclientx.subscribeStateChanges(new IZkStateListener() {
public void handleStateChanged(KeeperState state) throws Exception {
}
public void handleNewSession() throws Exception {
initCid(path);
}
@Override
public void handleSessionEstablishmentError(Throwable error) throws Exception {
logger.error("failed to connect to zookeeper", error);
}
});
}
} finally {
MDC.remove(CanalConstants.MDC_DESTINATION);
}
}首先获取了/otter/canal/destinations/{destination}/cluster/ip:port的内容，其实就是server的地址，最后一个ip:port是个zk的临时节点。然后订阅一下节点事件，当节点有事件推送过来后，做一些动作。
如果canal.auto.scan配置为true（默认为true），首先定义一个InstanceAction，包含了启动、停止、重启instance的动作。
定义一个SpringInstanceConfigMonitor，配置定时扫描的事件为canal.auto.scan.interval，默认5s，扫描canal.conf.dir目录下的文件，与上面定义的InstanceAction结合起来。
上面的构造方法其实就是定义一些必要的内容，真正的启动在这个方法中。
创建临时节点/otter/canal/cluster/ip:port，同时启动监听器.
embededCanalServer.start();这个start里面，一个是将当前server的running状态置为true，同时根据destination构建CanalInstance。
遍历Map
ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(destination);
if (!config.getLazy() && !runningMonitor.isStart()) {
runningMonitor.start();
}
public synchronized void start() {
super.start();
try {
processStart();
if (zkClient != null) {
// 如果需要尽可能释放instance资源，不需要监听running节点，不然即使stop了这台机器，另一台机器立马会start
String path = ZookeeperPathUtils.getDestinationServerRunning(destination);
zkClient.subscribeDataChanges(path, dataListener);
initRunning();
} else {
processActiveEnter();// 没有zk，直接启动
}
} catch (Exception e) {
logger.error("start failed", e);
// 没有正常启动，重置一下状态，避免干扰下一次start
stop();
}
}这里面启动的内容我们来看看。
监听路径/otter/canal/destinations/{destination}/running节点的变化
zkClient.subscribeDataChanges(path, dataListener);其实除了监听器，在本身的ServerRunningMonitor的start方法中，也有initRunning方法。这块启动canalInstance的方法，我们下一篇文章分析。
在扫描之前，把destination和InstanceAction绑定到缓存中。
instanceConfigMonitors.get(config.getMode()).register(destination, defaultAction);首先启动一个全局扫描，然后再对应的destination配置文件的扫描。
if (autoScan) {
instanceConfigMonitors.get(globalInstanceConfig.getMode()).start();
for (InstanceConfigMonitor monitor : instanceConfigMonitors.values()) {
if (!monitor.isStart()) {
monitor.start();
}
}
}这个start方法启动了一个定时器，默认5s扫描一次。扫描的内容就是配置文件路径下的内容，针对文件的新增、删除、修改，对应InstanceAction中的start，stop和reload方法。也就是说，我们在canal运行的过程中，通过动态修改配置文件，来实现动态调整运行时参数，主要可以用来进行重复消费，位点的迁移等等。
CanalServerWithNetty的启动，首先需要启动CanalServerWithEmbedded，主要的业务逻辑在SessionHandler中。这块其实是暴露外部服务，给canal client进行调用。
Runtime.getRuntime().addShutdownHook(new Thread() {
public void run() {
try {
logger.info("## stop the canal server");
controller.stop();
} catch (Throwable e) {
logger.warn("##something goes wrong when stopping canal Server:", e);
} finally {
logger.info("## canal server is down.");
}
}
});在server停止时，调用controller.stop()方法。
public void stop() throws Throwable {
canalServer.stop();
if (autoScan) {
for (InstanceConfigMonitor monitor : instanceConfigMonitors.values()) {
if (monitor.isStart()) {
monitor.stop();
}
}
}
for (ServerRunningMonitor runningMonitor : ServerRunningMonitors.getRunningMonitors().values()) {
if (runningMonitor.isStart()) {
runningMonitor.stop();
}
}
// 释放canal的工作节点
releaseCid(ZookeeperPathUtils.getCanalClusterNode(ip + ":" + port));
logger.info("## stop the canal server[{}:{}]", ip, port);
if (zkclientx != null) {
zkclientx.close();
}
}主要是停止controller，server相关的monitor，instance相关的monitor，然后释放zk节点，关闭zk连接。
【Canal源码分析】Canal Server的启动和停止过程
标签：版本 net change val ring nec failed com oid
原文地址：https://www.cnblogs.com/f-zhao/p/9083099.html
