本文主要研究一下canal的MysqlDetectingTimeTask
canal-1.1.4/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java
class MysqlDetectingTimeTask extends TimerTask {
private boolean reconnect = false;
private MysqlConnection mysqlConnection;
public MysqlDetectingTimeTask(MysqlConnection mysqlConnection){
this.mysqlConnection = mysqlConnection;
}
public void run() {
try {
if (reconnect) {
reconnect = false;
mysqlConnection.reconnect();
} else if (!mysqlConnection.isConnected()) {
mysqlConnection.connect();
}
Long startTime = System.currentTimeMillis();
// 可能心跳sql为select 1
if (StringUtils.startsWithIgnoreCase(detectingSQL.trim(), "select")
|| StringUtils.startsWithIgnoreCase(detectingSQL.trim(), "show")
|| StringUtils.startsWithIgnoreCase(detectingSQL.trim(), "explain")
|| StringUtils.startsWithIgnoreCase(detectingSQL.trim(), "desc")) {
mysqlConnection.query(detectingSQL);
} else {
mysqlConnection.update(detectingSQL);
}
Long costTime = System.currentTimeMillis() - startTime;
if (haController != null && haController instanceof HeartBeatCallback) {
((HeartBeatCallback) haController).onSuccess(costTime);
}
} catch (SocketTimeoutException e) {
if (haController != null && haController instanceof HeartBeatCallback) {
((HeartBeatCallback) haController).onFailed(e);
}
reconnect = true;
logger.warn("connect failed by ", e);
} catch (IOException e) {
if (haController != null && haController instanceof HeartBeatCallback) {
((HeartBeatCallback) haController).onFailed(e);
}
reconnect = true;
logger.warn("connect failed by ", e);
} catch (Throwable e) {
if (haController != null && haController instanceof HeartBeatCallback) {
((HeartBeatCallback) haController).onFailed(e);
}
reconnect = true;
logger.warn("connect failed by ", e);
}
}
public MysqlConnection getMysqlConnection() {
return mysqlConnection;
}
}
canal-1.1.4/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/HeartBeatCallback.java
public interface HeartBeatCallback {
/**
* 心跳发送成功
*/
public void onSuccess(long costTime);
/**
* 心跳发送失败
*/
public void onFailed(Throwable e);
}
canal-1.1.4/parse/src/main/java/com/alibaba/otter/canal/parse/ha/HeartBeatHAController.java
public class HeartBeatHAController extends AbstractCanalLifeCycle implements CanalHAController, HeartBeatCallback {
private static final Logger logger = LoggerFactory.getLogger(HeartBeatHAController.class);
// default 3 times
private int detectingRetryTimes = 3;
private int failedTimes = 0;
private boolean switchEnable = false;
private CanalHASwitchable eventParser;
public HeartBeatHAController(){
}
public void onSuccess(long costTime) {
failedTimes = 0;
}
public void onFailed(Throwable e) {
failedTimes++;
// 检查一下是否超过失败次数
synchronized (this) {
if (failedTimes > detectingRetryTimes) {
if (switchEnable) {
eventParser.doSwitch();// 通知执行一次切换
failedTimes = 0;
} else {
logger.warn("HeartBeat failed Times:{} , should auto switch ?", failedTimes);
}
}
}
}
// ============================= setter / getter
// ============================
public void setCanalHASwitchable(CanalHASwitchable canalHASwitchable) {
this.eventParser = canalHASwitchable;
}
public void setDetectingRetryTimes(int detectingRetryTimes) {
this.detectingRetryTimes = detectingRetryTimes;
}
public void setSwitchEnable(boolean switchEnable) {
this.switchEnable = switchEnable;
}
}
canal-1.1.4/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java
public class MysqlEventParser extends AbstractMysqlEventParser implements CanalEventParser, CanalHASwitchable {
//......
// 处理主备切换的逻辑
public void doSwitch() {
AuthenticationInfo newRunningInfo = (runningInfo.equals(masterInfo) ? standbyInfo : masterInfo);
this.doSwitch(newRunningInfo);
}
public void doSwitch(AuthenticationInfo newRunningInfo) {
// 1. 需要停止当前正在复制的过程
// 2. 找到新的position点
// 3. 重新建立链接,开始复制数据
// 切换ip
String alarmMessage = null;
if (this.runningInfo.equals(newRunningInfo)) {
alarmMessage = "same runingInfo switch again : " + runningInfo.getAddress().toString();
logger.warn(alarmMessage);
return;
}
if (newRunningInfo == null) {
alarmMessage = "no standby config, just do nothing, will continue try:"
+ runningInfo.getAddress().toString();
logger.warn(alarmMessage);
sendAlarm(destination, alarmMessage);
return;
} else {
stop();
alarmMessage = "try to ha switch, old:" + runningInfo.getAddress().toString() + ", new:"
+ newRunningInfo.getAddress().toString();
logger.warn(alarmMessage);
sendAlarm(destination, alarmMessage);
runningInfo = newRunningInfo;
start();
}
}
//......
}
MysqlDetectingTimeTask继承了TimerTask,其run方法在reconnect为true时,更新reconnect为false,然后执行mysqlConnection.reconnect();若reconnect为false且mysqlConnection.isConnected()为false则执行mysqlConnection.connect();之后执行detectingSQL语句,然后记录costTime,最后执行((HeartBeatCallback) haController).onSuccess(costTime);依次捕获SocketTimeoutException、IOException、Throwable,然后则执行((HeartBeatCallback) haController).onFailed(e),更新reconnect为true