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

Merge remote-tracking branch 'upstream/master'

Browse Source
This commit is contained in:
winger
2018-11-05 18:39:47 +08:00
parent d8e411462e 59e65ed4a6
commit 1e20cac0da
96 changed files with 6093 additions and 514 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -353,7 +353,11 @@ public abstract class AbstractEventParser<EVENT> extends AbstractCanalLifeCycle
eventSink.interrupt();
if (multiStageCoprocessor != null && multiStageCoprocessor.isStart()) {
multiStageCoprocessor.stop();
try {
multiStageCoprocessor.stop();
} catch (Throwable t) {
logger.debug("multi processor rejected:", t);
}
}
try {

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

@@ -60,6 +60,7 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
convert.setFilterQueryDdl(filterQueryDdl);
convert.setFilterRows(filterRows);
convert.setFilterTableError(filterTableError);
convert.setUseDruidDdlFilter(useDruidDdlFilter);
return convert;
}

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

@@ -37,21 +37,23 @@ 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.event.FormatDescriptionLogEvent;
public class MysqlConnection implements ErosaConnection {
private static final Logger logger = LoggerFactory.getLogger(MysqlConnection.class);
private static final Logger logger = LoggerFactory.getLogger(MysqlConnection.class);
private MysqlConnector connector;
private long slaveId;
private Charset charset = Charset.forName("UTF-8");
private Charset charset = Charset.forName("UTF-8");
private BinlogFormat binlogFormat;
private BinlogImage binlogImage;
// tsdb releated
private AuthenticationInfo authInfo;
protected int connTimeout = 5 * 1000; // 5秒
protected int soTimeout = 60 * 60 * 1000; // 1小时
protected int connTimeout = 5 * 1000; // 5秒
protected int soTimeout = 60 * 60 * 1000; // 1小时
private int binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_OFF;
// dump binlog bytes, 暂不包括meta与TSDB
private AtomicLong receivedBinlogBytes;
@@ -118,7 +120,7 @@ public class MysqlConnection implements ErosaConnection {
*/
public void seek(String binlogfilename, Long binlogPosition, SinkFunction func) throws IOException {
updateSettings();
loadBinlogChecksum();
sendBinlogDump(binlogfilename, binlogPosition);
DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize());
fetcher.start(connector.getChannel());
@@ -128,6 +130,7 @@ public class MysqlConnection implements ErosaConnection {
decoder.handle(LogEvent.QUERY_EVENT);
decoder.handle(LogEvent.XID_EVENT);
LogContext context = new LogContext();
context.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum));
while (fetcher.fetch()) {
accumulateReceivedBytes(fetcher.limit());
LogEvent event = null;
@@ -145,12 +148,14 @@ public class MysqlConnection implements ErosaConnection {
public void dump(String binlogfilename, Long binlogPosition, SinkFunction func) throws IOException {
updateSettings();
loadBinlogChecksum();
sendRegisterSlave();
sendBinlogDump(binlogfilename, binlogPosition);
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();
context.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum));
while (fetcher.fetch()) {
accumulateReceivedBytes(fetcher.limit());
LogEvent event = null;
@@ -173,6 +178,7 @@ public class MysqlConnection implements ErosaConnection {
@Override
public void dump(GTIDSet gtidSet, SinkFunction func) throws IOException {
updateSettings();
loadBinlogChecksum();
sendBinlogDumpGTID(gtidSet);
DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize());
@@ -180,6 +186,7 @@ public class MysqlConnection implements ErosaConnection {
fetcher.start(connector.getChannel());
LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
LogContext context = new LogContext();
context.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum));
// fix bug: #890 将gtid传输至context中，供decode使用
context.setGtidSet(gtidSet);
while (fetcher.fetch()) {
@@ -207,9 +214,11 @@ public class MysqlConnection implements ErosaConnection {
@Override
public void dump(String binlogfilename, Long binlogPosition, MultiStageCoprocessor coprocessor) throws IOException {
updateSettings();
loadBinlogChecksum();
sendRegisterSlave();
sendBinlogDump(binlogfilename, binlogPosition);
((MysqlMultiStageCoprocessor) coprocessor).setConnection(this);
((MysqlMultiStageCoprocessor) coprocessor).setBinlogChecksum(binlogChecksum);
DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize());
try {
fetcher.start(connector.getChannel());
@@ -234,9 +243,10 @@ public class MysqlConnection implements ErosaConnection {
@Override
public void dump(GTIDSet gtidSet, MultiStageCoprocessor coprocessor) throws IOException {
updateSettings();
loadBinlogChecksum();
sendBinlogDumpGTID(gtidSet);
((MysqlMultiStageCoprocessor) coprocessor).setConnection(this);
((MysqlMultiStageCoprocessor) coprocessor).setBinlogChecksum(binlogChecksum);
DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize());
try {
fetcher.start(connector.getChannel());
@@ -482,6 +492,52 @@ public class MysqlConnection implements ErosaConnection {
}
}
/**
* 获取主库checksum信息
*
* <pre>
* mariadb区别于mysql会在binlog的第一个事件Rotate_Event里也会采用checksum逻辑,而mysql是在第二个binlog事件之后才感知是否需要处理checksum
* 导致maraidb只要是开启checksum就会出现binlog文件名解析乱码
* fixed issue : https://github.com/alibaba/canal/issues/1081
* </pre>
*/
private void loadBinlogChecksum() {
if (checkMariaDB()) {
ResultSetPacket rs = null;
try {
rs = query("select @@global.binlog_checksum");
} catch (IOException e) {
throw new CanalParseException(e);
}
List<String> columnValues = rs.getFieldValues();
if (columnValues != null && columnValues.size() >= 1 && columnValues.get(0).toUpperCase().equals("CRC32")) {
binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_CRC32;
} else {
binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_OFF;
}
}
}
/**
* 获取是否为mariadb
*/
private boolean checkMariaDB() {
ResultSetPacket rs = null;
try {
rs = query("SELECT @@version");
} catch (IOException e) {
throw new CanalParseException(e);
}
List<String> columnValues = rs.getFieldValues();
if (columnValues != null && columnValues.size() >= 1) {
return StringUtils.containsIgnoreCase(columnValues.get(0), "MariaDB");
}
return false;
}
private void accumulateReceivedBytes(long x) {
if (receivedBinlogBytes != null) {
receivedBinlogBytes.addAndGet(x);

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

@@ -69,6 +69,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
// update by yishun.chen,特殊异常处理参数
private int dumpErrorCount = 0; // binlogDump失败异常计数
private int dumpErrorCountThreshold = 2; // binlogDump失败异常计数阀值
private boolean rdsOssMode = false;
protected ErosaConnection buildErosaConnection() {
return buildMysqlConnection(this.runningInfo);
@@ -352,7 +353,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
return logPosition.getPostion();
}
if (masterPosition!=null && StringUtils.isNotEmpty(masterPosition.getGtid())) {
if (masterPosition != null && StringUtils.isNotEmpty(masterPosition.getGtid())) {
return masterPosition;
}
}
@@ -493,6 +494,12 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
dumpErrorCount = 0;
return findPosition;
}
Long timestamp = logPosition.getPostion().getTimestamp();
if (isRdsOssMode() && (timestamp != null && timestamp > 0)) {
// 如果binlog位点不存在，并且属于timestamp不为空,可以返回null走到oss binlog处理
return null;
}
}
// 其余情况
logger.warn("prepare to find start position just last position\n {}",
@@ -745,14 +752,13 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
logPosition.setPostion(entryPosition);
}
if (entry == null) {
return true;
}
String logfilename = entry.getHeader().getLogfileName();
Long logfileoffset = entry.getHeader().getLogfileOffset();
Long logposTimestamp = entry.getHeader().getExecuteTime();
Long serverId = entry.getHeader().getServerId();
// 直接用event的位点来处理,解决一个binlog文件里没有任何事件导致死循环无法退出的问题
String logfilename = event.getHeader().getLogFileName();
// 记录的是binlog end offest,
// 因为与其对比的offest是show master status里的end offest
Long logfileoffset = event.getHeader().getLogPos();
Long logposTimestamp = event.getHeader().getWhen() * 1000;
Long serverId = event.getHeader().getServerId();
// 如果最小的一条记录都不满足条件，可直接退出
if (logposTimestamp >= startTimestamp) {
@@ -764,6 +770,10 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
return false;
}
if (entry == null) {
return true;
}
// 记录一下上一个事务结束的位置，即下一个事务的position
// position = current +
// data.length，代表该事务的下一条offest，避免多余的事务重复
@@ -905,4 +915,12 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
this.dumpErrorCountThreshold = dumpErrorCountThreshold;
}
public boolean isRdsOssMode() {
return rdsOssMode;
}
public void setRdsOssMode(boolean rdsOssMode) {
this.rdsOssMode = rdsOssMode;
}
}

17
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java
View File

@@ -33,6 +33,7 @@ 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.event.DeleteRowsLogEvent;
import com.taobao.tddl.dbsync.binlog.event.FormatDescriptionLogEvent;
import com.taobao.tddl.dbsync.binlog.event.RowsLogEvent;
import com.taobao.tddl.dbsync.binlog.event.UpdateRowsLogEvent;
import com.taobao.tddl.dbsync.binlog.event.WriteRowsLogEvent;
@@ -69,6 +70,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
private WorkerPool<MessageEvent> workerPool;
private BatchEventProcessor<MessageEvent> simpleParserStage;
private BatchEventProcessor<MessageEvent> sinkStoreStage;
private LogContext logContext;
public MysqlMultiStageCoprocessor(int ringBufferSize, int parserThreadCount, LogEventConvert logEventConvert,
EventTransactionBuffer transactionBuffer, String destination){
@@ -95,9 +97,10 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
SequenceBarrier sequenceBarrier = disruptorMsgBuffer.newBarrier();
ExceptionHandler exceptionHandler = new SimpleFatalExceptionHandler();
// stage 2
this.logContext = new LogContext();
simpleParserStage = new BatchEventProcessor<MessageEvent>(disruptorMsgBuffer,
sequenceBarrier,
new SimpleParserStage());
new SimpleParserStage(logContext));
simpleParserStage.setExceptionHandler(exceptionHandler);
disruptorMsgBuffer.addGatingSequences(simpleParserStage.getSequence());
@@ -128,6 +131,12 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
workerPool.start(parserExecutor);
}
public void setBinlogChecksum(int binlogChecksum) {
if (binlogChecksum != LogEvent.BINLOG_CHECKSUM_ALG_OFF) {
logContext.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum));
}
}
@Override
public void stop() {
// fix bug #968，对于pool与
@@ -239,9 +248,9 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
private LogDecoder decoder;
private LogContext context;
public SimpleParserStage(){
public SimpleParserStage(LogContext context){
decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
context = new LogContext();
this.context = context;
if (gtidSet != null) {
context.setGtidSet(gtidSet);
}
@@ -266,6 +275,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
needDmlParse = true;
break;
case LogEvent.UPDATE_ROWS_EVENT_V1:
case LogEvent.PARTIAL_UPDATE_ROWS_EVENT:
case LogEvent.UPDATE_ROWS_EVENT:
tableMeta = logEventConvert.parseRowsEventForTableMeta((UpdateRowsLogEvent) logEvent);
needDmlParse = true;
@@ -468,4 +478,5 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
public void setGtidSet(GTIDSet gtidSet) {
this.gtidSet = gtidSet;
}
}

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

@@ -120,6 +120,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
case LogEvent.WRITE_ROWS_EVENT:
return parseRowsEvent((WriteRowsLogEvent) logEvent);
case LogEvent.UPDATE_ROWS_EVENT_V1:
case LogEvent.PARTIAL_UPDATE_ROWS_EVENT:
case LogEvent.UPDATE_ROWS_EVENT:
return parseRowsEvent((UpdateRowsLogEvent) logEvent);
case LogEvent.DELETE_ROWS_EVENT_V1:
@@ -260,7 +261,9 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
return null;
}
if (!isSeek) {
boolean isDml = (type == EventType.INSERT || type == EventType.UPDATE || type == EventType.DELETE);
if (!isSeek && !isDml) {
// 使用新的表结构元数据管理方式
EntryPosition position = createPosition(event.getHeader());
tableMetaCache.apply(position, event.getDbName(), queryString, null);
@@ -268,8 +271,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
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) {
if (type != EventType.QUERY && !isDml) {
rowChangeBuider.setIsDdl(true);
}
rowChangeBuider.setSql(queryString);
@@ -503,7 +505,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
int type = event.getHeader().getType();
if (LogEvent.WRITE_ROWS_EVENT_V1 == type || LogEvent.WRITE_ROWS_EVENT == type) {
eventType = EventType.INSERT;
} else if (LogEvent.UPDATE_ROWS_EVENT_V1 == type || LogEvent.UPDATE_ROWS_EVENT == type) {
} else if (LogEvent.UPDATE_ROWS_EVENT_V1 == type || LogEvent.UPDATE_ROWS_EVENT == type
|| LogEvent.PARTIAL_UPDATE_ROWS_EVENT == type) {
eventType = EventType.UPDATE;
} else if (LogEvent.DELETE_ROWS_EVENT_V1 == type || LogEvent.DELETE_ROWS_EVENT == type) {
eventType = EventType.DELETE;
@@ -522,7 +525,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
boolean tableError = false;
int rowsCount = 0;
while (buffer.nextOneRow(columns)) {
while (buffer.nextOneRow(columns, false)) {
// 处理row记录
RowData.Builder rowDataBuilder = RowData.newBuilder();
if (EventType.INSERT == eventType) {
@@ -534,7 +537,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
} else {
// update需要处理before/after
tableError |= parseOneRow(rowDataBuilder, event, buffer, columns, false, tableMeta);
if (!buffer.nextOneRow(changeColumns)) {
if (!buffer.nextOneRow(changeColumns, true)) {
rowChangeBuider.addRowDatas(rowDataBuilder.build());
break;
}
@@ -567,8 +570,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
}
private EntryPosition createPosition(LogHeader logHeader) {
return new EntryPosition(logHeader.getLogFileName(),
logHeader.getLogPos(),
return new EntryPosition(logHeader.getLogFileName(), logHeader.getLogPos() - logHeader.getEventLen(), // startPos
logHeader.getWhen() * 1000L,
logHeader.getServerId()); // 记录到秒
}
@@ -577,7 +579,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
boolean isAfter, TableMeta tableMeta) throws UnsupportedEncodingException {
int columnCnt = event.getTable().getColumnCnt();
ColumnInfo[] columnInfo = event.getTable().getColumnInfo();
// mysql8.0针对set @@global.binlog_row_metadata='FULL' 可以记录部分的metadata信息
boolean existOptionalMetaData = event.getTable().isExistOptionalMetaData();
boolean tableError = false;
// check table fileds count，只能处理加字段
boolean existRDSNoPrimaryKey = false;
@@ -631,22 +634,43 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
continue;
}
if (existRDSNoPrimaryKey && i == columnCnt - 1 && info.type == LogEvent.MYSQL_TYPE_LONGLONG) {
// 不解析最后一列
buffer.nextValue(info.type, info.meta, false);
continue;
}
Column.Builder columnBuilder = Column.newBuilder();
FieldMeta fieldMeta = null;
if (tableMeta != null && !tableError) {
// 处理file meta
fieldMeta = tableMeta.getFields().get(i);
}
if (existRDSNoPrimaryKey && i == columnCnt - 1 && info.type == LogEvent.MYSQL_TYPE_LONGLONG) {
// 不解析最后一列
buffer.nextValue(fieldMeta.getColumnName(), i, info.type, info.meta, false);
continue;
}
if (fieldMeta != null && existOptionalMetaData) {
// check column info
boolean check = StringUtils.equalsIgnoreCase(fieldMeta.getColumnName(), info.name);
check &= (fieldMeta.isUnsigned() == info.unsigned);
check &= (fieldMeta.isNullable() == info.nullable);
if (!check) {
throw new CanalParseException("MySQL8.0 unmatch column metadata & pls submit issue , db : "
+ fieldMeta.toString() + " , binlog : " + info.toString()
+ " , on : " + event.getHeader().getLogFileName() + ":"
+ event.getHeader().getLogPos());
}
}
Column.Builder columnBuilder = Column.newBuilder();
if (fieldMeta != null) {
columnBuilder.setName(fieldMeta.getColumnName());
columnBuilder.setIsKey(fieldMeta.isKey());
// 增加mysql type类型,issue 73
columnBuilder.setMysqlType(fieldMeta.getColumnType());
} else if (existOptionalMetaData) {
columnBuilder.setName(info.name);
columnBuilder.setIsKey(info.pk);
// mysql8.0里没有mysql type类型
// columnBuilder.setMysqlType(fieldMeta.getColumnType());
}
columnBuilder.setIndex(i);
columnBuilder.setIsNull(false);
@@ -664,12 +688,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
isSingleBit = true;
}
}
buffer.nextValue(info.type, info.meta, isBinary);
if (existRDSNoPrimaryKey && i == columnCnt - 1 && info.type == LogEvent.MYSQL_TYPE_LONGLONG) {
// 不解析最后一列
continue;
}
buffer.nextValue(fieldMeta.getColumnName(), i, info.type, info.meta, isBinary);
int javaType = buffer.getJavaType();
if (buffer.isNull()) {
columnBuilder.setIsNull(true);
@@ -683,7 +703,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
case Types.BIGINT:
// 处理unsigned类型
Number number = (Number) value;
if (fieldMeta != null && fieldMeta.isUnsigned() && number.longValue() < 0) {
boolean isUnsigned = (fieldMeta != null ? fieldMeta.isUnsigned() : (existOptionalMetaData ? info.unsigned : false));
if (isUnsigned && number.longValue() < 0) {
switch (buffer.getLength()) {
case 1: /* MYSQL_TYPE_TINY */
columnBuilder.setValue(String.valueOf(Integer.valueOf(TINYINT_MAX_VALUE
@@ -772,7 +793,6 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
default:
columnBuilder.setValue(value.toString());
}
}
columnBuilder.setSqlType(javaType);
@@ -818,6 +838,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
Header.Builder headerBuilder = Header.newBuilder();
headerBuilder.setVersion(version);
headerBuilder.setLogfileName(logHeader.getLogFileName());
// 记录的是该binlog的start offest
headerBuilder.setLogfileOffset(logHeader.getLogPos() - logHeader.getEventLen());
headerBuilder.setServerId(logHeader.getServerId());
headerBuilder.setServerenCode(UTF_8);// 经过java输出后所有的编码为unicode
@@ -981,4 +1002,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
public void setFilterRows(boolean filterRows) {
this.filterRows = filterRows;
}
public void setUseDruidDdlFilter(boolean useDruidDdlFilter) {
this.useDruidDdlFilter = useDruidDdlFilter;
}
}

14
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java
View File

@@ -41,6 +41,7 @@ import org.apache.http.ssl.TrustStrategy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.parse.inbound.mysql.rds.data.BinlogFile;
/**
@@ -103,6 +104,9 @@ public class BinlogDownloadQueue {
public BinlogFile tryOne() throws Throwable {
BinlogFile binlogFile = binlogList.poll();
if (binlogFile == null) {
throw new CanalParseException("download binlog is null");
}
download(binlogFile);
hostId = binlogFile.getHostInstanceID();
this.currentSize++;
@@ -131,7 +135,7 @@ public class BinlogDownloadQueue {
if (StringUtils.isNotEmpty(needCompareName) && StringUtils.endsWith(needCompareName, "tar")) {
needCompareName = needCompareName.substring(0, needCompareName.indexOf("."));
}
return fileName.equalsIgnoreCase(needCompareName) && binlogList.isEmpty();
return (needCompareName == null || fileName.equalsIgnoreCase(needCompareName)) && binlogList.isEmpty();
}
public void prepare() throws InterruptedException {
@@ -162,6 +166,14 @@ public class BinlogDownloadQueue {
this.currentSize = 0;
binlogList.clear();
downloadQueue.clear();
try {
downloadThread.interrupt();
downloadThread.join();// 等待其结束
} catch (InterruptedException e) {
// ignore
} finally {
downloadThread = null;
}
}
private void download(BinlogFile binlogFile) throws Throwable {

27
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogEventParserProxy.java
View File

@@ -22,14 +22,14 @@ import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser;
*/
public class RdsBinlogEventParserProxy extends MysqlEventParser {
private String rdsOpenApiUrl = "https://rds.aliyuncs.com/"; // openapi地址
private String accesskey; // 云账号的ak
private String secretkey; // 云账号sk
private String instanceId; // rds实例id
private String directory; // binlog目录
private int batchFileSize = 4; // 最多下载的binlog文件数量
private String rdsOpenApiUrl = "https://rds.aliyuncs.com/"; // openapi地址
private String accesskey; // 云账号的ak
private String secretkey; // 云账号sk
private String instanceId; // rds实例id
private String directory; // binlog目录
private int batchFileSize = 4; // 最多下载的binlog文件数量
private RdsLocalBinlogEventParser rdsLocalBinlogEventParser = new RdsLocalBinlogEventParser();
private RdsLocalBinlogEventParser rdsLocalBinlogEventParser = null;
private ExecutorService executorService = Executors.newSingleThreadExecutor(new ThreadFactory() {
@Override
@@ -43,8 +43,11 @@ public class RdsBinlogEventParserProxy extends MysqlEventParser {
@Override
public void start() {
if (StringUtils.isNotEmpty(accesskey) && StringUtils.isNotEmpty(secretkey)
if (rdsLocalBinlogEventParser == null && StringUtils.isNotEmpty(accesskey) && StringUtils.isNotEmpty(secretkey)
&& StringUtils.isNotEmpty(instanceId)) {
rdsLocalBinlogEventParser = new RdsLocalBinlogEventParser();
// rds oss mode
setRdsOssMode(true);
final ParserExceptionHandler targetHandler = this.getParserExceptionHandler();
if (directory == null) {
directory = System.getProperty("java.io.tmpdir", "/tmp") + "/" + destination;
@@ -119,10 +122,18 @@ public class RdsBinlogEventParserProxy extends MysqlEventParser {
long serverId = rdsBinlogEventParserProxy.getServerId();
rdsLocalBinlogEventParser.setServerId(serverId);
rdsBinlogEventParserProxy.stop();
} catch (Throwable e) {
logger.info("handle exception failed", e);
}
try {
logger.info("start rds mysql binlog parser!");
rdsLocalBinlogEventParser.start();
} catch (Throwable e) {
logger.info("handle exception failed", e);
rdsLocalBinlogEventParser.stop();
RdsBinlogEventParserProxy rdsBinlogEventParserProxy = RdsBinlogEventParserProxy.this;
rdsBinlogEventParserProxy.start();// 继续重试
}
}
});

10
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsLocalBinlogEventParser.java
View File

@@ -57,7 +57,11 @@ public class RdsLocalBinlogEventParser extends LocalBinlogEventParser implements
if (entryPosition == null) {
throw new PositionNotFoundException("position not found!");
}
long startTimeInMill = entryPosition.getTimestamp();
Long startTimeInMill = entryPosition.getTimestamp();
if (startTimeInMill == null || startTimeInMill <= 0) {
throw new PositionNotFoundException("position timestamp is empty!");
}
startTime = startTimeInMill;
List<BinlogFile> binlogFiles = RdsBinlogOpenApi.listBinlogFiles(url,
accesskey,
@@ -65,6 +69,10 @@ public class RdsLocalBinlogEventParser extends LocalBinlogEventParser implements
instanceId,
new Date(startTime),
new Date(endTime));
if (binlogFiles.isEmpty()) {
throw new CanalParseException("start timestamp : " + startTimeInMill + " binlog files is empty");
}
binlogDownloadQueue = new BinlogDownloadQueue(binlogFiles, batchFileSize, directory);
binlogDownloadQueue.silenceDownload();
needWait = true;

44
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java
View File

@@ -7,6 +7,7 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;
@@ -48,18 +49,26 @@ public class DatabaseTableMeta implements TableMetaTSDB {
private static Logger logger = LoggerFactory.getLogger(DatabaseTableMeta.class);
private static Pattern pattern = Pattern.compile("Duplicate entry '.*' for key '*'");
private static Pattern h2Pattern = Pattern.compile("Unique index or primary key violation");
private static ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r, "[scheduler-table-meta-snapshot]");
thread.setDaemon(true);
return thread;
}
});
private String destination;
private MemoryTableMeta memoryTableMeta;
private MysqlConnection connection; // 查询meta信息的链接
private CanalEventFilter filter;
private CanalEventFilter blackFilter;
private EntryPosition lastPosition;
private ScheduledExecutorService scheduler;
private MetaHistoryDAO metaHistoryDAO;
private MetaSnapshotDAO metaSnapshotDAO;
private int snapshotInterval = 24;
private int snapshotExpire = 360;
private ScheduledFuture<?> scheduleSnapshotFuture;
public DatabaseTableMeta(){
}
@@ -68,19 +77,10 @@ public class DatabaseTableMeta implements TableMetaTSDB {
public boolean init(final String destination) {
this.destination = destination;
this.memoryTableMeta = new MemoryTableMeta();
this.scheduler = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r, "[scheduler-table-meta-snapshot]");
thread.setDaemon(true);
return thread;
}
});
// 24小时生成一份snapshot
if (snapshotInterval > 0) {
scheduler.scheduleWithFixedDelay(new Runnable() {
scheduleSnapshotFuture = scheduler.scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
@@ -105,6 +105,26 @@ public class DatabaseTableMeta implements TableMetaTSDB {
}
return true;
}
@Override
public void destory() {
if (memoryTableMeta != null) {
memoryTableMeta.destory();
}
if (connection != null) {
try {
connection.disconnect();
} catch (IOException e) {
logger.error("ERROR # disconnect meta connection for address:{}", connection.getConnector()
.getAddress(), e);
}
}
if (scheduleSnapshotFuture != null) {
scheduleSnapshotFuture.cancel(false);
}
}
@Override
public TableMeta find(String schema, String table) {

5
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java
View File

@@ -58,6 +58,11 @@ public class MemoryTableMeta implements TableMetaTSDB {
public boolean init(String destination) {
return true;
}
@Override
public void destory() {
tableMetas.clear();
}
public boolean apply(EntryPosition position, String schema, String ddl, String extra) {
tableMetas.clear();

5
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaTSDB.java
View File

@@ -18,6 +18,11 @@ public interface TableMetaTSDB {
*/
public boolean init(String destination);
/**
* 销毁资源
*/
public void destory();
/**
* 获取当前的表结构
*/

172
parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java
View File

@@ -1,7 +1,12 @@
package com.alibaba.otter.canal.parse;
import java.io.IOException;
import java.io.Serializable;
import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
import java.nio.charset.Charset;
import java.util.BitSet;
import java.util.List;
import org.apache.commons.lang.StringUtils;
import org.junit.Assert;
@@ -10,29 +15,49 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.parse.driver.mysql.MysqlConnector;
import com.alibaba.otter.canal.parse.driver.mysql.MysqlQueryExecutor;
import com.alibaba.otter.canal.parse.driver.mysql.MysqlUpdateExecutor;
import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.client.BinlogDumpCommandPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.client.RegisterSlaveCommandPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ErrorPacket;
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.mysql.dbsync.DirectLogFetcher;
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.event.DeleteRowsLogEvent;
import com.taobao.tddl.dbsync.binlog.event.FormatDescriptionLogEvent;
import com.taobao.tddl.dbsync.binlog.event.QueryLogEvent;
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;
import com.taobao.tddl.dbsync.binlog.event.TableMapLogEvent;
import com.taobao.tddl.dbsync.binlog.event.TableMapLogEvent.ColumnInfo;
import com.taobao.tddl.dbsync.binlog.event.UpdateRowsLogEvent;
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;
public class DirectLogFetcherTest {
protected final Logger logger = LoggerFactory.getLogger(this.getClass());
protected String binlogFileName = "mysql-bin.000001";
protected Charset charset = Charset.forName("utf-8");
private boolean isMariaDB;
private int binlogChecksum;
@Test
public void testSimple() {
DirectLogFetcher fetcher = new DirectLogFetcher();
try {
MysqlConnector connector = new MysqlConnector(new InetSocketAddress("127.0.0.1", 3306), "xxxx", "xxxx");
MysqlConnector connector = new MysqlConnector(new InetSocketAddress("127.0.0.1", 3306), "root", "hello");
connector.connect();
updateSettings(connector);
loadBinlogChecksum(connector);
sendRegisterSlave(connector, 3);
sendBinlogDump(connector, "mysql-bin.000001", 4L, 3);
@@ -40,6 +65,8 @@ public class DirectLogFetcherTest {
LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
LogContext context = new LogContext();
context.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum));
while (fetcher.fetch()) {
LogEvent event = null;
event = decoder.decode(fetcher, context);
@@ -57,21 +84,22 @@ public class DirectLogFetcherTest {
break;
case LogEvent.WRITE_ROWS_EVENT_V1:
case LogEvent.WRITE_ROWS_EVENT:
// parseRowsEvent((WriteRowsLogEvent) event);
parseRowsEvent((WriteRowsLogEvent) event);
break;
case LogEvent.UPDATE_ROWS_EVENT_V1:
case LogEvent.PARTIAL_UPDATE_ROWS_EVENT:
case LogEvent.UPDATE_ROWS_EVENT:
// parseRowsEvent((UpdateRowsLogEvent) event);
parseRowsEvent((UpdateRowsLogEvent) event);
break;
case LogEvent.DELETE_ROWS_EVENT_V1:
case LogEvent.DELETE_ROWS_EVENT:
// parseRowsEvent((DeleteRowsLogEvent) event);
parseRowsEvent((DeleteRowsLogEvent) event);
break;
case LogEvent.QUERY_EVENT:
// parseQueryEvent((QueryLogEvent) event);
parseQueryEvent((QueryLogEvent) event);
break;
case LogEvent.ROWS_QUERY_LOG_EVENT:
// parseRowsQueryEvent((RowsQueryLogEvent) event);
parseRowsQueryEvent((RowsQueryLogEvent) event);
break;
case LogEvent.ANNOTATE_ROWS_EVENT:
break;
@@ -188,9 +216,141 @@ public class DirectLogFetcherTest {
}
}
private void loadBinlogChecksum(MysqlConnector connector) {
checkMariaDB(connector);
if (isMariaDB) {
ResultSetPacket rs = null;
try {
rs = query("select @@global.binlog_checksum", connector);
} catch (IOException e) {
throw new CanalParseException(e);
}
List<String> columnValues = rs.getFieldValues();
if (columnValues != null && columnValues.size() >= 1 && columnValues.get(0).toUpperCase().equals("CRC32")) {
binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_CRC32;
} else {
binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_OFF;
}
}
}
private void checkMariaDB(MysqlConnector connector) {
ResultSetPacket rs = null;
try {
rs = query("SELECT @@version", connector);
} catch (IOException e) {
throw new CanalParseException(e);
}
List<String> columnValues = rs.getFieldValues();
if (columnValues != null && columnValues.size() >= 1) {
isMariaDB = StringUtils.containsIgnoreCase(columnValues.get(0), "MariaDB");
}
}
public ResultSetPacket query(String cmd, MysqlConnector connector) throws IOException {
MysqlQueryExecutor exector = new MysqlQueryExecutor(connector);
return exector.query(cmd);
}
public void update(String cmd, MysqlConnector connector) throws IOException {
MysqlUpdateExecutor exector = new MysqlUpdateExecutor(connector);
exector.update(cmd);
}
protected void parseQueryEvent(QueryLogEvent event) {
System.out.println(String.format("================> binlog[%s:%s] , name[%s]",
binlogFileName,
event.getHeader().getLogPos() - event.getHeader().getEventLen(),
event.getCatalog()));
System.out.println("sql : " + event.getQuery());
}
protected void parseRowsQueryEvent(RowsQueryLogEvent event) throws Exception {
System.out.println(String.format("================> binlog[%s:%s]", binlogFileName, event.getHeader()
.getLogPos() - event.getHeader().getEventLen()));
System.out.println("sql : " + new String(event.getRowsQuery().getBytes("ISO-8859-1"), charset.name()));
}
protected void parseAnnotateRowsEvent(AnnotateRowsEvent event) throws Exception {
System.out.println(String.format("================> binlog[%s:%s]", binlogFileName, event.getHeader()
.getLogPos() - event.getHeader().getEventLen()));
System.out.println("sql : " + new String(event.getRowsQuery().getBytes("ISO-8859-1"), charset.name()));
}
protected void parseXidEvent(XidLogEvent event) throws Exception {
System.out.println(String.format("================> binlog[%s:%s]", binlogFileName, event.getHeader()
.getLogPos() - event.getHeader().getEventLen()));
System.out.println("xid : " + event.getXid());
}
protected void parseRowsEvent(RowsLogEvent event) {
try {
System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s]",
binlogFileName,
event.getHeader().getLogPos() - event.getHeader().getEventLen(),
event.getTable().getDbName(),
event.getTable().getTableName()));
RowsLogBuffer buffer = event.getRowsBuf(charset.name());
BitSet columns = event.getColumns();
BitSet changeColumns = event.getChangeColumns();
while (buffer.nextOneRow(columns)) {
// 处理row记录
int type = event.getHeader().getType();
if (LogEvent.WRITE_ROWS_EVENT_V1 == type || LogEvent.WRITE_ROWS_EVENT == type) {
// insert的记录放在before字段中
parseOneRow(event, buffer, columns, true);
} else if (LogEvent.DELETE_ROWS_EVENT_V1 == type || LogEvent.DELETE_ROWS_EVENT == type) {
// delete的记录放在before字段中
parseOneRow(event, buffer, columns, false);
} else {
// update需要处理before/after
System.out.println("-------> before");
parseOneRow(event, buffer, columns, false);
if (!buffer.nextOneRow(changeColumns, true)) {
break;
}
System.out.println("-------> after");
parseOneRow(event, buffer, changeColumns, true);
}
}
} catch (Exception e) {
throw new RuntimeException("parse row data failed.", e);
}
}
protected void parseOneRow(RowsLogEvent event, RowsLogBuffer buffer, BitSet cols, boolean isAfter)
throws UnsupportedEncodingException {
TableMapLogEvent map = event.getTable();
if (map == null) {
throw new RuntimeException("not found TableMap with tid=" + event.getTableId());
}
final int columnCnt = map.getColumnCnt();
final ColumnInfo[] columnInfo = map.getColumnInfo();
for (int i = 0; i < columnCnt; i++) {
if (!cols.get(i)) {
continue;
}
ColumnInfo info = columnInfo[i];
buffer.nextValue(null, i, info.type, info.meta);
if (buffer.isNull()) {
//
} else {
final Serializable value = buffer.getValue();
if (value instanceof byte[]) {
System.out.println(new String((byte[]) value));
} else {
System.out.println(value);
}
}
}
}
}

5
parse/src/test/java/com/alibaba/otter/canal/parse/MysqlBinlogParsePerformanceTest.java
View File

@@ -89,6 +89,7 @@ public class MysqlBinlogParsePerformanceTest {
parseRowsEvent((WriteRowsLogEvent) event, sum);
break;
case LogEvent.UPDATE_ROWS_EVENT_V1:
case LogEvent.PARTIAL_UPDATE_ROWS_EVENT:
case LogEvent.UPDATE_ROWS_EVENT:
parseRowsEvent((UpdateRowsLogEvent) event, sum);
break;
@@ -154,7 +155,7 @@ public class MysqlBinlogParsePerformanceTest {
parseOneRow(event, buffer, columns, false);
} else {
parseOneRow(event, buffer, columns, false);
if (!buffer.nextOneRow(changeColumns)) {
if (!buffer.nextOneRow(changeColumns, true)) {
break;
}
parseOneRow(event, buffer, changeColumns, true);
@@ -182,7 +183,7 @@ public class MysqlBinlogParsePerformanceTest {
}
ColumnInfo info = columnInfo[i];
buffer.nextValue(info.type, info.meta);
buffer.nextValue(null, i, info.type, info.meta);
if (buffer.isNull()) {
} else {
buffer.getValue();

2
parse/src/test/resources/tsdb/h2-tsdb.xml
View File

@@ -7,7 +7,7 @@
default-autowire="byName">
<!-- 基于db的实现 -->
<bean id="tableMetaTSDB" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.DatabaseTableMeta">
<bean id="tableMetaTSDB" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.DatabaseTableMeta" destroy-method="destory">
<property name="metaHistoryDAO" ref="metaHistoryDAO"/>
<property name="metaSnapshotDAO" ref="metaSnapshotDAO"/>
</bean>

2
parse/src/test/resources/tsdb/mysql-tsdb.xml
View File

@@ -7,7 +7,7 @@
default-autowire="byName">
<!-- 基于db的实现 -->
<bean id="tableMetaTSDB" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.DatabaseTableMeta">
<bean id="tableMetaTSDB" class="com.alibaba.otter.canal.parse.inbound.mysql.tsdb.DatabaseTableMeta" destroy-method="destory">
<property name="metaHistoryDAO" ref="metaHistoryDAO"/>
<property name="metaSnapshotDAO" ref="metaSnapshotDAO"/>
</bean>