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 commit '9e816bc48f9955f9e2839873993cef9627c2f389' into branch-rds-support

Browse Source 
Conflicts:
	instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java
	instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/model/CanalParameter.java
	parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java
	parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java
This commit is contained in:
charles.lin
2018-08-09 02:01:00 +08:00
parent 61c7f7738d 9e816bc48f
commit 68d8afae3e
9 changed files with 145 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -14,27 +14,30 @@ 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.DefaultTableMetaTSDBFactory;
import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDB;
import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDBBuilder;
import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDBFactory;
import com.alibaba.otter.canal.protocol.position.EntryPosition;
public abstract class AbstractMysqlEventParser extends AbstractEventParser {
protected final Logger logger = LoggerFactory.getLogger(this.getClass());
protected static final long BINLOG_START_OFFEST = 4L;
protected final Logger logger = LoggerFactory.getLogger(this.getClass());
protected static final long BINLOG_START_OFFEST = 4L;
protected TableMetaTSDBFactory tableMetaTSDBFactory = new DefaultTableMetaTSDBFactory();
protected boolean enableTsdb = false;
protected String tsdbSpringXml;
protected TableMetaTSDB tableMetaTSDB;
protected boolean enableTsdb = false;
protected String tsdbSpringXml;
protected TableMetaTSDB tableMetaTSDB;
// 编码信息
protected byte connectionCharsetNumber = (byte) 33;
protected Charset connectionCharset = Charset.forName("UTF-8");
protected boolean filterQueryDcl = false;
protected boolean filterQueryDml = false;
protected boolean filterQueryDdl = false;
protected boolean filterRows = false;
protected boolean filterTableError = false;
protected boolean useDruidDdlFilter = true;
protected byte connectionCharsetNumber = (byte) 33;
protected Charset connectionCharset = Charset.forName("UTF-8");
protected boolean filterQueryDcl = false;
protected boolean filterQueryDml = false;
protected boolean filterQueryDdl = false;
protected boolean filterRows = false;
protected boolean filterTableError = false;
protected boolean useDruidDdlFilter = true;
protected BinlogParser buildParser() {
LogEventConvert convert = new LogEventConvert();
@@ -97,7 +100,7 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
// 设置当前正在加载的通道，加载spring查找文件时会用到该变量
System.setProperty("canal.instance.destination", destination);
// 初始化
tableMetaTSDB = TableMetaTSDBBuilder.build(destination, tsdbSpringXml);
tableMetaTSDB = tableMetaTSDBFactory.build(destination, tsdbSpringXml);
} finally {
System.setProperty("canal.instance.destination", "");
}
@@ -110,7 +113,7 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
public void stop() throws CanalParseException {
if (enableTsdb) {
TableMetaTSDBBuilder.destory(destination);
tableMetaTSDBFactory.destory(destination);
tableMetaTSDB = null;
}
@@ -182,7 +185,7 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
if (this.enableTsdb) {
if (tableMetaTSDB == null) {
// 初始化
tableMetaTSDB = TableMetaTSDBBuilder.build(destination, tsdbSpringXml);
tableMetaTSDB = tableMetaTSDBFactory.build(destination, tsdbSpringXml);
}
}
}
@@ -192,9 +195,13 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
if (this.enableTsdb) {
if (tableMetaTSDB == null) {
// 初始化
tableMetaTSDB = TableMetaTSDBBuilder.build(destination, tsdbSpringXml);
tableMetaTSDB = tableMetaTSDBFactory.build(destination, tsdbSpringXml);
}
}
}
public void setTableMetaTSDBFactory(TableMetaTSDBFactory tableMetaTSDBFactory) {
this.tableMetaTSDBFactory = tableMetaTSDBFactory;
}
}

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

@@ -10,11 +10,6 @@ import java.util.Map;
import java.util.TimerTask;
import java.util.concurrent.atomic.AtomicLong;
import com.alibaba.otter.canal.parse.inbound.*;
import com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.TableMetaCacheInterface;
import com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.TableMetaCacheWithStorage;
import com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.TableMetaStorage;
import com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.TableMetaStorageFactory;
import org.apache.commons.lang.StringUtils;
import org.springframework.util.CollectionUtils;
@@ -25,6 +20,9 @@ import com.alibaba.otter.canal.parse.driver.mysql.packets.server.FieldPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ResultSetPacket;
import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.parse.ha.CanalHAController;
import com.alibaba.otter.canal.parse.inbound.ErosaConnection;
import com.alibaba.otter.canal.parse.inbound.HeartBeatCallback;
import com.alibaba.otter.canal.parse.inbound.SinkFunction;
import com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection.BinlogFormat;
import com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection.BinlogImage;
import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert;
@@ -64,8 +62,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
// 心跳检查信息
private String detectingSQL; // 心跳sql
private MysqlConnection metaConnection; // 查询meta信息的链接
private TableMetaCacheInterface tableMetaCache; // 对应meta
// cache
private TableMetaCache tableMetaCache; // 对应meta
private int fallbackIntervalInSeconds = 60; // 切换回退时间
private BinlogFormat[] supportBinlogFormats; // 支持的binlogFormat,如果设置会执行强校验
private BinlogImage[] supportBinlogImages; // 支持的binlogImage,如果设置会执行强校验
@@ -74,8 +71,6 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
private int dumpErrorCount = 0; // binlogDump失败异常计数
private int dumpErrorCountThreshold = 2; // binlogDump失败异常计数阀值
private TableMetaStorageFactory tableMetaStorageFactory;
protected ErosaConnection buildErosaConnection() {
return buildMysqlConnection(this.runningInfo);
}
@@ -129,13 +124,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
((DatabaseTableMeta) tableMetaTSDB).setBlackFilter(eventBlackFilter);
}
TableMetaStorage storage = null;
if (tableMetaStorageFactory != null) {
storage = tableMetaStorageFactory.getTableMetaStorage();
}
tableMetaCache = new TableMetaCacheWithStorage(metaConnection, storage);
tableMetaCache = new TableMetaCache(metaConnection, tableMetaTSDB);
((LogEventConvert) binlogParser).setTableMetaCache(tableMetaCache);
}
}
@@ -921,11 +910,4 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
this.dumpErrorCountThreshold = dumpErrorCountThreshold;
}
public TableMetaStorageFactory getTableMetaStorageFactory() {
return tableMetaStorageFactory;
}
public void setTableMetaStorageFactory(TableMetaStorageFactory tableMetaStorageFactory) {
this.tableMetaStorageFactory = tableMetaStorageFactory;
}
}

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

@@ -548,12 +548,16 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
tableError |= parseOneRow(rowDataBuilder, event, buffer, changeColumns, true, tableMeta);
}
rowsCount ++;
rowsCount++;
rowChangeBuider.addRowDatas(rowDataBuilder.build());
}
TableMapLogEvent table = event.getTable();
Header header = createHeader(event.getHeader(), table.getDbName(), table.getTableName(), eventType, rowsCount);
Header header = createHeader(event.getHeader(),
table.getDbName(),
table.getTableName(),
eventType,
rowsCount);
RowChange rowChange = rowChangeBuider.build();
if (tableError) {
@@ -808,12 +812,12 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
return createEntry(header, EntryType.ROWDATA, rowChangeBuider.build().toByteString());
}
private Header createHeader(LogHeader logHeader, String schemaName, String tableName, EventType eventType) {
return createHeader(logHeader, schemaName, tableName, eventType, -1);
}
private Header createHeader(LogHeader logHeader, String schemaName, String tableName, EventType eventType, Integer rowsCount) {
private Header createHeader(LogHeader logHeader, String schemaName, String tableName, EventType eventType,
Integer rowsCount) {
// header会做信息冗余,方便以后做检索或者过滤
Header.Builder headerBuilder = Header.newBuilder();
headerBuilder.setVersion(version);

19
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DefaultTableMetaTSDBFactory.java Normal file
View File

@@ -0,0 +1,19 @@
package com.alibaba.otter.canal.parse.inbound.mysql.tsdb;
/**
* @author agapple 2017年10月11日 下午8:45:40
* @since 1.0.25
*/
public class DefaultTableMetaTSDBFactory implements TableMetaTSDBFactory {
/**
* 代理一下tableMetaTSDB的获取,使用隔离的spring定义
*/
public TableMetaTSDB build(String destination, String springXml) {
return TableMetaTSDBBuilder.build(destination, springXml);
}
public void destory(String destination) {
TableMetaTSDBBuilder.destory(destination);
}
}

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

@@ -10,12 +10,15 @@ import org.springframework.context.support.ClassPathXmlApplicationContext;
import com.google.common.collect.Maps;
/**
* @author agapple 2017年10月11日 下午8:45:40
* tableMeta构造器
*
* @author agapple 2018年8月8日 上午11:01:08
* @since 1.0.25
*/
public class TableMetaTSDBBuilder {
protected final static Logger logger = LoggerFactory.getLogger(TableMetaTSDBBuilder.class);
protected final static Logger logger = LoggerFactory.getLogger(DefaultTableMetaTSDBFactory.class);
private static ConcurrentMap<String, ClassPathXmlApplicationContext> contexts = Maps.newConcurrentMap();
/**

18
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaTSDBFactory.java Normal file
View File

@@ -0,0 +1,18 @@
package com.alibaba.otter.canal.parse.inbound.mysql.tsdb;
/**
* tableMeta构造器,允许重载实现
*
* @author agapple 2018年8月8日 上午11:01:08
* @since 1.0.26
*/
public interface TableMetaTSDBFactory {
/**
* 代理一下tableMetaTSDB的获取,使用隔离的spring定义
*/
public TableMetaTSDB build(String destination, String springXml);
public void destory(String destination);
}