chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

optimize gtid

Browse Source
This commit is contained in:
jianghang.loujh
2023-09-07 09:48:58 +08:00
parent 34e9440383
commit d4a09704f2
6 changed files with 20 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/LogHeader.java
View File

@@ -326,7 +326,7 @@ public final class LogHeader {
if (gtidSet != null) { if (gtidSet != null) {
gtidMap.put(GTID_SET_STRING, gtidSet.toString()); gtidMap.put(GTID_SET_STRING, gtidSet.toString());
if (gtidEvent != null && gtidEvent instanceof GtidLogEvent) { if (gtidEvent != null && gtidEvent instanceof GtidLogEvent) {
GtidLogEvent event = (GtidLogEvent)gtidEvent; GtidLogEvent event = (GtidLogEvent) gtidEvent;
gtidMap.put(CURRENT_GTID_STRING, event.getGtidStr()); gtidMap.put(CURRENT_GTID_STRING, event.getGtidStr());
gtidMap.put(CURRENT_GTID_SN, String.valueOf(event.getSequenceNumber())); gtidMap.put(CURRENT_GTID_SN, String.valueOf(event.getSequenceNumber()));
gtidMap.put(CURRENT_GTID_LAST_COMMIT, String.valueOf(event.getLastCommitted())); gtidMap.put(CURRENT_GTID_LAST_COMMIT, String.valueOf(event.getLastCommitted()));

7
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/MariaGTIDSet.java
View File

@@ -1,11 +1,11 @@
package com.alibaba.otter.canal.parse.driver.mysql.packets; package com.alibaba.otter.canal.parse.driver.mysql.packets;
import org.apache.commons.lang.StringUtils;
import java.io.IOException; import java.io.IOException;
import java.util.HashMap; import java.util.HashMap;
import java.util.Map; import java.util.Map;
import org.apache.commons.lang.StringUtils;
/** /**
* 类 MariaGTIDSet.java 的实现 * 类 MariaGTIDSet.java 的实现
* *
@@ -13,8 +13,9 @@ import java.util.Map;
* @version 1.0.0 * @version 1.0.0
*/ */
public class MariaGTIDSet implements GTIDSet { public class MariaGTIDSet implements GTIDSet {
//MariaDB 10.0.2+ representation of Gtid //MariaDB 10.0.2+ representation of Gtid
Map<Long, MariaGtid> gtidMap = new HashMap<>(); private Map<Long, MariaGtid> gtidMap = new HashMap<>();
@Override @Override
public byte[] encode() throws IOException { public byte[] encode() throws IOException {

3
driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/GtidUtil.java
View File

@@ -15,7 +15,8 @@ public class GtidUtil {
public static GTIDSet parseGtidSet(String gtid, boolean isMariaDB) { public static GTIDSet parseGtidSet(String gtid, boolean isMariaDB) {
if (isMariaDB) { if (isMariaDB) {
return MariaGTIDSet.parse(gtid); return MariaGTIDSet.parse(gtid);
} else {
return MysqlGTIDSet.parse(gtid);
} }
return MysqlGTIDSet.parse(gtid);
} }
} }

19
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java
View File

@@ -1,16 +1,12 @@
package com.alibaba.otter.canal.parse.inbound; package com.alibaba.otter.canal.parse.inbound;
import static com.alibaba.otter.canal.parse.driver.mysql.utils.GtidUtil.parseGtidSet;
import java.io.IOException; import java.io.IOException;
import java.util.Arrays; import java.util.*;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicLong;
import com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection;
import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.exception.ExceptionUtils; import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.commons.lang.math.RandomUtils; import org.apache.commons.lang.math.RandomUtils;
@@ -25,7 +21,7 @@ import com.alibaba.otter.canal.parse.CanalEventParser;
import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet; import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
import com.alibaba.otter.canal.parse.exception.CanalParseException; import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.parse.exception.PositionNotFoundException; import com.alibaba.otter.canal.parse.exception.PositionNotFoundException;
import com.taobao.tddl.dbsync.binlog.exception.TableIdNotFoundException; import com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection;
import com.alibaba.otter.canal.parse.inbound.mysql.MysqlMultiStageCoprocessor; import com.alibaba.otter.canal.parse.inbound.mysql.MysqlMultiStageCoprocessor;
import com.alibaba.otter.canal.parse.index.CanalLogPositionManager; import com.alibaba.otter.canal.parse.index.CanalLogPositionManager;
import com.alibaba.otter.canal.parse.support.AuthenticationInfo; import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
@@ -38,8 +34,7 @@ import com.alibaba.otter.canal.protocol.position.LogIdentity;
import com.alibaba.otter.canal.protocol.position.LogPosition; import com.alibaba.otter.canal.protocol.position.LogPosition;
import com.alibaba.otter.canal.sink.CanalEventSink; import com.alibaba.otter.canal.sink.CanalEventSink;
import com.alibaba.otter.canal.sink.exception.CanalSinkException; import com.alibaba.otter.canal.sink.exception.CanalSinkException;
import com.taobao.tddl.dbsync.binlog.exception.TableIdNotFoundException;
import static com.alibaba.otter.canal.parse.driver.mysql.utils.GtidUtil.parseGtidSet;
/** /**
* 抽象的EventParser, 最大化共用mysql/oracle版本的实现 * 抽象的EventParser, 最大化共用mysql/oracle版本的实现
@@ -183,7 +178,7 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
} }
if (erosaConnection instanceof MysqlConnection) { if (erosaConnection instanceof MysqlConnection) {
isMariaDB = ((MysqlConnection)erosaConnection).isMariaDB(); isMariaDB = ((MysqlConnection) erosaConnection).isMariaDB();
} }
// 4. 获取最后的位置信息 // 4. 获取最后的位置信息
long start = System.currentTimeMillis(); long start = System.currentTimeMillis();
@@ -249,7 +244,7 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
multiStageCoprocessor = buildMultiStageCoprocessor(); multiStageCoprocessor = buildMultiStageCoprocessor();
if (isGTIDMode() && StringUtils.isNotEmpty(startPosition.getGtid())) { if (isGTIDMode() && StringUtils.isNotEmpty(startPosition.getGtid())) {
// 判断所属instance是否启用GTID模式，是的话调用ErosaConnection中GTID对应方法dump数据 // 判断所属instance是否启用GTID模式，是的话调用ErosaConnection中GTID对应方法dump数据
GTIDSet gtidSet = parseGtidSet(startPosition.getGtid(),isMariaDB); GTIDSet gtidSet = parseGtidSet(startPosition.getGtid(), isMariaDB);
((MysqlMultiStageCoprocessor) multiStageCoprocessor).setGtidSet(gtidSet); ((MysqlMultiStageCoprocessor) multiStageCoprocessor).setGtidSet(gtidSet);
multiStageCoprocessor.start(); multiStageCoprocessor.start();
erosaConnection.dump(gtidSet, multiStageCoprocessor); erosaConnection.dump(gtidSet, multiStageCoprocessor);

4
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java
View File

@@ -350,9 +350,9 @@ public class MysqlConnection implements ErosaConnection {
private void sendBinlogDumpGTID(GTIDSet gtidSet) throws IOException { private void sendBinlogDumpGTID(GTIDSet gtidSet) throws IOException {
if (isMariaDB()) { if (isMariaDB()) {
sendMariaBinlogDumpGTID(gtidSet); sendMariaBinlogDumpGTID(gtidSet);
return; } else {
sendMySQLBinlogDumpGTID(gtidSet);
} }
sendMySQLBinlogDumpGTID(gtidSet);
} }
private void sendMySQLBinlogDumpGTID(GTIDSet gtidSet) throws IOException { private void sendMySQLBinlogDumpGTID(GTIDSet gtidSet) throws IOException {

7
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java
View File

@@ -343,7 +343,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
protected EntryPosition findStartPosition(ErosaConnection connection) throws IOException { protected EntryPosition findStartPosition(ErosaConnection connection) throws IOException {
if (isGTIDMode()) { if (isGTIDMode()) {
// GTID模式下，CanalLogPositionManager里取最后的gtid，没有则取instanc配置中的 // GTID模式下，CanalLogPositionManager里取最后的gtid，没有则取instance配置中的
LogPosition logPosition = getLogPositionManager().getLatestIndexBy(destination); LogPosition logPosition = getLogPositionManager().getLatestIndexBy(destination);
if (logPosition != null) { if (logPosition != null) {
// 如果以前是非GTID模式，后来调整为了GTID模式，那么为了保持兼容，需要判断gtid是否为空 // 如果以前是非GTID模式，后来调整为了GTID模式，那么为了保持兼容，需要判断gtid是否为空
@@ -381,7 +381,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
protected EntryPosition findEndPositionWithMasterIdAndTimestamp(MysqlConnection connection) { protected EntryPosition findEndPositionWithMasterIdAndTimestamp(MysqlConnection connection) {
MysqlConnection mysqlConnection = (MysqlConnection) connection; MysqlConnection mysqlConnection = (MysqlConnection) connection;
final EntryPosition endPosition = findEndPosition(mysqlConnection); final EntryPosition endPosition = findEndPosition(mysqlConnection);
if (tableMetaTSDB != null) { if (tableMetaTSDB != null || isGTIDMode()) {
long startTimestamp = System.currentTimeMillis(); long startTimestamp = System.currentTimeMillis();
return findAsPerTimestampInSpecificLogFile(mysqlConnection, return findAsPerTimestampInSpecificLogFile(mysqlConnection,
startTimestamp, startTimestamp,
@@ -426,7 +426,8 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
} }
if (entryPosition == null) { if (entryPosition == null) {
entryPosition = findEndPositionWithMasterIdAndTimestamp(mysqlConnection); // 默认从当前最后一个位置进行消费 entryPosition =
findEndPositionWithMasterIdAndTimestamp(mysqlConnection); // 默认从当前最后一个位置进行消费
} }
// 判断一下是否需要按时间订阅 // 判断一下是否需要按时间订阅