From c0619c421723be4b2b4cb61c95cbeb3a2ade5c10 Mon Sep 17 00:00:00 2001 From: "jianghang.loujh" Date: Thu, 25 Dec 2025 11:54:25 +0800 Subject: [PATCH] merge code --- .../parse/driver/mysql/packets/MariaGtid.java | 1 - .../inbound/mysql/LocalBinLogConnection.java | 44 ++++++++----------- 2 files changed, 19 insertions(+), 26 deletions(-) diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/MariaGtid.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/MariaGtid.java index 37a2c518..743c9f07 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/MariaGtid.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/MariaGtid.java @@ -76,6 +76,5 @@ public class MariaGtid { sb.append(domainId).append("-"); sb.append(serverId).append("-"); return sb.append(sequence).toString(); - // return String.format("%s-%s-%s", domainId, serverId, sequence); } } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java index d596e644..8be5536c 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java @@ -18,11 +18,7 @@ import com.alibaba.otter.canal.parse.inbound.ErosaConnection; import com.alibaba.otter.canal.parse.inbound.MultiStageCoprocessor; import com.alibaba.otter.canal.parse.inbound.SinkFunction; import com.alibaba.otter.canal.parse.inbound.mysql.local.BinLogFileQueue; -import com.taobao.tddl.dbsync.binlog.FileLogFetcher; -import com.taobao.tddl.dbsync.binlog.LogContext; -import com.taobao.tddl.dbsync.binlog.LogDecoder; -import com.taobao.tddl.dbsync.binlog.LogEvent; -import com.taobao.tddl.dbsync.binlog.LogPosition; +import com.taobao.tddl.dbsync.binlog.*; import com.taobao.tddl.dbsync.binlog.event.QueryLogEvent; /** @@ -32,27 +28,29 @@ import com.taobao.tddl.dbsync.binlog.event.QueryLogEvent; */ public class LocalBinLogConnection implements ErosaConnection { - private static final Logger logger = LoggerFactory.getLogger(LocalBinLogConnection.class); - private BinLogFileQueue binlogs = null; + private static final Logger logger = LoggerFactory.getLogger(LocalBinLogConnection.class); + /** rdsOosMode 主从信息 */ + private final Set rdsOssMasterSlaveInfo = new HashSet<>(4); + private BinLogFileQueue binlogs = null; private boolean needWait; private String directory; - private int bufferSize = 16 * 1024; - private boolean running = false; + private int bufferSize = 16 * 1024; + private boolean running = false; private long serverId; - /** rdsOosMode binlog 的 serverId 是两个 */ - private boolean isRdsOssMode = false; - - /** rdsOosMode 主从信息 */ - private final Set rdsOssMasterSlaveInfo = new HashSet<>(4); - - private boolean firstUpdateRdsOssMasterSlave = true; + private boolean isRdsOssMode = false; + private boolean firstUpdateRdsOssMasterSlave = true; private FileParserListener parserListener; public LocalBinLogConnection(){ } + public LocalBinLogConnection(String directory, boolean needWait){ + this.needWait = needWait; + this.directory = directory; + } + public boolean isRdsOssMode() { return isRdsOssMode; } @@ -61,11 +59,6 @@ public class LocalBinLogConnection implements ErosaConnection { isRdsOssMode = rdsOssMode; } - public LocalBinLogConnection(String directory, boolean needWait){ - this.needWait = needWait; - this.directory = directory; - } - @Override public void connect() throws IOException { if (this.binlogs == null) { @@ -117,7 +110,7 @@ public class LocalBinLogConnection implements ErosaConnection { List iterateEvents = decoder.processIterateDecode(event, context); if (!iterateEvents.isEmpty()) { // 处理compress event - for(LogEvent itEvent : iterateEvents) { + for (LogEvent itEvent : iterateEvents) { if (!func.sink(itEvent)) { needContinue = false; break; @@ -158,9 +151,10 @@ public class LocalBinLogConnection implements ErosaConnection { } /** - * 1. 非 rdsOos 模式下需要要校验 serverId 是否一致 防止解析其他实例的 binlog - * 2. rdsOos 高可用模式下解析 binlog 会有两个 serverId,分别对应着主从节点 binlog解析出来的 serverId - * 主从的关系可能会变, 但是 serverId一直都会是这两个 serverId + * 1. 非 rdsOos 模式下需要要校验 serverId 是否一致 防止解析其他实例的 binlog
+ * 2. rdsOos 高可用模式下解析 binlog + * 会有两个 serverId,分别对应着主从节点 binlog解析出来的 serverId 主从的关系可能会变, 但是 serverId一直都会是这两个 + * serverId * * @param event */