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

支持通过GTID同步binlog

Browse Source
This commit is contained in:
hiwjd
2018-05-02 15:28:11 +08:00
parent f46133d116
commit 21f3f8c973
22 changed files with 13479 additions and 12615 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -8,6 +8,8 @@ import java.util.TimerTask;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import com.alibaba.otter.canal.parse.driver.mysql.packets.MysqlGTIDSet;
import com.google.protobuf.InvalidProtocolBufferException;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.commons.lang.math.RandomUtils;
@@ -87,6 +89,9 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
protected TimerTask heartBeatTimerTask;
protected Throwable exception = null;
protected boolean isGTIDMode = false; // 是否是GTID模式
protected String GTIDSetStr = null;
protected abstract BinlogParser buildParser();
protected abstract ErosaConnection buildErosaConnection();
@@ -214,12 +219,17 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
};
// 4. 开始dump数据
if (StringUtils.isEmpty(startPosition.getJournalName()) && startPosition.getTimestamp() != null) {
erosaConnection.dump(startPosition.getTimestamp(), sinkHandler);
// 判断所属instance是否启用GTID模式，是的话调用ErosaConnection中GTID对应方法dump数据
if (isGTIDMode()) {
erosaConnection.dump(MysqlGTIDSet.parse(startPosition.getGtid()), sinkHandler);
} else {
erosaConnection.dump(startPosition.getJournalName(),
startPosition.getPosition(),
sinkHandler);
if (StringUtils.isEmpty(startPosition.getJournalName()) && startPosition.getTimestamp() != null) {
erosaConnection.dump(startPosition.getTimestamp(), sinkHandler);
} else {
erosaConnection.dump(startPosition.getJournalName(),
startPosition.getPosition(),
sinkHandler);
}
}
} catch (TableIdNotFoundException e) {
@@ -368,6 +378,9 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
position.setTimestamp(entry.getHeader().getExecuteTime());
// add serverId at 2016-06-28
position.setServerId(entry.getHeader().getServerId());
// set gtid
position.setGtid(entry.getHeader().getGtid());
logPosition.setPostion(position);
LogIdentity identity = new LogIdentity(runningInfo.getAddress(), -1L);
@@ -530,4 +543,20 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
return exception;
}
public boolean isGTIDMode() {
return isGTIDMode;
}
public void setIsGTIDMode(boolean isGTIDMode) {
this.isGTIDMode = isGTIDMode;
}
public String getGTIDSetStr() {
return GTIDSetStr;
}
public void setGTIDSetStr(String GTIDSetStr) {
this.GTIDSetStr = GTIDSetStr;
}
}

11
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/ErosaConnection.java
View File

@@ -1,5 +1,7 @@
package com.alibaba.otter.canal.parse.inbound;
import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
import java.io.IOException;
/**
@@ -24,5 +26,14 @@ public interface ErosaConnection {
public void dump(long timestamp, SinkFunction func) throws IOException;
/**
* 通过GTID同步binlog
*
* @param gtidSet
* @param func
* @throws IOException
*/
public void dump(GTIDSet gtidSet, SinkFunction func) throws IOException;
ErosaConnection fork();
}

18
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java
View File

@@ -1,7 +1,10 @@
package com.alibaba.otter.canal.parse.inbound.mysql;
import java.io.IOException;
import java.nio.charset.Charset;
import com.alibaba.otter.canal.parse.driver.mysql.packets.MysqlGTIDSet;
import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvertGTID;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -34,7 +37,20 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
protected boolean useDruidDdlFilter = true;
protected BinlogParser buildParser() {
LogEventConvert convert = new LogEventConvert();
LogEventConvert convert;
if (isGTIDMode()) {
EntryPosition position;
try {
position = findStartPosition(null);
} catch (IOException e) {
throw new RuntimeException("findStartPosition failed.", e);
}
convert = new LogEventConvertGTID(MysqlGTIDSet.parse(position.getGtid()));
} else {
convert = new LogEventConvert();
}
if (eventFilter != null && eventFilter instanceof AviaterRegexFilter) {
convert.setNameFilter((AviaterRegexFilter) eventFilter);
}

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

@@ -4,6 +4,8 @@ import java.io.File;
import java.io.IOException;
import java.util.List;
import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
import org.apache.commons.lang.NotImplementedException;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -200,6 +202,11 @@ public class LocalBinLogConnection implements ErosaConnection {
dump(binlogFilename, binlogFileOffset, func);
}
@Override
public void dump(GTIDSet gtidSet, SinkFunction func) throws IOException {
throw new NotImplementedException();
}
public ErosaConnection fork() {
LocalBinLogConnection connection = new LocalBinLogConnection();

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

@@ -8,6 +8,8 @@ import java.nio.charset.Charset;
import java.util.List;
import java.util.concurrent.TimeUnit;
import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
import com.alibaba.otter.canal.parse.driver.mysql.packets.client.BinlogDumpGTIDCommandPacket;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -159,6 +161,29 @@ public class MysqlConnection implements ErosaConnection {
}
}
@Override
public void dump(GTIDSet gtidSet, SinkFunction func) throws IOException {
updateSettings();
sendBinlogDumpGTID(gtidSet);
DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize());
fetcher.start(connector.getChannel());
LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
LogContext context = new LogContext();
while (fetcher.fetch()) {
LogEvent event = null;
event = decoder.decode(fetcher, context);
if (event == null) {
throw new CanalParseException("parse failed");
}
if (!func.sink(event)) {
break;
}
}
}
public void dump(long timestamp, SinkFunction func) throws IOException {
throw new NullPointerException("Not implement yet");
}
@@ -222,6 +247,20 @@ public class MysqlConnection implements ErosaConnection {
PacketManager.writePkg(connector.getChannel(), semiAckHeader.toBytes(), cmdBody);
}
private void sendBinlogDumpGTID(GTIDSet gtidSet) throws IOException {
BinlogDumpGTIDCommandPacket binlogDumpCmd = new BinlogDumpGTIDCommandPacket();
binlogDumpCmd.slaveServerId = this.slaveId;
binlogDumpCmd.gtidSet = gtidSet;
byte[] cmdBody = binlogDumpCmd.toBytes();
logger.info("COM_BINLOG_DUMP_GTID:{}", binlogDumpCmd);
HeaderPacket binlogDumpHeader = new HeaderPacket();
binlogDumpHeader.setPacketBodyLength(cmdBody.length);
binlogDumpHeader.setPacketSequenceNumber((byte) 0x00);
PacketManager.writePkg(connector.getChannel(), binlogDumpHeader.toBytes(), cmdBody);
connector.setDumping(true);
}
public MysqlConnection fork() {
MysqlConnection connection = new MysqlConnection();
connection.setCharset(getCharset());

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

@@ -10,6 +10,10 @@ import java.util.Map;
import java.util.TimerTask;
import java.util.concurrent.atomic.AtomicLong;
import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter;
import com.alibaba.otter.canal.parse.driver.mysql.packets.MysqlGTIDSet;
import com.alibaba.otter.canal.parse.inbound.BinlogParser;
import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvertGTID;
import org.apache.commons.lang.StringUtils;
import org.springframework.util.CollectionUtils;
@@ -344,6 +348,14 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
}
protected EntryPosition findStartPosition(ErosaConnection connection) throws IOException {
if (isGTIDMode()) {
// GTID模式下，CanalLogPositionManager里取最后的gtid，没有则取instanc配置中的
LogPosition logPosition = getLogPositionManager().getLatestIndexBy(destination);
if (logPosition != null) {
return logPosition.getPostion();
}
return masterPosition;
}
EntryPosition startPosition = findStartPositionInternal(connection);
if (needTransactionPosition.get()) {
logger.warn("prepare to find last position : {}", startPosition.toString());

12
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java
View File

@@ -56,6 +56,7 @@ import com.taobao.tddl.dbsync.binlog.event.UserVarLogEvent;
import com.taobao.tddl.dbsync.binlog.event.WriteRowsLogEvent;
import com.taobao.tddl.dbsync.binlog.event.XidLogEvent;
import com.taobao.tddl.dbsync.binlog.event.mariadb.AnnotateRowsEvent;
import com.taobao.tddl.dbsync.binlog.event.GtidLogEvent;
/**
* 基于{@linkplain LogEvent}转化为Entry对象的处理
@@ -128,6 +129,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
return parseIntrvarLogEvent((IntvarLogEvent) logEvent);
case LogEvent.RAND_EVENT:
return parseRandLogEvent((RandLogEvent) logEvent);
case LogEvent.GTID_LOG_EVENT:
return parseGTIDLogEvent((GtidLogEvent) logEvent);
default:
break;
}
@@ -143,6 +146,15 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
}
}
private Entry parseGTIDLogEvent(GtidLogEvent logEvent) {
LogHeader logHeader = logEvent.getHeader();
Header header = createHeader("", logHeader, "", "", EventType.GTID);
Pair.Builder builder = Pair.newBuilder();
builder.setKey("gtid");
builder.setValue(String.format("%s:%d", logEvent.getSid(), logEvent.getGno()));
return createEntry(header, EntryType.GTIDLOG, builder.build().toByteString());
}
private Entry parseQueryEvent(QueryLogEvent event, boolean isSeek) {
String queryString = event.getQuery();
if (StringUtils.endsWithIgnoreCase(queryString, BEGIN)) {

55
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvertGTID.java Normal file
View File

@@ -0,0 +1,55 @@
package com.alibaba.otter.canal.parse.inbound.mysql.dbsync;
import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.parse.inbound.BinlogParser;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.google.protobuf.InvalidProtocolBufferException;
import com.taobao.tddl.dbsync.binlog.LogEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Created by hiwjd on 2018/5/2.
* hiwjd0@gmail.com
*/
public class LogEventConvertGTID extends LogEventConvert implements BinlogParser<LogEvent> {
public static final Logger logger = LoggerFactory.getLogger(LogEventConvertGTID.class);
// latest gtid
private GTIDSet gtidSet;
public LogEventConvertGTID(GTIDSet gtidSet) {
this.gtidSet = gtidSet;
}
@Override
public CanalEntry.Entry parse(LogEvent event, boolean isSeek) throws CanalParseException {
CanalEntry.Entry entry = super.parse(event, isSeek);
if (entry == null) {
return null;
}
if (entry.getEntryType() == CanalEntry.EntryType.GTIDLOG) {
try {
CanalEntry.Pair pair = CanalEntry.Pair.parseFrom(entry.getStoreValue());
gtidSet.update(pair.getValue());
} catch (InvalidProtocolBufferException e) {
logger.error("retrive gtid failed.", e);
throw new CanalParseException("retrive gtid failed.", e);
}
return null;
}
if (gtidSet != null) {
String gtid = gtidSet.toString();
CanalEntry.Header header = entry.getHeader().toBuilder().setGtid(gtid).build();
entry = entry.toBuilder().setHeader(header).build();
}
return entry;
}
}