启动入口CanalLauncher,假定intance为newAssert
CanalController
初始化
1.初始化instance config
2.new出embededCanalServer和canalServerWithNetty
3.初始化zkclientx
4.初始化ServerRunningMonitors,为每一个instance创建一个serverRunningMonitor,serverRunningMonitor里面的核心组件是serverRunningListener,这个listener封装了embededCanalServer的相关具体操作
5.如果是自动扫描,初始化instanceAction和instanceConfigMonitors
start
embededCanalServer.start();
这个不是真正的启动,只是生成了两个canalInstance
我们从上篇整体架构中已经知道,生成canalInstance有两种方式,远程获取canal配置或者加载本地spring配置,默认使用的是本地spring配置。
生成instance的时候同时生成了它包含的组件
`
public interface CanalInstance extends CanalLifeCycle {
String getDestination();
CanalEventParser getEventParser();
CanalEventSink getEventSink();
CanalEventStore getEventStore();
CanalMetaManager getMetaManager();
CanalAlarmHandler getAlarmHandler();
/**
* 客户端发生订阅/取消订阅行为
*/
boolean subscribeChange(ClientIdentity identity);
}
`
真正的启动在serverRunningMonitor
public void start() {
super.start();
processStart();
if (zkClient != null) {
// 如果需要尽可能释放instance资源,不需要监听running节点，不然即使stop了这台机器,另一台机器立马会start
String path = ZookeeperPathUtils.getDestinationServerRunning(destination);
zkClient.subscribeDataChanges(path, dataListener);
initRunning();
} else {
processActiveEnter();// 没有zk,直接启动
}
}
如上图,
前置逻辑--processStart
public void processStart() {
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);
}
});
}
先initCid,即初始化了如下zk节点
/otter/canal/destinations/newAsset/cluster/10.254.124.165:11111
并订阅该节点的状态变化
监听running节点
我们先看下running节点的内容
监听指定instance的running节点
/otter/canal/destinations/newAsset/running
zkClient.subscribeDataChanges(path, dataListener)
我们看下这个
dataListenerdataListener = new IZkDataListener() {
public void handleDataChange(String dataPath, Object data) throws Exception {
MDC.put("destination", destination);
ServerRunningData runningData = JsonUtils.unmarshalFromByte((byte[]) data, ServerRunningData.class);
if (!isMine(runningData.getAddress())) {
mutex.set(false);
}
if (!runningData.isActive() && isMine(runningData.getAddress())) { // 说明出现了主动释放的操作，并且本机之前是active
release = true;
releaseRunning();// 彻底释放mainstem
}
activeData = (ServerRunningData) runningData;
}
public void handleDataDeleted(String dataPath) throws Exception {
MDC.put("destination", destination);
mutex.set(false);
if (!release && activeData != null && isMine(activeData.getAddress())) {
// 如果上一次active的状态就是本机，则即时触发一下active抢占
initRunning();
} else {
// 否则就是等待delayTime，避免因网络瞬端或者zk异常，导致出现频繁的切换操作
delayExector.schedule(new Runnable() {
public void run() {
initRunning();
}
}, delayTime, TimeUnit.SECONDS);
}
}
};
initRunning
private void initRunning() {
if (!isStart()) {
return;
}
String path = ZookeeperPathUtils.getDestinationServerRunning(destination);
// 序列化
byte[] bytes = JsonUtils.marshalToByte(serverData);
try {
mutex.set(false);
//创建zk的runinning节点,写入serverData
zkClient.create(path, bytes, CreateMode.EPHEMERAL);
activeData = serverData;
//轮到自己作为activie,需要载入上一个active的上下文数据
processActiveEnter();
mutex.set(true);
} catch (ZkNodeExistsException e) {
}
}
很显然,对于newAsset的running节点的订阅和监控,可以使得newAsset在多个节点漂移
我们看下这个processActiveEnter,这个是ServerRunningListener的方法
public void processActiveEnter() {
try {
MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
embededCanalServer.start(destination);
} finally {
MDC.remove(CanalConstants.MDC_DESTINATION);
}
}
终于开始启动实例了。。。。
我们看看canalInstance有哪些组件
public interface CanalInstance extends CanalLifeCycle {
String getDestination();
CanalEventParser getEventParser();
CanalEventSink getEventSink();
CanalEventStore getEventStore();
CanalMetaManager getMetaManager();
CanalAlarmHandler getAlarmHandler();
/**
* 客户端发生订阅/取消订阅行为
*/
boolean subscribeChange(ClientIdentity identity);
}
总结
canalLauncher->serverRunningMonitor->serverRunningListener->embededServer->canalInstance
