观察者模式，定义添加修改删除对应的操作
系统很多Monitor/Listener都是类似
Monitor内含listener，调用再触发
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();
}
}
private void processStop() {
if (listener != null) {
try {
listener.processStop();
} catch (Exception e) {
logger.error("processStop failed", e);
}
}
}
对生命周期中运行态的判断
protected volatile boolean running = false; // 是否处于运行中
public boolean isStart() {
return running;
}
数据库时间对齐 记录下 str_to_date(concat(date_format(DATE_SUB(now(),interval 1 DAY), '%Y-%m-%d'), ' 23:59:59.999'), '%Y-%m-%d %H:%i:%s.%f')
单例
private static class SingletonHolder {
private static final CanalServerWithEmbedded CANAL_SERVER_WITH_EMBEDDED = new CanalServerWithEmbedded();
}
public CanalServerWithEmbedded(){
// 希望也保留用户new单独实例的需求,兼容历史
}
public static CanalServerWithEmbedded instance() {
return SingletonHolder.CANAL_SERVER_WITH_EMBEDDED;
}
代理模式？CanalServerWithNetty起始就是CanalServerWithEmbedded套了个壳
源码中的自定义互斥类
捕获并处理一个线程对象中抛出的未检测异常，以避免程序终止
private static void setGlobalUncaughtExceptionHandler() {
Thread.setDefaultUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
@Override
public void uncaughtException(Thread t, Throwable e) {
logger.error("UnCaughtException", e);
}
});
}