extends AbstractCanalLifeCycle
public void setIsGTIDMode(boolean isGTIDMode) {
this.isGTIDMode = isGTIDMode;
}
+
+ public boolean isParallel() {
+ return parallel;
+ }
+
+ public void setParallel(boolean parallel) {
+ this.parallel = parallel;
+ }
+
+ public int getParallelThreadSize() {
+ return parallelThreadSize;
+ }
+
+ public void setParallelThreadSize(int parallelThreadSize) {
+ this.parallelThreadSize = parallelThreadSize;
+ }
+
+ public int getParallelBufferSize() {
+ return parallelBufferSize;
+ }
+
+ public void setParallelBufferSize(int parallelBufferSize) {
+ this.parallelBufferSize = parallelBufferSize;
+ }
+
}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/ErosaConnection.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/ErosaConnection.java
index 33f696ee..6beb4a9e 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/ErosaConnection.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/ErosaConnection.java
@@ -28,12 +28,16 @@ public interface ErosaConnection {
/**
* 通过GTID同步binlog
- *
- * @param gtidSet
- * @param func
- * @throws IOException
*/
public void dump(GTIDSet gtidSet, SinkFunction func) throws IOException;
+ // -------------
+
+ public void dump(String binlogfilename, Long binlogPosition, MultiStageCoprocessor coprocessor) throws IOException;
+
+ public void dump(long timestamp, MultiStageCoprocessor coprocessor) throws IOException;
+
+ public void dump(GTIDSet gtidSet, MultiStageCoprocessor coprocessor) throws IOException;
+
ErosaConnection fork();
}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/MultiStageCoprocessor.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/MultiStageCoprocessor.java
new file mode 100644
index 00000000..1c5c495d
--- /dev/null
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/MultiStageCoprocessor.java
@@ -0,0 +1,29 @@
+package com.alibaba.otter.canal.parse.inbound;
+
+import com.alibaba.otter.canal.common.CanalLifeCycle;
+import com.taobao.tddl.dbsync.binlog.LogBuffer;
+
+/**
+ * 针对解析器提供一个多阶段协同的处理
+ *
+ *
+ * 1. 网络接收 (单线程)
+ * 2. 事件基本解析 (单线程，事件类型、DDL解析构造TableMeta、维护位点信息)
+ * 3. 事件深度解析 (多线程, DML事件数据的完整解析)
+ * 4. 投递到store (单线程)
+ *
+ *
+ * @author agapple 2018年7月3日 下午4:54:17
+ * @since 1.0.26
+ */
+public interface MultiStageCoprocessor extends CanalLifeCycle {
+
+ /**
+ * 网络数据投递
+ */
+ public void publish(LogBuffer buffer);
+
+ public void publish(LogBuffer buffer, String binlogFileName);
+
+ public void reset();
+}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java
index 3b0e605d..8188cf83 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java
@@ -11,6 +11,7 @@ import com.alibaba.otter.canal.parse.driver.mysql.packets.MysqlGTIDSet;
import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.parse.inbound.AbstractEventParser;
import com.alibaba.otter.canal.parse.inbound.BinlogParser;
+import com.alibaba.otter.canal.parse.inbound.MultiStageCoprocessor;
import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert;
import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDB;
import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDBBuilder;
@@ -117,6 +118,14 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
}
}
+ protected MultiStageCoprocessor buildMultiStageCoprocessor() {
+ return new MysqlMultiStageCoprocessor(parallelBufferSize,
+ parallelThreadSize,
+ (LogEventConvert) binlogParser,
+ transactionBuffer,
+ destination);
+ }
+
// ============================ setter / getter =========================
public void setConnectionCharsetNumber(byte connectionCharsetNumber) {
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 3c882eab..f97c4ccd 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
@@ -11,9 +11,11 @@ import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.parse.driver.mysql.packets.GTIDSet;
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.LogBuffer;
import com.taobao.tddl.dbsync.binlog.LogContext;
import com.taobao.tddl.dbsync.binlog.LogDecoder;
import com.taobao.tddl.dbsync.binlog.LogEvent;
@@ -207,6 +209,128 @@ public class LocalBinLogConnection implements ErosaConnection {
throw new NotImplementedException();
}
+ @Override
+ public void dump(String binlogfilename, Long binlogPosition, MultiStageCoprocessor coprocessor) throws IOException {
+ File current = new File(directory, binlogfilename);
+
+ FileLogFetcher fetcher = new FileLogFetcher(bufferSize);
+ try {
+ fetcher.open(current, binlogPosition);
+ while (running) {
+ boolean needContinue = true;
+ while (fetcher.fetch()) {
+ LogBuffer buffer = fetcher.duplicate();
+ fetcher.consume(fetcher.limit());
+ coprocessor.publish(buffer, binlogfilename); // set filename
+ }
+
+ if (needContinue) {// 读取下一个
+ fetcher.close(); // 关闭上一个文件
+
+ File nextFile;
+ if (needWait) {
+ nextFile = binlogs.waitForNextFile(current);
+ } else {
+ nextFile = binlogs.getNextFile(current);
+ }
+
+ if (nextFile == null) {
+ break;
+ }
+
+ current = nextFile;
+ fetcher.open(current);
+ binlogfilename = nextFile.getName();
+ } else {
+ break;// 跳出
+ }
+ }
+ } catch (InterruptedException e) {
+ logger.warn("LocalBinLogConnection dump interrupted");
+ } finally {
+ if (fetcher != null) {
+ fetcher.close();
+ }
+ }
+ }
+
+ @Override
+ public void dump(long timestampMills, MultiStageCoprocessor coprocessor) throws IOException {
+ List currentBinlogs = binlogs.currentBinlogs();
+ File current = currentBinlogs.get(currentBinlogs.size() - 1);
+ long timestampSeconds = timestampMills / 1000;
+
+ String binlogFilename = null;
+ long binlogFileOffset = 0;
+
+ FileLogFetcher fetcher = new FileLogFetcher(bufferSize);
+ LogDecoder decoder = new LogDecoder();
+ decoder.handle(LogEvent.FORMAT_DESCRIPTION_EVENT);
+ decoder.handle(LogEvent.QUERY_EVENT);
+ decoder.handle(LogEvent.XID_EVENT);
+ LogContext context = new LogContext();
+ try {
+ fetcher.open(current);
+ context.setLogPosition(new LogPosition(current.getName()));
+ while (running) {
+ boolean needContinue = true;
+ String lastXidLogFilename = current.getName();
+ long lastXidLogFileOffset = 0;
+
+ binlogFilename = lastXidLogFilename;
+ binlogFileOffset = lastXidLogFileOffset;
+ while (fetcher.fetch()) {
+ LogEvent event = decoder.decode(fetcher, context);
+ if (event != null) {
+ if (event.getWhen() > timestampSeconds) {
+ break;
+ }
+
+ needContinue = false;
+ if (LogEvent.QUERY_EVENT == event.getHeader().getType()) {
+ if (StringUtils.endsWithIgnoreCase(((QueryLogEvent) event).getQuery(), "BEGIN")) {
+ binlogFilename = lastXidLogFilename;
+ binlogFileOffset = lastXidLogFileOffset;
+ } else if (StringUtils.endsWithIgnoreCase(((QueryLogEvent) event).getQuery(), "COMMIT")) {
+ lastXidLogFilename = current.getName();
+ lastXidLogFileOffset = event.getLogPos();
+ }
+ } else if (LogEvent.XID_EVENT == event.getHeader().getType()) {
+ lastXidLogFilename = current.getName();
+ lastXidLogFileOffset = event.getLogPos();
+ }
+ }
+ }
+
+ if (needContinue) {// 读取下一个
+ fetcher.close(); // 关闭上一个文件
+
+ File nextFile = binlogs.getBefore(current);
+ if (nextFile == null) {
+ break;
+ }
+
+ current = nextFile;
+ fetcher.open(current);
+ context.setLogPosition(new LogPosition(current.getName()));
+ } else {
+ break;// 跳出
+ }
+ }
+ } finally {
+ if (fetcher != null) {
+ fetcher.close();
+ }
+ }
+
+ dump(binlogFilename, binlogFileOffset, coprocessor);
+ }
+
+ @Override
+ public void dump(GTIDSet gtidSet, MultiStageCoprocessor coprocessor) throws IOException {
+ throw new NotImplementedException();
+ }
+
public ErosaConnection fork() {
LocalBinLogConnection connection = new LocalBinLogConnection();
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java
index 0e2194ae..46f7e300 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java
@@ -26,9 +26,11 @@ import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ResultSetPacket
import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager;
import com.alibaba.otter.canal.parse.exception.CanalParseException;
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.dbsync.DirectLogFetcher;
import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
+import com.taobao.tddl.dbsync.binlog.LogBuffer;
import com.taobao.tddl.dbsync.binlog.LogContext;
import com.taobao.tddl.dbsync.binlog.LogDecoder;
import com.taobao.tddl.dbsync.binlog.LogEvent;
@@ -156,7 +158,7 @@ public class MysqlConnection implements ErosaConnection {
}
if (event.getSemival() == 1) {
- sendSemiAck(context.getLogPosition().getFileName(), binlogPosition);
+ sendSemiAck(context.getLogPosition().getFileName(), context.getLogPosition().getPosition());
}
}
}
@@ -167,20 +169,24 @@ public class MysqlConnection implements ErosaConnection {
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);
+ try {
+ 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 (event == null) {
+ throw new CanalParseException("parse failed");
+ }
- if (!func.sink(event)) {
- break;
+ if (!func.sink(event)) {
+ break;
+ }
}
+ } finally {
+ fetcher.close();
}
}
@@ -188,6 +194,49 @@ public class MysqlConnection implements ErosaConnection {
throw new NullPointerException("Not implement yet");
}
+ @Override
+ public void dump(String binlogfilename, Long binlogPosition, MultiStageCoprocessor coprocessor) throws IOException {
+ updateSettings();
+ sendRegisterSlave();
+ sendBinlogDump(binlogfilename, binlogPosition);
+ ((MysqlMultiStageCoprocessor) coprocessor).setConnection(this);
+ DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize());
+ try {
+ fetcher.start(connector.getChannel());
+ while (fetcher.fetch()) {
+ LogBuffer buffer = fetcher.duplicate();
+ fetcher.consume(fetcher.limit());
+ coprocessor.publish(buffer);
+ }
+ } finally {
+ fetcher.close();
+ }
+ }
+
+ @Override
+ public void dump(long timestamp, MultiStageCoprocessor coprocessor) throws IOException {
+ throw new NullPointerException("Not implement yet");
+ }
+
+ @Override
+ public void dump(GTIDSet gtidSet, MultiStageCoprocessor coprocessor) throws IOException {
+ updateSettings();
+ sendBinlogDumpGTID(gtidSet);
+
+ ((MysqlMultiStageCoprocessor) coprocessor).setConnection(this);
+ DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize());
+ try {
+ fetcher.start(connector.getChannel());
+ while (fetcher.fetch()) {
+ LogBuffer buffer = fetcher.duplicate();
+ fetcher.consume(fetcher.limit());
+ coprocessor.publish(buffer);
+ }
+ } finally {
+ fetcher.close();
+ }
+ }
+
private void sendRegisterSlave() throws IOException {
RegisterSlaveCommandPacket cmd = new RegisterSlaveCommandPacket();
cmd.reportHost = authInfo.getAddress().getAddress().getHostAddress();
@@ -233,7 +282,7 @@ public class MysqlConnection implements ErosaConnection {
connector.setDumping(true);
}
- private void sendSemiAck(String binlogfilename, Long binlogPosition) throws IOException {
+ public void sendSemiAck(String binlogfilename, Long binlogPosition) throws IOException {
SemiAckCommandPacket semiAckCmd = new SemiAckCommandPacket();
semiAckCmd.binlogFileName = binlogfilename;
semiAckCmd.binlogPosition = binlogPosition;
@@ -538,4 +587,5 @@ public class MysqlConnection implements ErosaConnection {
public void setAuthInfo(AuthenticationInfo authInfo) {
this.authInfo = authInfo;
}
+
}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java
new file mode 100644
index 00000000..6e02e22d
--- /dev/null
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java
@@ -0,0 +1,391 @@
+package com.alibaba.otter.canal.parse.inbound.mysql;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import org.apache.commons.lang.StringUtils;
+
+import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
+import com.alibaba.otter.canal.common.utils.NamedThreadFactory;
+import com.alibaba.otter.canal.parse.exception.CanalParseException;
+import com.alibaba.otter.canal.parse.inbound.ErosaConnection;
+import com.alibaba.otter.canal.parse.inbound.EventTransactionBuffer;
+import com.alibaba.otter.canal.parse.inbound.MultiStageCoprocessor;
+import com.alibaba.otter.canal.parse.inbound.TableMeta;
+import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert;
+import com.alibaba.otter.canal.protocol.CanalEntry;
+import com.lmax.disruptor.BatchEventProcessor;
+import com.lmax.disruptor.BlockingWaitStrategy;
+import com.lmax.disruptor.EventFactory;
+import com.lmax.disruptor.EventHandler;
+import com.lmax.disruptor.FatalExceptionHandler;
+import com.lmax.disruptor.LifecycleAware;
+import com.lmax.disruptor.RingBuffer;
+import com.lmax.disruptor.Sequence;
+import com.lmax.disruptor.SequenceBarrier;
+import com.lmax.disruptor.WorkHandler;
+import com.lmax.disruptor.WorkerPool;
+import com.taobao.tddl.dbsync.binlog.LogBuffer;
+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.event.DeleteRowsLogEvent;
+import com.taobao.tddl.dbsync.binlog.event.RowsLogEvent;
+import com.taobao.tddl.dbsync.binlog.event.UpdateRowsLogEvent;
+import com.taobao.tddl.dbsync.binlog.event.WriteRowsLogEvent;
+
+/**
+ * 针对解析器提供一个多阶段协同的处理
+ *
+ *
+ * 1. 网络接收 (单线程)
+ * 2. 事件基本解析 (单线程，事件类型、DDL解析构造TableMeta、维护位点信息)
+ * 3. 事件深度解析 (多线程, DML事件数据的完整解析)
+ * 4. 投递到store (单线程)
+ *
+ *
+ * @author agapple 2018年7月3日 下午4:54:17
+ * @since 1.0.26
+ */
+public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implements MultiStageCoprocessor {
+
+ private LogEventConvert logEventConvert;
+ private EventTransactionBuffer transactionBuffer;
+ private ErosaConnection connection;
+
+ private int parserThreadCount;
+ private int ringBufferSize;
+ private RingBuffer disruptorMsgBuffer;
+ private ExecutorService parserExecutor;
+ private ExecutorService stageExecutor;
+ private String destination;
+ private CanalParseException exception;
+
+ public MysqlMultiStageCoprocessor(int ringBufferSize, int parserThreadCount, LogEventConvert logEventConvert,
+ EventTransactionBuffer transactionBuffer, String destination){
+ this.ringBufferSize = ringBufferSize;
+ this.parserThreadCount = parserThreadCount;
+ this.logEventConvert = logEventConvert;
+ this.transactionBuffer = transactionBuffer;
+ this.destination = destination;
+ }
+
+ @Override
+ public void start() {
+ super.start();
+ this.exception = null;
+ this.disruptorMsgBuffer = RingBuffer.createSingleProducer(new MessageEventFactory(),
+ ringBufferSize,
+ new BlockingWaitStrategy());
+
+ this.parserExecutor = Executors.newFixedThreadPool(parserThreadCount,
+ new NamedThreadFactory("MultiStageCoprocessor-Parser-" + destination));
+
+ this.stageExecutor = Executors.newFixedThreadPool(2, new NamedThreadFactory("MultiStageCoprocessor-other-"
+ + destination));
+ SequenceBarrier sequenceBarrier = disruptorMsgBuffer.newBarrier();
+
+ // stage 2
+ BatchEventProcessor simpleParserStage = new BatchEventProcessor(disruptorMsgBuffer,
+ sequenceBarrier,
+ new SimpleParserStage());
+ disruptorMsgBuffer.addGatingSequences(simpleParserStage.getSequence());
+
+ // stage 3
+ SequenceBarrier dmlParserSequenceBarrier = disruptorMsgBuffer.newBarrier(simpleParserStage.getSequence());
+ WorkHandler[] workHandlers = new DmlParserStage[parserThreadCount];
+ for (int i = 0; i < parserThreadCount; i++) {
+ workHandlers[i] = new DmlParserStage();
+ }
+ WorkerPool workerPool = new WorkerPool(disruptorMsgBuffer,
+ dmlParserSequenceBarrier,
+ new FatalExceptionHandler(),
+ workHandlers);
+ Sequence[] sequence = workerPool.getWorkerSequences();
+ disruptorMsgBuffer.addGatingSequences(sequence);
+
+ // stage 4
+ SequenceBarrier sinkSequenceBarrier = disruptorMsgBuffer.newBarrier(sequence);
+ BatchEventProcessor sinkStoreStage = new BatchEventProcessor(disruptorMsgBuffer,
+ sinkSequenceBarrier,
+ new SinkStoreStage());
+ disruptorMsgBuffer.addGatingSequences(sinkStoreStage.getSequence());
+
+ // start work
+ stageExecutor.submit(simpleParserStage);
+ stageExecutor.submit(sinkStoreStage);
+ workerPool.start(parserExecutor);
+ }
+
+ @Override
+ public void stop() {
+ try {
+ parserExecutor.shutdownNow();
+ } catch (Throwable e) {
+ // ignore
+ }
+
+ try {
+ stageExecutor.shutdownNow();
+ } catch (Throwable e) {
+ // ignore
+ }
+ disruptorMsgBuffer = null;
+ super.stop();
+ }
+
+ /**
+ * 网络数据投递
+ */
+ public void publish(LogBuffer buffer) {
+ if (!isStart() && exception != null) {
+ throw exception;
+ }
+ long next = disruptorMsgBuffer.next();
+ MessageEvent event = disruptorMsgBuffer.get(next);
+ event.setBuffer(buffer);
+ disruptorMsgBuffer.publish(next);
+ }
+
+ public void publish(LogBuffer buffer, String binlogFileName) {
+ if (!isStart() && exception != null) {
+ throw exception;
+ }
+ long next = disruptorMsgBuffer.next();
+ MessageEvent event = disruptorMsgBuffer.get(next);
+ event.setBuffer(buffer);
+ event.setBinlogFileName(binlogFileName);
+ disruptorMsgBuffer.publish(next);
+ }
+
+ @Override
+ public void reset() {
+ stop();
+ start();
+ }
+
+ private class SimpleParserStage implements EventHandler, LifecycleAware {
+
+ private LogDecoder decoder;
+ private LogContext context;
+
+ public SimpleParserStage(){
+ decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
+ context = new LogContext();
+ }
+
+ public void onEvent(MessageEvent event, long sequence, boolean endOfBatch) throws Exception {
+ try {
+ LogBuffer buffer = event.getBuffer();
+ if (StringUtils.isNotEmpty(event.getBinlogFileName())
+ && !context.getLogPosition().getFileName().equals(event.getBinlogFileName())) {
+ // set roate binlog file name
+ context.setLogPosition(new LogPosition(event.getBinlogFileName(), context.getLogPosition()
+ .getPosition()));
+ }
+
+ LogEvent logEvent = decoder.decode(buffer, context);
+ event.setEvent(logEvent);
+
+ int eventType = logEvent.getHeader().getType();
+ TableMeta tableMeta = null;
+ boolean needDmlParse = false;
+ switch (eventType) {
+ case LogEvent.WRITE_ROWS_EVENT_V1:
+ case LogEvent.WRITE_ROWS_EVENT:
+ tableMeta = logEventConvert.parseRowsEventForTableMeta((WriteRowsLogEvent) logEvent);
+ needDmlParse = true;
+ break;
+ case LogEvent.UPDATE_ROWS_EVENT_V1:
+ case LogEvent.UPDATE_ROWS_EVENT:
+ tableMeta = logEventConvert.parseRowsEventForTableMeta((UpdateRowsLogEvent) logEvent);
+ needDmlParse = true;
+ break;
+ case LogEvent.DELETE_ROWS_EVENT_V1:
+ case LogEvent.DELETE_ROWS_EVENT:
+ tableMeta = logEventConvert.parseRowsEventForTableMeta((DeleteRowsLogEvent) logEvent);
+ needDmlParse = true;
+ break;
+ case LogEvent.ROWS_QUERY_LOG_EVENT:
+ needDmlParse = true;
+ break;
+ default:
+ CanalEntry.Entry entry = logEventConvert.parse(event.getEvent(), false);
+ event.setEntry(entry);
+ }
+
+ // 记录一下DML的表结构
+ event.setNeedDmlParse(needDmlParse);
+ event.setTable(tableMeta);
+ } catch (Throwable e) {
+ exception = new CanalParseException(e);
+ throw exception;
+ }
+ }
+
+ @Override
+ public void onStart() {
+
+ }
+
+ @Override
+ public void onShutdown() {
+ stop();
+ }
+ }
+
+ private class DmlParserStage implements WorkHandler, LifecycleAware {
+
+ @Override
+ public void onEvent(MessageEvent event) throws Exception {
+ try {
+ if (event.isNeedDmlParse()) {
+ int eventType = event.getEvent().getHeader().getType();
+ CanalEntry.Entry entry = null;
+ switch (eventType) {
+ case LogEvent.ROWS_QUERY_LOG_EVENT:
+ entry = logEventConvert.parse(event.getEvent(), false);
+ break;
+ default:
+ // 单独解析dml事件
+ entry = logEventConvert.parseRowsEvent((RowsLogEvent) event.getEvent(), event.getTable());
+ }
+
+ event.setEntry(entry);
+ }
+ } catch (Throwable e) {
+ exception = new CanalParseException(e);
+ throw exception;
+ }
+ }
+
+ @Override
+ public void onStart() {
+
+ }
+
+ @Override
+ public void onShutdown() {
+ stop();
+ }
+ }
+
+ private class SinkStoreStage implements EventHandler, LifecycleAware {
+
+ public void onEvent(MessageEvent event, long sequence, boolean endOfBatch) throws Exception {
+ try {
+ if (event.getEntry() != null) {
+ transactionBuffer.add(event.getEntry());
+ }
+
+ LogEvent logEvent = event.getEvent();
+ if (connection instanceof MysqlConnection && logEvent.getSemival() == 1) {
+ // semi ack回报
+ ((MysqlConnection) connection).sendSemiAck(logEvent.getHeader().getLogFileName(),
+ logEvent.getHeader().getLogPos());
+ }
+
+ // clear for gc
+ event.setBuffer(null);
+ event.setBinlogFileName(null);
+ event.setEvent(null);
+ event.setTable(null);
+ event.setEntry(null);
+ event.setNeedDmlParse(false);
+ } catch (Throwable e) {
+ exception = new CanalParseException(e);
+ throw exception;
+ }
+ }
+
+ @Override
+ public void onStart() {
+
+ }
+
+ @Override
+ public void onShutdown() {
+ stop();
+ }
+ }
+
+ class MessageEvent {
+
+ private String binlogFileName; // for local binlog parse
+ private LogBuffer buffer;
+ private CanalEntry.Entry entry;
+ private boolean needDmlParse = false;
+ private TableMeta table;
+ private LogEvent event;
+
+ public String getBinlogFileName() {
+ return binlogFileName;
+ }
+
+ public void setBinlogFileName(String binlogFileName) {
+ this.binlogFileName = binlogFileName;
+ }
+
+ public LogBuffer getBuffer() {
+ return buffer;
+ }
+
+ public void setBuffer(LogBuffer buffer) {
+ this.buffer = buffer;
+ }
+
+ public LogEvent getEvent() {
+ return event;
+ }
+
+ public void setEvent(LogEvent event) {
+ this.event = event;
+ }
+
+ public CanalEntry.Entry getEntry() {
+ return entry;
+ }
+
+ public void setEntry(CanalEntry.Entry entry) {
+ this.entry = entry;
+ }
+
+ public boolean isNeedDmlParse() {
+ return needDmlParse;
+ }
+
+ public void setNeedDmlParse(boolean needDmlParse) {
+ this.needDmlParse = needDmlParse;
+ }
+
+ public TableMeta getTable() {
+ return table;
+ }
+
+ public void setTable(TableMeta table) {
+ this.table = table;
+ }
+
+ }
+
+ class MessageEventFactory implements EventFactory {
+
+ public MessageEvent newInstance() {
+ return new MessageEvent();
+ }
+ }
+
+ public void setLogEventConvert(LogEventConvert logEventConvert) {
+ this.logEventConvert = logEventConvert;
+ }
+
+ public void setTransactionBuffer(EventTransactionBuffer transactionBuffer) {
+ this.transactionBuffer = transactionBuffer;
+ }
+
+ public void setConnection(ErosaConnection connection) {
+ this.connection = connection;
+ }
+
+}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DirectLogFetcher.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DirectLogFetcher.java
index e3ac5ed3..a85962cc 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DirectLogFetcher.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DirectLogFetcher.java
@@ -171,10 +171,13 @@ public class DirectLogFetcher extends LogFetcher {
private final boolean fetch0(final int off, final int len) throws IOException {
ensureCapacity(off + len);
- byte[] read = channel.read(len, READ_TIMEOUT_MILLISECONDS);
- System.arraycopy(read, 0, this.buffer, off, len);
+ // byte[] read = channel.read(len, READ_TIMEOUT_MILLISECONDS);
+ // System.arraycopy(read, 0, this.buffer, off, len);
- if (limit < off + len) limit = off + len;
+ channel.read(buffer, off, len, READ_TIMEOUT_MILLISECONDS);
+ if (limit < off + len) {
+ limit = off + len;
+ }
return true;
}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java
index b299165c..f99d551c 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java
@@ -46,7 +46,6 @@ import com.taobao.tddl.dbsync.binlog.event.IntvarLogEvent;
import com.taobao.tddl.dbsync.binlog.event.LogHeader;
import com.taobao.tddl.dbsync.binlog.event.QueryLogEvent;
import com.taobao.tddl.dbsync.binlog.event.RandLogEvent;
-import com.taobao.tddl.dbsync.binlog.event.RotateLogEvent;
import com.taobao.tddl.dbsync.binlog.event.RowsLogBuffer;
import com.taobao.tddl.dbsync.binlog.event.RowsLogEvent;
import com.taobao.tddl.dbsync.binlog.event.RowsQueryLogEvent;
@@ -89,7 +88,6 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
private volatile AviaterRegexFilter nameBlackFilter;
private TableMetaCache tableMetaCache;
- private String binlogFileName = "mysql-bin.000001";
private Charset charset = Charset.defaultCharset();
private boolean filterQueryDcl = false;
private boolean filterQueryDml = false;
@@ -119,9 +117,6 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
int eventType = logEvent.getHeader().getType();
switch (eventType) {
- case LogEvent.ROTATE_EVENT:
- binlogFileName = ((RotateLogEvent) logEvent).getFilename();
- break;
case LogEvent.QUERY_EVENT:
return parseQueryEvent((QueryLogEvent) logEvent, isSeek);
case LogEvent.XID_EVENT:
@@ -158,7 +153,6 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
public void reset() {
// do nothing
- binlogFileName = "mysql-bin.000001";
if (tableMetaCache != null) {
tableMetaCache.clearTableMeta();
}
@@ -174,7 +168,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
gtidSet.update(value);
}
- Header header = createHeader("", logHeader, "", "", EventType.GTID);
+ Header header = createHeader(logHeader, "", "", EventType.GTID);
return createEntry(header, EntryType.GTIDLOG, builder.build().toByteString());
}
@@ -187,7 +181,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
beginBuilder.addProps(createSpecialPair(XA_TYPE, XA_START));
beginBuilder.addProps(createSpecialPair(XA_XID, getXaXid(queryString, XA_START)));
TransactionBegin transactionBegin = beginBuilder.build();
- Header header = createHeader(binlogFileName, event.getHeader(), "", "", null);
+ Header header = createHeader(event.getHeader(), "", "", null);
return createEntry(header, EntryType.TRANSACTIONBEGIN, transactionBegin.toByteString());
} else if (StringUtils.startsWithIgnoreCase(queryString, XA_END)) {
// xa start use TransactionEnd
@@ -196,11 +190,11 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
endBuilder.addProps(createSpecialPair(XA_TYPE, XA_END));
endBuilder.addProps(createSpecialPair(XA_XID, getXaXid(queryString, XA_END)));
TransactionEnd transactionEnd = endBuilder.build();
- Header header = createHeader(binlogFileName, event.getHeader(), "", "", null);
+ Header header = createHeader(event.getHeader(), "", "", null);
return createEntry(header, EntryType.TRANSACTIONEND, transactionEnd.toByteString());
} else if (StringUtils.startsWithIgnoreCase(queryString, XA_COMMIT)) {
// xa commit
- Header header = createHeader(binlogFileName, event.getHeader(), "", "", EventType.XACOMMIT);
+ Header header = createHeader(event.getHeader(), "", "", EventType.XACOMMIT);
RowChange.Builder rowChangeBuider = RowChange.newBuilder();
rowChangeBuider.setSql(queryString);
rowChangeBuider.addProps(createSpecialPair(XA_TYPE, XA_COMMIT));
@@ -209,7 +203,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
return createEntry(header, EntryType.ROWDATA, rowChangeBuider.build().toByteString());
} else if (StringUtils.startsWithIgnoreCase(queryString, XA_ROLLBACK)) {
// xa rollback
- Header header = createHeader(binlogFileName, event.getHeader(), "", "", EventType.XAROLLBACK);
+ Header header = createHeader(event.getHeader(), "", "", EventType.XAROLLBACK);
RowChange.Builder rowChangeBuider = RowChange.newBuilder();
rowChangeBuider.setSql(queryString);
rowChangeBuider.addProps(createSpecialPair(XA_TYPE, XA_ROLLBACK));
@@ -218,11 +212,11 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
return createEntry(header, EntryType.ROWDATA, rowChangeBuider.build().toByteString());
} else if (StringUtils.endsWithIgnoreCase(queryString, BEGIN)) {
TransactionBegin transactionBegin = createTransactionBegin(event.getSessionId());
- Header header = createHeader(binlogFileName, event.getHeader(), "", "", null);
+ Header header = createHeader(event.getHeader(), "", "", null);
return createEntry(header, EntryType.TRANSACTIONBEGIN, transactionBegin.toByteString());
} else if (StringUtils.endsWithIgnoreCase(queryString, COMMIT)) {
TransactionEnd transactionEnd = createTransactionEnd(0L); // MyISAM可能不会有xid事件
- Header header = createHeader(binlogFileName, event.getHeader(), "", "", null);
+ Header header = createHeader(event.getHeader(), "", "", null);
return createEntry(header, EntryType.TRANSACTIONEND, transactionEnd.toByteString());
} else {
boolean notFilter = false;
@@ -265,7 +259,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
tableMetaCache.apply(position, event.getDbName(), queryString, null);
}
- Header header = createHeader(binlogFileName, event.getHeader(), schemaName, tableName, type);
+ Header header = createHeader(event.getHeader(), schemaName, tableName, type);
RowChange.Builder rowChangeBuider = RowChange.newBuilder();
if (type != EventType.QUERY && type != EventType.INSERT && type != EventType.UPDATE
&& type != EventType.DELETE) {
@@ -425,52 +419,78 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
private Entry parseXidEvent(XidLogEvent event) {
TransactionEnd transactionEnd = createTransactionEnd(event.getXid());
- Header header = createHeader(binlogFileName, event.getHeader(), "", "", null);
+ Header header = createHeader(event.getHeader(), "", "", null);
return createEntry(header, EntryType.TRANSACTIONEND, transactionEnd.toByteString());
}
- private Entry parseRowsEvent(RowsLogEvent event) {
+ public TableMeta parseRowsEventForTableMeta(RowsLogEvent event) {
+ TableMapLogEvent table = event.getTable();
+ if (table == null) {
+ // tableId对应的记录不存在
+ throw new TableIdNotFoundException("not found tableId:" + event.getTableId());
+ }
+
+ boolean isHeartBeat = isAliSQLHeartBeat(table.getDbName(), table.getTableName());
+ boolean isRDSHeartBeat = tableMetaCache.isOnRDS() && isRDSHeartBeat(table.getDbName(), table.getTableName());
+
+ String fullname = table.getDbName() + "." + table.getTableName();
+ // check name filter
+ if (nameFilter != null && !nameFilter.filter(fullname)) {
+ return null;
+ }
+ if (nameBlackFilter != null && nameBlackFilter.filter(fullname)) {
+ return null;
+ }
+
+ // if (isHeartBeat || isRDSHeartBeat) {
+ // // 忽略rds模式的mysql.ha_health_check心跳数据
+ // return null;
+ // }
+ TableMeta tableMeta = null;
+ if (isRDSHeartBeat) {
+ // 处理rds模式的mysql.ha_health_check心跳数据
+ // 主要RDS的心跳表基本无权限,需要mock一个tableMeta
+ FieldMeta idMeta = new FieldMeta("id", "bigint(20)", true, false, "0");
+ FieldMeta typeMeta = new FieldMeta("type", "char(1)", false, true, "0");
+ tableMeta = new TableMeta(table.getDbName(), table.getTableName(), Arrays.asList(idMeta, typeMeta));
+ } else if (isHeartBeat) {
+ // 处理alisql模式的test.heartbeat心跳数据
+ // 心跳表基本无权限,需要mock一个tableMeta
+ FieldMeta idMeta = new FieldMeta("id", "smallint(6)", false, true, null);
+ FieldMeta typeMeta = new FieldMeta("ts", "int(11)", true, false, null);
+ tableMeta = new TableMeta(table.getDbName(), table.getTableName(), Arrays.asList(idMeta, typeMeta));
+ }
+
+ EntryPosition position = createPosition(event.getHeader());
+ if (tableMetaCache != null && tableMeta == null) {// 入错存在table meta
+ tableMeta = getTableMeta(table.getDbName(), table.getTableName(), true, position);
+ if (tableMeta == null) {
+ if (!filterTableError) {
+ throw new CanalParseException("not found [" + fullname + "] in db , pls check!");
+ }
+ }
+ }
+
+ return tableMeta;
+ }
+
+ public Entry parseRowsEvent(RowsLogEvent event) {
+ return parseRowsEvent(event, null);
+ }
+
+ public Entry parseRowsEvent(RowsLogEvent event, TableMeta tableMeta) {
if (filterRows) {
return null;
}
try {
- TableMapLogEvent table = event.getTable();
- if (table == null) {
- // tableId对应的记录不存在
- throw new TableIdNotFoundException("not found tableId:" + event.getTableId());
+ if (tableMeta == null) { // 如果没有外部指定
+ tableMeta = parseRowsEventForTableMeta(event);
}
- boolean isHeartBeat = isAliSQLHeartBeat(table.getDbName(), table.getTableName());
- boolean isRDSHeartBeat = tableMetaCache.isOnRDS()
- && isRDSHeartBeat(table.getDbName(), table.getTableName());
-
- String fullname = table.getDbName() + "." + table.getTableName();
- // check name filter
- if (nameFilter != null && !nameFilter.filter(fullname)) {
+ if (tableMeta == null) {
+ // 拿不到表结构,执行忽略
return null;
}
- if (nameBlackFilter != null && nameBlackFilter.filter(fullname)) {
- return null;
- }
-
- // if (isHeartBeat || isRDSHeartBeat) {
- // // 忽略rds模式的mysql.ha_health_check心跳数据
- // return null;
- // }
- TableMeta tableMeta = null;
- if (isRDSHeartBeat) {
- // 处理rds模式的mysql.ha_health_check心跳数据
- // 主要RDS的心跳表基本无权限,需要mock一个tableMeta
- FieldMeta idMeta = new FieldMeta("id", "bigint(20)", true, false, "0");
- FieldMeta typeMeta = new FieldMeta("type", "char(1)", false, true, "0");
- tableMeta = new TableMeta(table.getDbName(), table.getTableName(), Arrays.asList(idMeta, typeMeta));
- } else if (isHeartBeat) {
- // 处理alisql模式的test.heartbeat心跳数据
- // 心跳表基本无权限,需要mock一个tableMeta
- FieldMeta idMeta = new FieldMeta("id", "smallint(6)", false, true, null);
- FieldMeta typeMeta = new FieldMeta("ts", "int(11)", true, false, null);
- tableMeta = new TableMeta(table.getDbName(), table.getTableName(), Arrays.asList(idMeta, typeMeta));
- }
EventType eventType = null;
int type = event.getHeader().getType();
@@ -484,12 +504,9 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
throw new CanalParseException("unsupport event type :" + event.getHeader().getType());
}
- Header header = createHeader(binlogFileName,
- event.getHeader(),
- table.getDbName(),
- table.getTableName(),
- eventType);
- EntryPosition position = createPosition(event.getHeader());
+ TableMapLogEvent table = event.getTable();
+ Header header = createHeader(event.getHeader(), table.getDbName(), table.getTableName(), eventType);
+
RowChange.Builder rowChangeBuider = RowChange.newBuilder();
rowChangeBuider.setTableId(event.getTableId());
rowChangeBuider.setIsDdl(false);
@@ -498,18 +515,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
RowsLogBuffer buffer = event.getRowsBuf(charset.name());
BitSet columns = event.getColumns();
BitSet changeColumns = event.getChangeColumns();
- boolean tableError = false;
- if (tableMetaCache != null && tableMeta == null) {// 入错存在table meta
- // cache
- tableMeta = getTableMeta(table.getDbName(), table.getTableName(), true, position);
- if (tableMeta == null) {
- tableError = true;
- if (!filterTableError) {
- throw new CanalParseException("not found [" + fullname + "] in db , pls check!");
- }
- }
- }
+ boolean tableError = false;
while (buffer.nextOneRow(columns)) {
// 处理row记录
RowData.Builder rowDataBuilder = RowData.newBuilder();
@@ -539,7 +546,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
logger.warn("table parser error : {}storeValue: {}", entry.toString(), rowChange.toString());
return null;
} else {
- Entry entry = createEntry(header, EntryType.ROWDATA, rowChangeBuider.build().toByteString());
+ Entry entry = createEntry(header, EntryType.ROWDATA, rowChange.toByteString());
return entry;
}
} catch (Exception e) {
@@ -548,7 +555,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
}
private EntryPosition createPosition(LogHeader logHeader) {
- return new EntryPosition(binlogFileName,
+ return new EntryPosition(logHeader.getLogFileName(),
logHeader.getLogPos(),
logHeader.getWhen() * 1000L,
logHeader.getServerId()); // 记录到秒
@@ -641,7 +648,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
isBinary = true;
} else if (StringUtils.containsIgnoreCase(fieldMeta.getColumnType(), "BINARY")) {
isBinary = true;
- } else if (StringUtils.containsIgnoreCase(fieldMeta.getColumnType(), "tinyint(1)")) {
+ } else if (StringUtils.containsIgnoreCase(fieldMeta.getColumnType(), "TINYINT(1)")) {
isSingleBit = true;
}
}
@@ -771,7 +778,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
}
private Entry buildQueryEntry(String queryString, LogHeader logHeader, String tableName) {
- Header header = createHeader(binlogFileName, logHeader, "", tableName, EventType.QUERY);
+ Header header = createHeader(logHeader, "", tableName, EventType.QUERY);
RowChange.Builder rowChangeBuider = RowChange.newBuilder();
rowChangeBuider.setSql(queryString);
rowChangeBuider.setEventType(EventType.QUERY);
@@ -779,19 +786,18 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
}
private Entry buildQueryEntry(String queryString, LogHeader logHeader) {
- Header header = createHeader(binlogFileName, logHeader, "", "", EventType.QUERY);
+ Header header = createHeader(logHeader, "", "", EventType.QUERY);
RowChange.Builder rowChangeBuider = RowChange.newBuilder();
rowChangeBuider.setSql(queryString);
rowChangeBuider.setEventType(EventType.QUERY);
return createEntry(header, EntryType.ROWDATA, rowChangeBuider.build().toByteString());
}
- private Header createHeader(String binlogFile, LogHeader logHeader, String schemaName, String tableName,
- EventType eventType) {
+ private Header createHeader(LogHeader logHeader, String schemaName, String tableName, EventType eventType) {
// header会做信息冗余,方便以后做检索或者过滤
Header.Builder headerBuilder = Header.newBuilder();
headerBuilder.setVersion(version);
- headerBuilder.setLogfileName(binlogFile);
+ headerBuilder.setLogfileName(logHeader.getLogFileName());
headerBuilder.setLogfileOffset(logHeader.getLogPos() - logHeader.getEventLen());
headerBuilder.setServerId(logHeader.getServerId());
headerBuilder.setServerenCode(UTF_8);// 经过java输出后所有的编码为unicode
diff --git a/pom.xml b/pom.xml
index 10b31084..8ab961ea 100644
--- a/pom.xml
+++ b/pom.xml
@@ -263,6 +263,11 @@
druid
1.1.9