From 4f00f5ea8da0ca78b1744376d2b08dd26a7e712b Mon Sep 17 00:00:00 2001 From: agapple Date: Sat, 14 Oct 2017 00:12:43 +0800 Subject: [PATCH] fixed tsdb test --- common/pom.xml | 34 +++- .../tddl/dbsync/binlog/BinlogPosition.java | 142 ---------------- deployer/src/main/resources/canal.properties | 3 + .../resources/example/instance.properties | 14 +- .../resources/spring/default-instance.xml | 5 + .../main/resources/spring/file-instance.xml | 40 +---- .../main/resources/spring/group-instance.xml | 2 + .../main/resources/spring/local-instance.xml | 5 + .../main/resources/spring/memory-instance.xml | 5 + .../main/resources/spring/tsdb/dal-dao.xml | 39 ----- .../resources/spring/tsdb/mybatis-config.xml | 20 --- .../main/resources/spring/tsdb/mysql-tsdb.xml | 63 +++++++ .../tsdb/{ => sql-map}/sqlmap-config.xml | 5 +- .../spring/tsdb/sql-map/sqlmap_history.xml | 38 ++--- .../spring/tsdb/sql-map/sqlmap_snapshot.xml | 53 +++--- .../spring/tsdb/sql/create_table.sql | 39 +++++ .../driver/mysql/socket/SocketChannel.java | 11 +- .../canal/meta/FileMixedMetaManager.java | 2 +- .../canal/meta/FileMixedMetaManagerTest.java | 10 +- parse/pom.xml | 4 +- .../parse/inbound/AbstractEventParser.java | 10 +- .../mysql/AbstractMysqlEventParser.java | 71 +++++--- .../inbound/mysql/LocalBinlogEventParser.java | 2 +- .../parse/inbound/mysql/MysqlEventParser.java | 51 +++++- .../inbound/mysql/dbsync/LogEventConvert.java | 19 ++- .../inbound/mysql/dbsync/TableMetaCache.java | 36 ++-- .../mysql/tsdb/DataSourceFactoryTSDB.java | 41 ----- .../inbound/mysql/tsdb/MemoryTableMeta.java | 33 +++- .../inbound/mysql/tsdb/TableMetaManager.java | 160 ++++++++++-------- .../inbound/mysql/tsdb/TableMetaTSDB.java | 14 +- .../mysql/tsdb/TableMetaTSDBBuilder.java | 51 ++++++ .../mysql/tsdb/dao/MetaHistoryDAO.java | 65 +++++-- .../tsdb/{model => dao}/MetaHistoryDO.java | 105 ++++++------ .../mysql/tsdb/dao/MetaSnapshotDAO.java | 61 +++++-- .../tsdb/{model => dao}/MetaSnapshotDO.java | 33 ++-- .../main/resources/ddl/derby/meta_history.sql | 22 +++ .../resources/ddl/derby/meta_snapshot.sql | 18 ++ .../main/resources/ddl/mysql/meta_history.sql | 21 +++ .../resources/ddl/mysql/meta_snapshot.sql | 17 ++ .../mysql/LocalBinlogEventParserTest.java | 2 +- .../parse/inbound/mysql/MysqlDumpTest.java | 10 +- .../mysql/tsdb/MemoryTableMetaTest.java | 11 +- .../mysql/tsdb/MetaHistoryDAOTest.java | 12 +- .../tsdb/TableMetaManagerBuilderTest.java | 18 ++ .../mysql/tsdb/TableMetaManagerTest.java | 14 +- parse/src/test/resources/ddl/create.sql | 2 +- parse/src/test/resources/tsdb/dal-dao.xml | 39 ----- parse/src/test/resources/tsdb/derby-tsdb.xml | 44 +++++ .../test/resources/tsdb/mybatis-config.xml | 20 --- parse/src/test/resources/tsdb/mysql-tsdb.xml | 47 +++++ .../tsdb/{ => sql-map}/sqlmap-config.xml | 5 +- .../resources/tsdb/sql-map/sqlmap_history.xml | 44 ++--- .../tsdb/sql-map/sqlmap_snapshot.xml | 63 ++++--- pom.xml | 34 +++- .../protocol/position/EntryPosition.java | 15 ++ 55 files changed, 1002 insertions(+), 742 deletions(-) delete mode 100644 dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/BinlogPosition.java delete mode 100644 deployer/src/main/resources/spring/tsdb/dal-dao.xml delete mode 100644 deployer/src/main/resources/spring/tsdb/mybatis-config.xml create mode 100644 deployer/src/main/resources/spring/tsdb/mysql-tsdb.xml rename deployer/src/main/resources/spring/tsdb/{ => sql-map}/sqlmap-config.xml (66%) create mode 100644 deployer/src/main/resources/spring/tsdb/sql/create_table.sql delete mode 100644 parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DataSourceFactoryTSDB.java create mode 100644 parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaTSDBBuilder.java rename parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/{model => dao}/MetaHistoryDO.java (55%) rename parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/{model => dao}/MetaSnapshotDO.java (72%) create mode 100644 parse/src/main/resources/ddl/derby/meta_history.sql create mode 100644 parse/src/main/resources/ddl/derby/meta_snapshot.sql create mode 100644 parse/src/main/resources/ddl/mysql/meta_history.sql create mode 100644 parse/src/main/resources/ddl/mysql/meta_snapshot.sql create mode 100644 parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaManagerBuilderTest.java delete mode 100644 parse/src/test/resources/tsdb/dal-dao.xml create mode 100644 parse/src/test/resources/tsdb/derby-tsdb.xml delete mode 100644 parse/src/test/resources/tsdb/mybatis-config.xml create mode 100644 parse/src/test/resources/tsdb/mysql-tsdb.xml rename parse/src/test/resources/tsdb/{ => sql-map}/sqlmap-config.xml (68%) diff --git a/common/pom.xml b/common/pom.xml index 6a6df57b..4789ac9e 100644 --- a/common/pom.xml +++ b/common/pom.xml @@ -1,4 +1,5 @@ - + 4.0.0 com.alibaba.otter @@ -34,10 +35,6 @@ commons-lang commons-lang - - org.springframework - spring - com.alibaba fastjson @@ -63,10 +60,37 @@ org.slf4j slf4j-api + + org.springframework + spring-core + + + org.springframework + spring-aop + + + org.springframework + spring-context + + + org.springframework + spring-jdbc + + + org.springframework + spring-orm + + + org.springframework + spring-test + ${spring_version} + test + junit junit + test diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/BinlogPosition.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/BinlogPosition.java deleted file mode 100644 index a602c05f..00000000 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/BinlogPosition.java +++ /dev/null @@ -1,142 +0,0 @@ -package com.taobao.tddl.dbsync.binlog; - - -/** - * Position inside binlog file - * - * @author Seppo Jaakola - * @version 1.0 - */ -public class BinlogPosition extends LogPosition { - - /* The source server_id of position, 0 invalid */ - protected final long masterId; - - /* The timestamp, in seconds, 0 invalid */ - protected final long timestamp; - - public BinlogPosition(String fileName, long position, long masterId, long timestamp){ - super(fileName, position); - this.masterId = masterId; - this.timestamp = timestamp; - } - - public BinlogPosition(LogPosition logPosition, long masterId, long timestamp){ - super(logPosition.getFileName(), logPosition.getPosition()); - this.masterId = masterId; - this.timestamp = timestamp; - } - - public BinlogPosition(BinlogPosition binlogPosition){ - super(binlogPosition.getFileName(), binlogPosition.getPosition()); - this.masterId = binlogPosition.masterId; - this.timestamp = binlogPosition.timestamp; - } - - private final static long[] pow10 = { 1, 10, 100, 1000, 10000, 100000, 1000000, 10000000, 100000000, 1000000000, - 10000000000L, 100000000000L, 1000000000000L, 10000000000000L, 100000000000000L, 1000000000000000L, - 10000000000000000L, 100000000000000000L, 1000000000000000000L }; - - public static String placeHolder(int bit, long number) { - if (bit > 18) { - throw new IllegalArgumentException("Bit must less than 18, but given " + bit); - } - - final long max = pow10[bit]; - if (number >= max) { - // 当 width < 数值的最大位数时，应该直接返回数值 - return String.valueOf(number); - } - - return String.valueOf(max + number).substring(1); - } - - public String format2String(final int positionMaxLen) { - String binlogSuffix = fileName; - String binlogOffset = placeHolder((int) positionMaxLen, position); - // 输出 '000001:0000000004@12+12314130' - StringBuffer buf = new StringBuffer(40); - buf.append(binlogSuffix); - buf.append(':'); - buf.append(binlogOffset); - if (masterId != 0) { - buf.append('#'); - buf.append(masterId); - } - if (timestamp != 0) { - buf.append('.'); - buf.append(timestamp); - } - return buf.toString(); - } - - public static BinlogPosition parseFromString(String source) { - int colonIndex = source.indexOf(':'); - int miscIndex = colonIndex + 1; - int sharpIndex = source.indexOf('#', miscIndex); - int semicolonIndex = source.indexOf(';', miscIndex); // NOTE: 向后兼容 - int dotIndex = source.lastIndexOf('.'); - if (colonIndex == -1) { - return null; // NOTE: 错误的位点 - } - - String binlogSuffix = source.substring(0, colonIndex); - long binlogPosition; - if (sharpIndex != -1) { - binlogPosition = Long.parseLong(source.substring(miscIndex, sharpIndex)); - } else if (semicolonIndex != -1) { - binlogPosition = Long.parseLong(source.substring(miscIndex, semicolonIndex)); // NOTE: - // 向后兼容 - } else if (dotIndex != -1) { - binlogPosition = Long.parseLong(source.substring(miscIndex, dotIndex)); - } else { - binlogPosition = Long.parseLong(source.substring(miscIndex)); - } - - long masterId = 0; // NOTE: 默认值为 0 - if (sharpIndex != -1) { - if (dotIndex != -1) { - masterId = Long.parseLong(source.substring(sharpIndex + 1, dotIndex)); - } else { - masterId = Long.parseLong(source.substring(sharpIndex + 1)); - } - } - - long timestamp = 0; // NOTE: 默认值为 0 - if (dotIndex != -1 && dotIndex > colonIndex) { - timestamp = Long.parseLong(source.substring(dotIndex + 1)); - } - - return new BinlogPosition(binlogSuffix, binlogPosition, // NL - masterId, - timestamp); - } - - public String getFilePattern() { - final int index = fileName.indexOf('.'); - if (index != -1) { - return fileName.substring(0, index); - } - return null; - } - - public void setFilePattern(String filePattern) { - // We tolerate the event ID with or without the binlog prefix. - if (fileName.indexOf('.') < 0) { - fileName = filePattern + '.' + fileName; - } - } - - public long getMasterId() { - return masterId; - } - - public long getTimestamp() { - return timestamp; - } - - @Override - public String toString() { - return format2String(10); - } -} diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index df6f49e2..09616194 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -36,6 +36,7 @@ canal.instance.network.sendBufferSize = 16384 canal.instance.network.soTimeout = 30 # binlog filter config +canal.instance.filter.druid.ddl = true canal.instance.filter.query.dcl = false canal.instance.filter.query.dml = false canal.instance.filter.query.ddl = false @@ -59,6 +60,8 @@ canal.conf.dir = ../conf canal.auto.scan = true canal.auto.scan.interval = 5 +canal.instance.tsdb.spring.xml=classpath:spring/tsdb/mysql-tsdb.xml + canal.instance.global.mode = spring canal.instance.global.lazy = false #canal.instance.global.manager.address = 127.0.0.1:1099 diff --git a/deployer/src/main/resources/example/instance.properties b/deployer/src/main/resources/example/instance.properties index 7f00f079..50640d99 100644 --- a/deployer/src/main/resources/example/instance.properties +++ b/deployer/src/main/resources/example/instance.properties @@ -3,14 +3,14 @@ canal.instance.mysql.slaveId=1234 # position info canal.instance.master.address=127.0.0.1:3306 -canal.instance.master.journal.name=mysql-bin.000004 -canal.instance.master.position=25678 -canal.instance.master.timestamp=1506088042 +canal.instance.master.journal.name=mysql-bin.000001 +canal.instance.master.position=68240 +canal.instance.master.timestamp= -# tsdb info +# table meta tsdb info canal.instance.tsdb.enable=true -canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/tsdb +canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb canal.instance.tsdb.dbUsername=canal canal.instance.tsdb.dbPassword=canal @@ -25,7 +25,7 @@ canal.instance.dbPassword=canal canal.instance.defaultDatabaseName=test canal.instance.connectionCharset=UTF-8 # table regex -canal.instance.filter.regex=.*\\..* +canal.instance.filter.regex=test\\..* # table black regex -canal.instance.filter.black.regex=tsdb\\..* +canal.instance.filter.black.regex=canal_tsdb\\..* ################################################# \ No newline at end of file diff --git a/deployer/src/main/resources/spring/default-instance.xml b/deployer/src/main/resources/spring/default-instance.xml index 5631b7cb..9c7ef5aa 100644 --- a/deployer/src/main/resources/spring/default-instance.xml +++ b/deployer/src/main/resources/spring/default-instance.xml @@ -178,9 +178,14 @@ + + + + + diff --git a/deployer/src/main/resources/spring/file-instance.xml b/deployer/src/main/resources/spring/file-instance.xml index a9f2823b..4c6f750d 100644 --- a/deployer/src/main/resources/spring/file-instance.xml +++ b/deployer/src/main/resources/spring/file-instance.xml @@ -163,48 +163,14 @@ + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + + diff --git a/deployer/src/main/resources/spring/group-instance.xml b/deployer/src/main/resources/spring/group-instance.xml index 6fefaa87..f5e872f3 100644 --- a/deployer/src/main/resources/spring/group-instance.xml +++ b/deployer/src/main/resources/spring/group-instance.xml @@ -160,6 +160,7 @@ + @@ -250,6 +251,7 @@ + diff --git a/deployer/src/main/resources/spring/local-instance.xml b/deployer/src/main/resources/spring/local-instance.xml index e5e4e695..e2fb44b3 100644 --- a/deployer/src/main/resources/spring/local-instance.xml +++ b/deployer/src/main/resources/spring/local-instance.xml @@ -155,9 +155,14 @@ + + + + + diff --git a/deployer/src/main/resources/spring/memory-instance.xml b/deployer/src/main/resources/spring/memory-instance.xml index 269b9deb..129ce2b0 100644 --- a/deployer/src/main/resources/spring/memory-instance.xml +++ b/deployer/src/main/resources/spring/memory-instance.xml @@ -151,9 +151,14 @@ + + + + + diff --git a/deployer/src/main/resources/spring/tsdb/dal-dao.xml b/deployer/src/main/resources/spring/tsdb/dal-dao.xml deleted file mode 100644 index 1842641f..00000000 --- a/deployer/src/main/resources/spring/tsdb/dal-dao.xml +++ /dev/null @@ -1,39 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/deployer/src/main/resources/spring/tsdb/mybatis-config.xml b/deployer/src/main/resources/spring/tsdb/mybatis-config.xml deleted file mode 100644 index 83ac0116..00000000 --- a/deployer/src/main/resources/spring/tsdb/mybatis-config.xml +++ /dev/null @@ -1,20 +0,0 @@ - - - - - - - - - - - - - - - - - - \ No newline at end of file diff --git a/deployer/src/main/resources/spring/tsdb/mysql-tsdb.xml b/deployer/src/main/resources/spring/tsdb/mysql-tsdb.xml new file mode 100644 index 00000000..5f02431d --- /dev/null +++ b/deployer/src/main/resources/spring/tsdb/mysql-tsdb.xml @@ -0,0 +1,63 @@ + + + + + + + + + + classpath:canal.properties + classpath:${canal.instance.destination:}/instance.properties + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/deployer/src/main/resources/spring/tsdb/sqlmap-config.xml b/deployer/src/main/resources/spring/tsdb/sql-map/sqlmap-config.xml similarity index 66% rename from deployer/src/main/resources/spring/tsdb/sqlmap-config.xml rename to deployer/src/main/resources/spring/tsdb/sql-map/sqlmap-config.xml index 983e9663..26690703 100644 --- a/deployer/src/main/resources/spring/tsdb/sqlmap-config.xml +++ b/deployer/src/main/resources/spring/tsdb/sql-map/sqlmap-config.xml @@ -3,7 +3,6 @@ "http://www.ibatis.com/dtd/sql-map-config-2.dtd"> - - - + + diff --git a/deployer/src/main/resources/spring/tsdb/sql-map/sqlmap_history.xml b/deployer/src/main/resources/spring/tsdb/sql-map/sqlmap_history.xml index a983e222..7445bf69 100644 --- a/deployer/src/main/resources/spring/tsdb/sql-map/sqlmap_history.xml +++ b/deployer/src/main/resources/spring/tsdb/sql-map/sqlmap_history.xml @@ -1,51 +1,45 @@ - - - - + + - insert into `canal_table_meta_history` () - values(now(),now(),#binlogFile#,#binlogOffest#,#binlogMasterId#,#binlogTimestamp#,#useSchema#,#schema#,#table#,#sql#,#type#,#extra#); - - SELECT last_insert_id() - + insert into meta_history () + values(CURRENT_TIMESTAMP,CURRENT_TIMESTAMP,#destination#,#binlogFile#,#binlogOffest#,#binlogMasterId#,#binlogTimestamp#,#useSchema#,#sqlSchema#,#sqlTable#,#sqlText#,#sqlType#,#extra#) + + + delete from meta_history + where destination=#destination# + - - - - \ No newline at end of file diff --git a/deployer/src/main/resources/spring/tsdb/sql-map/sqlmap_snapshot.xml b/deployer/src/main/resources/spring/tsdb/sql-map/sqlmap_snapshot.xml index 9065a7e0..79aa25fe 100644 --- a/deployer/src/main/resources/spring/tsdb/sql-map/sqlmap_snapshot.xml +++ b/deployer/src/main/resources/spring/tsdb/sql-map/sqlmap_snapshot.xml @@ -1,66 +1,51 @@ - - - - - - + + - + - insert into `canal_table_meta_snapshot` () - values(now(),now(),#binlogFile#,#binlogOffest#,#binlogMasterId#,#binlogTimestamp#,#data#,#extra#); - - SELECT last_insert_id() - + insert into meta_snapshot () + values(CURRENT_TIMESTAMP,CURRENT_TIMESTAMP,#destination#,#binlogFile#,#binlogOffest#,#binlogMasterId#,#binlogTimestamp#,#data#,#extra#) - update `canal_table_meta_snapshot` set gmt_modified=now(), + update meta_snapshot set gmt_modified=now(), binlog_file=#binlogFile#,binlog_offest=#binlogOffest#,binlog_master_id=#binlogMasterId#,binlog_timestamp=#binlogTimestamp#,data=#data#,extra=#extra# - where binlog_timestamp=0 + where destination=#destination# and binlog_timestamp=0 + + delete from meta_snapshot + where destination=#destination# + - - - \ No newline at end of file diff --git a/deployer/src/main/resources/spring/tsdb/sql/create_table.sql b/deployer/src/main/resources/spring/tsdb/sql/create_table.sql new file mode 100644 index 00000000..e5d6dbf4 --- /dev/null +++ b/deployer/src/main/resources/spring/tsdb/sql/create_table.sql @@ -0,0 +1,39 @@ +CREATE TABLE IF NOT EXISTS `meta_snapshot` ( + `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键', + `gmt_create` datetime NOT NULL COMMENT '创建时间', + `gmt_modified` datetime NOT NULL COMMENT '修改时间', + `destination` varchar(128) DEFAULT NULL COMMENT '通道名称', + `binlog_file` varchar(64) DEFAULT NULL COMMENT 'binlog文件名', + `binlog_offest` bigint(20) DEFAULT NULL COMMENT 'binlog偏移量', + `binlog_master_id` varchar(64) DEFAULT NULL COMMENT 'binlog节点id', + `binlog_timestamp` bigint(20) DEFAULT NULL COMMENT 'binlog应用的时间戳', + `data` longtext DEFAULT NULL COMMENT '表结构数据', + `extra` text DEFAULT NULL COMMENT '额外的扩展信息', + PRIMARY KEY (`id`), + UNIQUE KEY binlog_file_offest(`destination`,`binlog_master_id`,`binlog_file`,`binlog_offest`), + KEY `destination` (`destination`), + KEY `destination_timestamp` (`destination`,`binlog_timestamp`), + KEY `gmt_modified` (`gmt_modified`) +) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='表结构记录表快照表'; + +CREATE TABLE IF NOT EXISTS `meta_history` ( + `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键', + `gmt_create` datetime NOT NULL COMMENT '创建时间', + `gmt_modified` datetime NOT NULL COMMENT '修改时间', + `destination` varchar(128) DEFAULT NULL COMMENT '通道名称', + `binlog_file` varchar(64) DEFAULT NULL COMMENT 'binlog文件名', + `binlog_offest` bigint(20) DEFAULT NULL COMMENT 'binlog偏移量', + `binlog_master_id` varchar(64) DEFAULT NULL COMMENT 'binlog节点id', + `binlog_timestamp` bigint(20) DEFAULT NULL COMMENT 'binlog应用的时间戳', + `use_schema` varchar(1024) DEFAULT NULL COMMENT '执行sql时对应的schema', + `schema` varchar(1024) DEFAULT NULL COMMENT '对应的schema', + `table` varchar(1024) DEFAULT NULL COMMENT '对应的table', + `sql` longtext DEFAULT NULL COMMENT '执行的sql', + `type` varchar(256) DEFAULT NULL COMMENT 'sql类型', + `extra` text DEFAULT NULL COMMENT '额外的扩展信息', + PRIMARY KEY (`id`), + UNIQUE KEY binlog_file_offest(`destination`,`binlog_master_id`,`binlog_file`,`binlog_offest`), + KEY `destination` (`destination`), + KEY `destination_timestamp` (`destination`,`binlog_timestamp`), + KEY `gmt_modified` (`gmt_modified`) +) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='表结构变化明细表'; \ No newline at end of file diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannel.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannel.java index e825aa2b..a4568887 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannel.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannel.java @@ -35,21 +35,24 @@ public class SocketChannel { } public void writeChannel(byte[]... buf) throws IOException { - if (channel != null && channel.isWritable()) channel.writeAndFlush(Unpooled.copiedBuffer(buf)); - else throw new IOException("write failed ! please checking !"); + if (channel != null && channel.isWritable()) { + channel.writeAndFlush(Unpooled.copiedBuffer(buf)); + } else { + throw new IOException("write failed ! please checking !"); + } } public byte[] read(int readSize) throws IOException { do { if (readSize > cache.readableBytes()) { if (null == channel) { - throw new IOException("socket has Interrupted !"); + throw new java.nio.channels.ClosedByInterruptException(); } synchronized (this) { try { wait(100); } catch (InterruptedException e) { - throw new IOException("socket has Interrupted !"); + throw new java.nio.channels.ClosedByInterruptException(); } } } else { diff --git a/meta/src/main/java/com/alibaba/otter/canal/meta/FileMixedMetaManager.java b/meta/src/main/java/com/alibaba/otter/canal/meta/FileMixedMetaManager.java index 72e3646b..383fb351 100644 --- a/meta/src/main/java/com/alibaba/otter/canal/meta/FileMixedMetaManager.java +++ b/meta/src/main/java/com/alibaba/otter/canal/meta/FileMixedMetaManager.java @@ -367,7 +367,7 @@ public class FileMixedMetaManager extends MemoryMetaManager implements CanalMeta this.dataDir = new File(dataDir); } - public void setDataDir(File dataDir) { + public void setDataDirByFile(File dataDir) { this.dataDir = dataDir; } diff --git a/meta/src/test/java/com/alibaba/otter/canal/meta/FileMixedMetaManagerTest.java b/meta/src/test/java/com/alibaba/otter/canal/meta/FileMixedMetaManagerTest.java index 56ea51a8..4c330c86 100644 --- a/meta/src/test/java/com/alibaba/otter/canal/meta/FileMixedMetaManagerTest.java +++ b/meta/src/test/java/com/alibaba/otter/canal/meta/FileMixedMetaManagerTest.java @@ -31,7 +31,7 @@ public class FileMixedMetaManagerTest extends AbstractMetaManagerTest { @Test public void testSubscribeAll() { FileMixedMetaManager metaManager = new FileMixedMetaManager(); - metaManager.setDataDir(dataDir); + metaManager.setDataDirByFile(dataDir); metaManager.setPeriod(100); metaManager.start(); @@ -40,7 +40,7 @@ public class FileMixedMetaManagerTest extends AbstractMetaManagerTest { sleep(2000L); // 重新构建一次，能获得上一次zk上的记录 FileMixedMetaManager metaManager2 = new FileMixedMetaManager(); - metaManager2.setDataDir(dataDir); + metaManager2.setDataDirByFile(dataDir); metaManager2.setPeriod(100); metaManager2.start(); @@ -52,7 +52,7 @@ public class FileMixedMetaManagerTest extends AbstractMetaManagerTest { @Test public void testBatchAll() { FileMixedMetaManager metaManager = new FileMixedMetaManager(); - metaManager.setDataDir(dataDir); + metaManager.setDataDirByFile(dataDir); metaManager.setPeriod(100); metaManager.start(); @@ -67,7 +67,7 @@ public class FileMixedMetaManagerTest extends AbstractMetaManagerTest { @Test public void testCursorAll() { FileMixedMetaManager metaManager = new FileMixedMetaManager(); - metaManager.setDataDir(dataDir); + metaManager.setDataDirByFile(dataDir); metaManager.setPeriod(100); metaManager.start(); @@ -76,7 +76,7 @@ public class FileMixedMetaManagerTest extends AbstractMetaManagerTest { sleep(1000L); // 重新构建一次，能获得上一次zk上的记录 FileMixedMetaManager metaManager2 = new FileMixedMetaManager(); - metaManager2.setDataDir(dataDir); + metaManager2.setDataDirByFile(dataDir); metaManager2.setPeriod(100); metaManager2.start(); diff --git a/parse/pom.xml b/parse/pom.xml index 0adf86e0..66e4bb60 100644 --- a/parse/pom.xml +++ b/parse/pom.xml @@ -1,4 +1,5 @@ - + 4.0.0 com.alibaba.otter @@ -56,7 +57,6 @@ org.apache.ibatis ibatis-sqlmap - 2.3.4.726 diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java index 32d3aee5..1ea6af96 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java @@ -96,6 +96,10 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle protected void preDump(ErosaConnection connection) { } + protected boolean processTableMeta(EntryPosition position) { + return true; + } + protected void afterDump(ErosaConnection connection) { } @@ -145,7 +149,6 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle ErosaConnection erosaConnection = null; while (running) { try { - // 开始执行replication // 1. 构造Erosa连接 erosaConnection = buildErosaConnection(); @@ -163,6 +166,11 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle if (startPosition == null) { throw new CanalParseException("can't find start position for " + destination); } + + if (!processTableMeta(startPosition)) { + throw new CanalParseException("can't find init table meta for " + destination + + " with position : " + startPosition); + } logger.info("find start position : {}", startPosition.toString()); // 重新链接，因为在找position过程中可能有状态，需要断开后重建 erosaConnection.reconnect(); 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 094d0b1d..6d69a836 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 @@ -2,24 +2,27 @@ package com.alibaba.otter.canal.parse.inbound.mysql; import java.nio.charset.Charset; -import javax.annotation.Resource; - -import com.taobao.tddl.dbsync.binlog.BinlogPosition; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.alibaba.otter.canal.filter.CanalEventFilter; import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter; +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.mysql.dbsync.LogEventConvert; -import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaManager; +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.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 boolean enableTsdb = false; + protected String tsdbSpringXml; + protected TableMetaTSDB tableMetaTSDB; // 编码信息 protected byte connectionCharsetNumber = (byte) 33; protected Charset connectionCharset = Charset.forName("UTF-8"); @@ -28,9 +31,6 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { protected boolean filterQueryDdl = false; protected boolean filterRows = false; protected boolean filterTableError = false; - - @Resource - protected TableMetaManager tableMetaManager; protected boolean useDruidDdlFilter = true; protected BinlogParser buildParser() { @@ -49,9 +49,6 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { convert.setFilterQueryDdl(filterQueryDdl); convert.setFilterRows(filterRows); convert.setFilterTableError(filterTableError); - - // 初始化parser的时候也初始化管理mysql 表结构的管理器 - tableMetaManager.init(); return convert; } @@ -70,14 +67,34 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { * @param position * @return */ - protected boolean processTableMeta(BinlogPosition position) { - if (tableMetaManager != null) { - return tableMetaManager.rollback(position); + protected boolean processTableMeta(EntryPosition position) { + if (tableMetaTSDB != null) { + return tableMetaTSDB.rollback(position); } return true; } + public void start() throws CanalParseException { + if (enableTsdb) { + if (tableMetaTSDB == null) { + // 初始化 + tableMetaTSDB = TableMetaTSDBBuilder.build(destination, tsdbSpringXml); + } + } + + super.start(); + } + + public void stop() throws CanalParseException { + if (enableTsdb) { + TableMetaTSDBBuilder.destory(destination); + tableMetaTSDB = null; + } + + super.stop(); + } + public void setEventBlackFilter(CanalEventFilter eventBlackFilter) { super.setEventBlackFilter(eventBlackFilter); @@ -122,14 +139,6 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { this.filterTableError = filterTableError; } - public TableMetaManager getTableMetaManager() { - return tableMetaManager; - } - - public void setTableMetaManager(TableMetaManager tableMetaManager) { - this.tableMetaManager = tableMetaManager; - } - public boolean isUseDruidDdlFilter() { return useDruidDdlFilter; } @@ -137,4 +146,24 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { public void setUseDruidDdlFilter(boolean useDruidDdlFilter) { this.useDruidDdlFilter = useDruidDdlFilter; } + + public void setEnableTsdb(boolean enableTsdb) { + this.enableTsdb = enableTsdb; + if (this.enableTsdb) { + if (tableMetaTSDB == null) { + // 初始化 + tableMetaTSDB = TableMetaTSDBBuilder.build(destination, tsdbSpringXml); + } + } + } + + public void setTsdbSpringXml(String tsdbSpringXml) { + this.tsdbSpringXml = tsdbSpringXml; + + if (tableMetaTSDB == null) { + // 初始化 + tableMetaTSDB = TableMetaTSDBBuilder.build(destination, tsdbSpringXml); + } + } + } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java index d45c5615..d2c2fd2e 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java @@ -50,7 +50,7 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements throw new CanalParseException(e); } - tableMetaCache = new TableMetaCache(metaConnection, tableMetaManager); + tableMetaCache = new TableMetaCache(metaConnection, tableMetaTSDB); ((LogEventConvert) binlogParser).setTableMetaCache(tableMetaCache); } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java index 4a188b01..912787b4 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java @@ -26,6 +26,7 @@ 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; import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.TableMetaCache; +import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaManager; import com.alibaba.otter.canal.parse.support.AuthenticationInfo; import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.position.EntryPosition; @@ -117,12 +118,13 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE } } - if (tableMetaManager != null) { - tableMetaManager.setConnection(metaConnection); - tableMetaManager.setFilter(eventFilter); + if (tableMetaTSDB != null && tableMetaTSDB instanceof TableMetaManager) { + ((TableMetaManager) tableMetaTSDB).setConnection(metaConnection); + ((TableMetaManager) tableMetaTSDB).setFilter(eventFilter); + ((TableMetaManager) tableMetaTSDB).setBlackFilter(eventBlackFilter); } - tableMetaCache = new TableMetaCache(metaConnection, tableMetaManager); + tableMetaCache = new TableMetaCache(metaConnection, tableMetaTSDB); ((LogEventConvert) binlogParser).setTableMetaCache(tableMetaCache); } } @@ -337,6 +339,34 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE return endPosition; } + protected EntryPosition findEndPositionWithMasterIdAndTimestamp(MysqlConnection connection) { + MysqlConnection mysqlConnection = (MysqlConnection) connection; + final EntryPosition endPosition = findEndPosition(mysqlConnection); + if (tableMetaTSDB != null) { + long startTimestamp = System.currentTimeMillis(); + return findAsPerTimestampInSpecificLogFile(mysqlConnection, + startTimestamp, + endPosition, + endPosition.getJournalName()); + } else { + return endPosition; + } + } + + protected EntryPosition findPositionWithMasterIdAndTimestamp(MysqlConnection connection, EntryPosition fixedPosition) { + MysqlConnection mysqlConnection = (MysqlConnection) connection; + if (tableMetaTSDB != null && (fixedPosition.getTimestamp() == null || fixedPosition.getTimestamp() <= 0)) { + // 使用一个未来极大的时间，基于位点进行定位 + long startTimestamp = System.currentTimeMillis() + 102L * 365 * 24 * 3600 * 1000; // 当前时间的未来102年 + return findAsPerTimestampInSpecificLogFile(mysqlConnection, + startTimestamp, + fixedPosition, + fixedPosition.getJournalName()); + } else { + return fixedPosition; + } + } + protected EntryPosition findStartPositionInternal(ErosaConnection connection) { MysqlConnection mysqlConnection = (MysqlConnection) connection; LogPosition logPosition = logPositionManager.getLatestIndexBy(destination); @@ -350,7 +380,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE } if (entryPosition == null) { - entryPosition = findEndPosition(mysqlConnection); // 默认从当前最后一个位置进行消费 + entryPosition = findEndPositionWithMasterIdAndTimestamp(mysqlConnection); // 默认从当前最后一个位置进行消费 } // 判断一下是否需要按时间订阅 @@ -362,13 +392,15 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE return findByStartTimeStamp(mysqlConnection, entryPosition.getTimestamp()); } else { logger.warn("prepare to find start position just show master status"); - return findEndPosition(mysqlConnection); // 默认从当前最后一个位置进行消费 + return findEndPositionWithMasterIdAndTimestamp(mysqlConnection); // 默认从当前最后一个位置进行消费 } } else { if (entryPosition.getPosition() != null && entryPosition.getPosition() > 0L) { // 如果指定binlogName + offest，直接返回 + entryPosition = findPositionWithMasterIdAndTimestamp(mysqlConnection, entryPosition); logger.warn("prepare to find start position {}:{}:{}", - new Object[] { entryPosition.getJournalName(), entryPosition.getPosition(), "" }); + new Object[] { entryPosition.getJournalName(), entryPosition.getPosition(), + entryPosition.getTimestamp() }); return entryPosition; } else { EntryPosition specificLogFilePosition = null; @@ -695,6 +727,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE String logfilename = entry.getHeader().getLogfileName(); Long logfileoffset = entry.getHeader().getLogfileOffset(); Long logposTimestamp = entry.getHeader().getExecuteTime(); + Long serverId = entry.getHeader().getServerId(); if (CanalEntry.EntryType.TRANSACTIONBEGIN.equals(entry.getEntryType()) || CanalEntry.EntryType.TRANSACTIONEND.equals(entry.getEntryType())) { @@ -715,13 +748,13 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE // position = current + // data.length，代表该事务的下一条offest，避免多余的事务重复 if (CanalEntry.EntryType.TRANSACTIONEND.equals(entry.getEntryType())) { - entryPosition = new EntryPosition(logfilename, logfileoffset, logposTimestamp); + entryPosition = new EntryPosition(logfilename, logfileoffset, logposTimestamp, serverId); logger.debug("set {} to be pending start position before finding another proper one...", entryPosition); logPosition.setPostion(entryPosition); } else if (CanalEntry.EntryType.TRANSACTIONBEGIN.equals(entry.getEntryType())) { // 当前事务开始位点 - entryPosition = new EntryPosition(logfilename, logfileoffset, logposTimestamp); + entryPosition = new EntryPosition(logfilename, logfileoffset, logposTimestamp, serverId); logger.debug("set {} to be pending start position before finding another proper one...", entryPosition); logPosition.setPostion(entryPosition); 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 4f0ada45..538f6b95 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 @@ -35,8 +35,8 @@ import com.alibaba.otter.canal.protocol.CanalEntry.RowData; import com.alibaba.otter.canal.protocol.CanalEntry.TransactionBegin; import com.alibaba.otter.canal.protocol.CanalEntry.TransactionEnd; import com.alibaba.otter.canal.protocol.CanalEntry.Type; +import com.alibaba.otter.canal.protocol.position.EntryPosition; import com.google.protobuf.ByteString; -import com.taobao.tddl.dbsync.binlog.BinlogPosition; import com.taobao.tddl.dbsync.binlog.LogEvent; import com.taobao.tddl.dbsync.binlog.event.DeleteRowsLogEvent; import com.taobao.tddl.dbsync.binlog.event.IntvarLogEvent; @@ -189,8 +189,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar if (!isSeek) { // 使用新的表结构元数据管理方式 - BinlogPosition position = createPosition(event.getHeader()); - tableMetaCache.apply(position, event.getDbName(), queryString); + EntryPosition position = createPosition(event.getHeader()); + tableMetaCache.apply(position, event.getDbName(), queryString, null); } Header header = createHeader(binlogFileName, event.getHeader(), schemaName, tableName, type); @@ -386,7 +386,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar table.getDbName(), table.getTableName(), eventType); - BinlogPosition position = createPosition(event.getHeader()); + EntryPosition position = createPosition(event.getHeader()); RowChange.Builder rowChangeBuider = RowChange.newBuilder(); rowChangeBuider.setTableId(event.getTableId()); rowChangeBuider.setIsDdl(false); @@ -444,8 +444,11 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar } } - private BinlogPosition createPosition(LogHeader logHeader) { - return new BinlogPosition(binlogFileName, logHeader.getLogPos(), logHeader.getServerId(), logHeader.getWhen()); // 记录到秒 + private EntryPosition createPosition(LogHeader logHeader) { + return new EntryPosition(binlogFileName, + logHeader.getLogPos(), + logHeader.getWhen() * 1000L, + logHeader.getServerId()); // 记录到秒 } private boolean parseOneRow(RowData.Builder rowDataBuilder, RowsLogEvent event, RowsLogBuffer buffer, BitSet cols, @@ -468,7 +471,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar } } - BinlogPosition position = createPosition(event.getHeader()); + EntryPosition position = createPosition(event.getHeader()); if (!existRDSNoPrimaryKey) { // online ddl增加字段操作步骤： // 1. 新增一张临时表，将需要做ddl表的数据全量导入 @@ -716,7 +719,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar return true; } - private TableMeta getTableMeta(String dbName, String tbName, boolean useCache, BinlogPosition position) { + private TableMeta getTableMeta(String dbName, String tbName, boolean useCache, EntryPosition position) { try { return tableMetaCache.getTableMeta(dbName, tbName, useCache, position); } catch (Exception e) { diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java index 23229859..ff66c875 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java @@ -14,11 +14,11 @@ import com.alibaba.otter.canal.parse.exception.CanalParseException; import com.alibaba.otter.canal.parse.inbound.TableMeta; import com.alibaba.otter.canal.parse.inbound.TableMeta.FieldMeta; import com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection; -import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaManager; +import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDB; +import com.alibaba.otter.canal.protocol.position.EntryPosition; import com.google.common.cache.CacheBuilder; import com.google.common.cache.CacheLoader; import com.google.common.cache.LoadingCache; -import com.taobao.tddl.dbsync.binlog.BinlogPosition; /** * 处理table meta解析和缓存 @@ -37,15 +37,15 @@ public class TableMetaCache { private MysqlConnection connection; private boolean isOnRDS = false; - private TableMetaManager tableMetaManager; + private TableMetaTSDB tableMetaTSDB; // 第一层tableId,第二层schema.table,解决tableId重复，对应多张表 private LoadingCache tableMetaDB; - public TableMetaCache(MysqlConnection con, TableMetaManager tableMetaManager){ + public TableMetaCache(MysqlConnection con, TableMetaTSDB tableMetaTSDB){ this.connection = con; - this.tableMetaManager = tableMetaManager; + this.tableMetaTSDB = tableMetaTSDB; // 如果持久存储的表结构为空，从db里面获取下 - if (tableMetaManager == null) { + if (tableMetaTSDB == null) { this.tableMetaDB = CacheBuilder.newBuilder().build(new CacheLoader() { @Override @@ -124,14 +124,14 @@ public class TableMetaCache { return tableMetaDB.getUnchecked(getFullName(schema, table)); } - public TableMeta getTableMeta(String schema, String table, BinlogPosition position) { + public TableMeta getTableMeta(String schema, String table, EntryPosition position) { return getTableMeta(schema, table, true, position); } - public TableMeta getTableMeta(String schema, String table, boolean useCache, BinlogPosition position) { + public TableMeta getTableMeta(String schema, String table, boolean useCache, EntryPosition position) { TableMeta tableMeta = null; - if (tableMetaManager != null) { - tableMeta = tableMetaManager.find(schema, table); + if (tableMetaTSDB != null) { + tableMeta = tableMetaTSDB.find(schema, table); if (tableMeta == null) { // 因为条件变化，可能第一次的tableMeta没取到，需要从db获取一次，并记录到snapshot中 String fullName = getFullName(schema, table); @@ -142,8 +142,8 @@ public class TableMetaCache { createDDL = packet.getFieldValues().get(1); } // 强制覆盖掉内存值 - tableMetaManager.apply(position, schema, createDDL); - tableMeta = tableMetaManager.find(schema, table); + tableMetaTSDB.apply(position, schema, createDDL, "first"); + tableMeta = tableMetaTSDB.find(schema, table); } catch (IOException e) { throw new CanalParseException("fetch failed by table meta:" + fullName, e); } @@ -159,7 +159,7 @@ public class TableMetaCache { } public void clearTableMeta(String schema, String table) { - if (tableMetaManager != null) { + if (tableMetaTSDB != null) { // tsdb不需要做,会基于ddl sql自动清理 } else { tableMetaDB.invalidate(getFullName(schema, table)); @@ -167,7 +167,7 @@ public class TableMetaCache { } public void clearTableMetaWithSchemaName(String schema) { - if (tableMetaManager != null) { + if (tableMetaTSDB != null) { // tsdb不需要做,会基于ddl sql自动清理 } else { for (String name : tableMetaDB.asMap().keySet()) { @@ -180,7 +180,7 @@ public class TableMetaCache { } public void clearTableMeta() { - if (tableMetaManager != null) { + if (tableMetaTSDB != null) { // tsdb不需要做,会基于ddl sql自动清理 } else { tableMetaDB.invalidateAll(); @@ -195,9 +195,9 @@ public class TableMetaCache { * @param ddl * @return */ - public boolean apply(BinlogPosition position, String schema, String ddl) { - if (tableMetaManager != null) { - return tableMetaManager.apply(position, schema, ddl); + public boolean apply(EntryPosition position, String schema, String ddl, String extra) { + if (tableMetaTSDB != null) { + return tableMetaTSDB.apply(position, schema, ddl, extra); } else { // ignore return true; diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DataSourceFactoryTSDB.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DataSourceFactoryTSDB.java deleted file mode 100644 index 25f60ee3..00000000 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DataSourceFactoryTSDB.java +++ /dev/null @@ -1,41 +0,0 @@ -package com.alibaba.otter.canal.parse.inbound.mysql.tsdb; - -import javax.sql.DataSource; - -import com.alibaba.druid.pool.DruidDataSource; -import com.alibaba.druid.pool.vendor.MySqlExceptionSorter; -import com.alibaba.druid.pool.vendor.MySqlValidConnectionChecker; -import com.alibaba.otter.canal.parse.exception.CanalParseException; - -/** - * Created by wanshao Date: 2017/9/22 Time: 下午2:46 - **/ -public class DataSourceFactoryTSDB { - - public static DataSource getDataSource(String url, String userName, String password) { - try { - DruidDataSource druidDataSource = new DruidDataSource(); - druidDataSource.setUrl(url); - druidDataSource.setUsername(userName); - druidDataSource.setPassword(password); - druidDataSource.setTestWhileIdle(true); - druidDataSource.setTestOnBorrow(false); - druidDataSource.setTestOnReturn(false); - druidDataSource.setNotFullTimeoutRetryCount(2); - druidDataSource.setValidConnectionCheckerClassName(MySqlValidConnectionChecker.class.getName()); - druidDataSource.setExceptionSorterClassName(MySqlExceptionSorter.class.getName()); - druidDataSource.setValidationQuery("SELECT 1"); - druidDataSource.setInitialSize(1); - druidDataSource.setMinIdle(1); - druidDataSource.setMaxActive(30); - druidDataSource.setMaxWait(10 * 1000); - druidDataSource.setTimeBetweenEvictionRunsMillis(60 * 1000); - druidDataSource.setMinEvictableIdleTimeMillis(50 * 1000); - druidDataSource.setUseUnfairLock(true); - druidDataSource.init(); - return druidDataSource; - } catch (Throwable e) { - throw new CanalParseException("init druidDataSource failed", e); - } - } -} diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java index f3c6aeab..3bd220ef 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java @@ -6,14 +6,15 @@ import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; -import com.taobao.tddl.dbsync.binlog.BinlogPosition; import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.alibaba.druid.sql.ast.SQLDataType; +import com.alibaba.druid.sql.ast.SQLDataTypeImpl; import com.alibaba.druid.sql.ast.SQLExpr; import com.alibaba.druid.sql.ast.SQLStatement; +import com.alibaba.druid.sql.ast.expr.SQLCharExpr; import com.alibaba.druid.sql.ast.expr.SQLIdentifierExpr; import com.alibaba.druid.sql.ast.expr.SQLNullExpr; import com.alibaba.druid.sql.ast.expr.SQLPropertyExpr; @@ -30,11 +31,10 @@ import com.alibaba.druid.sql.repository.Schema; import com.alibaba.druid.sql.repository.SchemaObject; import com.alibaba.druid.sql.repository.SchemaRepository; import com.alibaba.druid.util.JdbcConstants; - import com.alibaba.otter.canal.parse.inbound.TableMeta; import com.alibaba.otter.canal.parse.inbound.TableMeta.FieldMeta; import com.alibaba.otter.canal.parse.inbound.mysql.ddl.DruidDdlParser; -import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDB; +import com.alibaba.otter.canal.protocol.position.EntryPosition; /** * 基于DDL维护的内存表结构 @@ -44,15 +44,19 @@ import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDB; */ public class MemoryTableMeta implements TableMetaTSDB { + private Logger logger = LoggerFactory.getLogger(MemoryTableMeta.class); private Map, TableMeta> tableMetas = new ConcurrentHashMap, TableMeta>(); private SchemaRepository repository = new SchemaRepository(JdbcConstants.MYSQL); - private Logger logger = LoggerFactory.getLogger(MemoryTableMeta.class); - public MemoryTableMeta(Logger logger){ - this.logger = logger; + public MemoryTableMeta(){ } - public boolean apply(BinlogPosition position, String schema, String ddl) { + @Override + public boolean init(String destination) { + return true; + } + + public boolean apply(EntryPosition position, String schema, String ddl, String extra) { tableMetas.clear(); synchronized (this) { if (StringUtils.isNotEmpty(schema)) { @@ -116,7 +120,7 @@ public class MemoryTableMeta implements TableMetaTSDB { } @Override - public boolean rollback(BinlogPosition position) { + public boolean rollback(EntryPosition position) { throw new RuntimeException("not support for memory"); } @@ -169,6 +173,17 @@ public class MemoryTableMeta implements TableMetaTSDB { dataTypStr += ")"; } + if (dataType instanceof SQLDataTypeImpl) { + SQLDataTypeImpl dataTypeImpl = (SQLDataTypeImpl) dataType; + if (dataTypeImpl.isUnsigned()) { + dataTypStr += " unsigned"; + } + + if (dataTypeImpl.isZerofill()) { + dataTypStr += " zerofill"; + } + } + if (column.getDefaultExpr() == null || column.getDefaultExpr() instanceof SQLNullExpr) { fieldMeta.setDefaultValue(null); } else { @@ -211,6 +226,8 @@ public class MemoryTableMeta implements TableMetaTSDB { + DruidDdlParser.unescapeName(((SQLPropertyExpr) sqlName).getName()); } else if (sqlName instanceof SQLIdentifierExpr) { return DruidDdlParser.unescapeName(((SQLIdentifierExpr) sqlName).getName()); + } else if (sqlName instanceof SQLCharExpr) { + return ((SQLCharExpr) sqlName).getText(); } else { return sqlName.toString(); } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaManager.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaManager.java index 9d926548..c8fe89e1 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaManager.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaManager.java @@ -11,13 +11,12 @@ import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import java.util.regex.Pattern; -import javax.annotation.Resource; - import org.apache.commons.beanutils.BeanUtils; import org.apache.commons.lang.ObjectUtils; import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.slf4j.MDC; import com.alibaba.druid.sql.repository.Schema; import com.alibaba.fastjson.JSON; @@ -32,10 +31,10 @@ import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.TableMetaCache; import com.alibaba.otter.canal.parse.inbound.mysql.ddl.DdlResult; import com.alibaba.otter.canal.parse.inbound.mysql.ddl.DruidDdlParser; import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryDAO; +import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryDO; import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaSnapshotDAO; -import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.model.MetaHistoryDO; -import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.model.MetaSnapshotDO; -import com.taobao.tddl.dbsync.binlog.BinlogPosition; +import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaSnapshotDO; +import com.alibaba.otter.canal.protocol.position.EntryPosition; /** * 基于console远程管理 see internal class: CanalTableMeta , ConsoleTableMetaTSDB @@ -45,23 +44,27 @@ import com.taobao.tddl.dbsync.binlog.BinlogPosition; */ public class TableMetaManager implements TableMetaTSDB { - private static Logger logger = LoggerFactory.getLogger(TableMetaManager.class); - private static Pattern pattern = Pattern.compile("Duplicate entry '.*' for key '*'"); - private static final BinlogPosition INIT_POSITION = BinlogPosition.parseFromString("0:0#-2.-1"); - private MemoryTableMeta memoryTableMeta; - private MysqlConnection connection; // 查询meta信息的链接 - private CanalEventFilter filter; - private BinlogPosition lastPosition; - private ScheduledExecutorService scheduler; + private static Logger logger = LoggerFactory.getLogger(TableMetaManager.class); + private static Pattern pattern = Pattern.compile("Duplicate entry '.*' for key '*'"); + private static final EntryPosition INIT_POSITION = new EntryPosition("0", 0L, -2L, -1L); + 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; - @Resource - private MetaHistoryDAO metaHistoryDAO; + public TableMetaManager(){ - @Resource - private MetaSnapshotDAO metaSnapshotDAO; + } - public void init() { - this.memoryTableMeta = new MemoryTableMeta(logger); + @Override + public boolean init(final String destination) { + this.destination = destination; + this.memoryTableMeta = new MemoryTableMeta(); this.scheduler = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() { @Override @@ -76,17 +79,14 @@ public class TableMetaManager implements TableMetaTSDB { @Override public void run() { try { - logger.info("-------- begin to produce snapshot for table meta"); + MDC.put("destination", destination); applySnapshotToDB(lastPosition, false); } catch (Throwable e) { - logger.error("scheudle faield", e); + logger.error("scheudle applySnapshotToDB faield", e); } } - }, 24, 24, TimeUnit.SECONDS); - } - - public TableMetaManager(){ - + }, 24, 24, TimeUnit.HOURS); + return true; } @Override @@ -97,13 +97,13 @@ public class TableMetaManager implements TableMetaTSDB { } @Override - public boolean apply(BinlogPosition position, String schema, String ddl) { + public boolean apply(EntryPosition position, String schema, String ddl, String extra) { // 首先记录到内存结构 synchronized (memoryTableMeta) { - if (memoryTableMeta.apply(position, schema, ddl)) { + if (memoryTableMeta.apply(position, schema, ddl, extra)) { this.lastPosition = position; // 同步每次变更给远程做历史记录 - return applyHistoryToDB(position, schema, ddl); + return applyHistoryToDB(position, schema, ddl, extra); } else { throw new RuntimeException("apply to memory is failed"); } @@ -111,11 +111,11 @@ public class TableMetaManager implements TableMetaTSDB { } @Override - public boolean rollback(BinlogPosition position) { + public boolean rollback(EntryPosition position) { // 每次rollback需要重新构建一次memory data - this.memoryTableMeta = new MemoryTableMeta(logger); + this.memoryTableMeta = new MemoryTableMeta(); boolean flag = false; - BinlogPosition snapshotPosition = buildMemFromSnapshot(position); + EntryPosition snapshotPosition = buildMemFromSnapshot(position); if (snapshotPosition != null) { applyHistoryOnMemory(snapshotPosition, position); flag = true; @@ -145,20 +145,25 @@ public class TableMetaManager implements TableMetaTSDB { ResultSetPacket packet = connection.query("show databases"); List schemas = new ArrayList(); for (String schema : packet.getFieldValues()) { - if (!filter.filter(schema)) { - schemas.add(schema); - } + schemas.add(schema); } for (String schema : schemas) { packet = connection.query("show tables from `" + schema + "`"); List tables = new ArrayList(); for (String table : packet.getFieldValues()) { - if (!filter.filter(table)) { - tables.add(table); + String fullName = schema + "." + table; + if (blackFilter == null || !blackFilter.filter(fullName)) { + if (filter == null || filter.filter(fullName)) { + tables.add(table); + } } } + if (tables.isEmpty()) { + continue; + } + StringBuilder sql = new StringBuilder(); for (String table : tables) { sql.append("show create table `" + schema + "`.`" + table + "`;"); @@ -168,7 +173,7 @@ public class TableMetaManager implements TableMetaTSDB { for (ResultSetPacket onePacket : packets) { if (onePacket.getFieldValues().size() > 1) { String oneTableCreateSql = onePacket.getFieldValues().get(1); - memoryTableMeta.apply(INIT_POSITION, schema, oneTableCreateSql); + memoryTableMeta.apply(INIT_POSITION, schema, oneTableCreateSql, null); } } } @@ -179,25 +184,26 @@ public class TableMetaManager implements TableMetaTSDB { } } - private boolean applyHistoryToDB(BinlogPosition position, String schema, String ddl) { + private boolean applyHistoryToDB(EntryPosition position, String schema, String ddl, String extra) { Map content = new HashMap(); - content.put("binlogFile", position.getFileName()); + content.put("destination", destination); + content.put("binlogFile", position.getJournalName()); content.put("binlogOffest", String.valueOf(position.getPosition())); - content.put("binlogMasterId", String.valueOf(position.getMasterId())); + content.put("binlogMasterId", String.valueOf(position.getServerId())); content.put("binlogTimestamp", String.valueOf(position.getTimestamp())); content.put("useSchema", schema); if (content.isEmpty()) { throw new RuntimeException("apply failed caused by content is empty in applyHistoryToDB"); } // 待补充 - List ddlResults = DruidDdlParser.parse(schema, ddl); + List ddlResults = DruidDdlParser.parse(ddl, schema); if (ddlResults.size() > 0) { DdlResult ddlResult = ddlResults.get(0); - content.put("schema", ddlResult.getSchemaName()); - content.put("table", ddlResult.getTableName()); - content.put("type", ddlResult.getType().name()); - content.put("sql", ddl); - // content.put("extra", ""); + content.put("sqlSchema", ddlResult.getSchemaName()); + content.put("sqlTable", ddlResult.getTableName()); + content.put("sqlType", ddlResult.getType().name()); + content.put("sqlText", ddl); + content.put("extra", extra); } MetaHistoryDO metaDO = new MetaHistoryDO(); @@ -212,7 +218,7 @@ public class TableMetaManager implements TableMetaTSDB { // 忽略掉重复的位点 logger.warn("dup apply for sql : " + ddl); } else { - throw new CanalParseException("apply history to db failed caused by : " + e.getMessage()); + throw new CanalParseException("apply history to db failed caused by : " + e.getMessage(), e); } } @@ -222,9 +228,9 @@ public class TableMetaManager implements TableMetaTSDB { /** * 发布数据到console上 */ - private boolean applySnapshotToDB(BinlogPosition position, boolean init) { + private boolean applySnapshotToDB(EntryPosition position, boolean init) { // 获取一份快照 - MemoryTableMeta tmpMemoryTableMeta = new MemoryTableMeta(logger); + MemoryTableMeta tmpMemoryTableMeta = new MemoryTableMeta(); Map schemaDdls = null; synchronized (memoryTableMeta) { if (!init && position == null) { @@ -233,7 +239,7 @@ public class TableMetaManager implements TableMetaTSDB { } schemaDdls = memoryTableMeta.snapshot(); for (Map.Entry entry : schemaDdls.entrySet()) { - tmpMemoryTableMeta.apply(position, entry.getKey(), entry.getValue()); + tmpMemoryTableMeta.apply(position, entry.getKey(), entry.getValue(), null); } } @@ -248,9 +254,10 @@ public class TableMetaManager implements TableMetaTSDB { } if (compareAll) { Map content = new HashMap(); - content.put("binlogFile", position.getFileName()); + content.put("destination", destination); + content.put("binlogFile", position.getJournalName()); content.put("binlogOffest", String.valueOf(position.getPosition())); - content.put("binlogMasterId", String.valueOf(position.getMasterId())); + content.put("binlogMasterId", String.valueOf(position.getServerId())); content.put("binlogTimestamp", String.valueOf(position.getTimestamp())); content.put("data", JSON.toJSONString(schemaDdls)); if (content.isEmpty()) { @@ -264,9 +271,9 @@ public class TableMetaManager implements TableMetaTSDB { } catch (Throwable e) { if (isUkDuplicateException(e)) { // 忽略掉重复的位点 - logger.warn("dup apply snapshot for data : " + snapshotDO.getData()); + logger.info("dup apply snapshot use position : " + position + " , just ignore"); } else { - throw new CanalParseException("apply failed caused by : " + e.getMessage()); + throw new CanalParseException("apply failed caused by : " + e.getMessage(), e); } } return true; @@ -296,18 +303,21 @@ public class TableMetaManager implements TableMetaTSDB { return result; } - private BinlogPosition buildMemFromSnapshot(BinlogPosition position) { + private EntryPosition buildMemFromSnapshot(EntryPosition position) { try { - MetaSnapshotDO snapshotDO = metaSnapshotDAO.findByTimestamp(position.getTimestamp()); + MetaSnapshotDO snapshotDO = metaSnapshotDAO.findByTimestamp(destination, position.getTimestamp()); + if (snapshotDO == null) { + return null; + } String binlogFile = snapshotDO.getBinlogFile(); Long binlogOffest = snapshotDO.getBinlogOffest(); String binlogMasterId = snapshotDO.getBinlogMasterId(); Long binlogTimestamp = snapshotDO.getBinlogTimestamp(); - BinlogPosition snapshotPosition = new BinlogPosition(binlogFile, + EntryPosition snapshotPosition = new EntryPosition(binlogFile, binlogOffest == null ? 0l : binlogOffest, - Long.valueOf(binlogMasterId == null ? "-2" : binlogMasterId), - binlogTimestamp == null ? 0l : binlogTimestamp); + binlogTimestamp == null ? 0l : binlogTimestamp, + Long.valueOf(binlogMasterId == null ? "-2" : binlogMasterId)); // data存储为Map，每个分库一套建表 String sqlData = snapshotDO.getData(); JSONObject jsonObj = JSON.parseObject(sqlData); @@ -315,43 +325,49 @@ public class TableMetaManager implements TableMetaTSDB { // 记录到内存 if (!memoryTableMeta.apply(snapshotPosition, ObjectUtils.toString(entry.getKey()), - ObjectUtils.toString(entry.getValue()))) { + ObjectUtils.toString(entry.getValue()), + null)) { return null; } } return snapshotPosition; } catch (Throwable e) { - throw new CanalParseException("apply failed caused by : " + e.getMessage()); + throw new CanalParseException("apply failed caused by : " + e.getMessage(), e); } } - private boolean applyHistoryOnMemory(BinlogPosition position, BinlogPosition rollbackPosition) { + private boolean applyHistoryOnMemory(EntryPosition position, EntryPosition rollbackPosition) { try { - List metaHistoryDOList = metaHistoryDAO.findByTimestamp(position.getTimestamp(), + List metaHistoryDOList = metaHistoryDAO.findByTimestamp(destination, + position.getTimestamp(), rollbackPosition.getTimestamp()); + if (metaHistoryDOList == null) { + return true; + } + for (MetaHistoryDO metaHistoryDO : metaHistoryDOList) { String binlogFile = metaHistoryDO.getBinlogFile(); Long binlogOffest = metaHistoryDO.getBinlogOffest(); String binlogMasterId = metaHistoryDO.getBinlogMasterId(); Long binlogTimestamp = metaHistoryDO.getBinlogTimestamp(); String useSchema = metaHistoryDO.getUseSchema(); - String sqlData = metaHistoryDO.getSql(); - BinlogPosition snapshotPosition = new BinlogPosition(binlogFile, + String sqlData = metaHistoryDO.getSqlText(); + EntryPosition snapshotPosition = new EntryPosition(binlogFile, binlogOffest == null ? 0L : binlogOffest, - Long.valueOf(binlogMasterId == null ? "-2" : binlogMasterId), - binlogTimestamp == null ? 0L : binlogTimestamp); + binlogTimestamp == null ? 0L : binlogTimestamp, + Long.valueOf(binlogMasterId == null ? "-2" : binlogMasterId)); // 如果是同一秒内,对比一下history的位点，如果比期望的位点要大，忽略之 if (snapshotPosition.getTimestamp() > rollbackPosition.getTimestamp()) { continue; - } else if (rollbackPosition.getMasterId() == snapshotPosition.getMasterId() + } else if (rollbackPosition.getServerId() == snapshotPosition.getServerId() && snapshotPosition.compareTo(rollbackPosition) > 0) { continue; } // 记录到内存 - if (!memoryTableMeta.apply(snapshotPosition, useSchema, sqlData)) { + if (!memoryTableMeta.apply(snapshotPosition, useSchema, sqlData, null)) { return false; } @@ -441,6 +457,10 @@ public class TableMetaManager implements TableMetaTSDB { this.metaSnapshotDAO = metaSnapshotDAO; } + public void setBlackFilter(CanalEventFilter blackFilter) { + this.blackFilter = blackFilter; + } + public MysqlConnection getConnection() { return connection; } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaTSDB.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaTSDB.java index 44cd9959..d712adbe 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaTSDB.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaTSDB.java @@ -3,17 +3,21 @@ package com.alibaba.otter.canal.parse.inbound.mysql.tsdb; import java.util.Map; import com.alibaba.otter.canal.parse.inbound.TableMeta; - -import com.taobao.tddl.dbsync.binlog.BinlogPosition; +import com.alibaba.otter.canal.protocol.position.EntryPosition; /** * 表结构的时间序列存储 * * @author agapple 2017年7月27日 下午4:06:30 - * @since 3.2.5 + * @since 1.0.25 */ public interface TableMetaTSDB { + /** + * 初始化 + */ + public boolean init(String destination); + /** * 获取当前的表结构 */ @@ -22,12 +26,12 @@ public interface TableMetaTSDB { /** * 添加ddl到时间序列库中 */ - public boolean apply(BinlogPosition position, String schema, String ddl); + public boolean apply(EntryPosition position, String schema, String ddl, String extra); /** * 回滚到指定位点的表结构 */ - public boolean rollback(BinlogPosition position); + public boolean rollback(EntryPosition position); /** * 生成快照内容 diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaTSDBBuilder.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaTSDBBuilder.java new file mode 100644 index 00000000..8e37ef71 --- /dev/null +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaTSDBBuilder.java @@ -0,0 +1,51 @@ +package com.alibaba.otter.canal.parse.inbound.mysql.tsdb; + +import java.util.concurrent.ConcurrentMap; + +import org.apache.commons.lang.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.context.support.ClassPathXmlApplicationContext; + +import com.google.common.collect.Maps; + +/** + * @author agapple 2017年10月11日 下午8:45:40 + * @since 1.0.25 + */ +public class TableMetaTSDBBuilder { + + protected final static Logger logger = LoggerFactory.getLogger(TableMetaTSDBBuilder.class); + private static ConcurrentMap contexts = Maps.newConcurrentMap(); + + /** + * 代理一下tableMetaTSDB的获取,使用隔离的spring定义 + */ + public static TableMetaTSDB build(String destination, String springXml) { + if (StringUtils.isNotEmpty(springXml)) { + ClassPathXmlApplicationContext applicationContext = contexts.get(destination); + if (applicationContext == null) { + synchronized (contexts) { + if (applicationContext == null) { + applicationContext = new ClassPathXmlApplicationContext(springXml); + contexts.put(destination, applicationContext); + } + } + } + TableMetaTSDB tableMetaTSDB = (TableMetaTSDB) applicationContext.getBean("tableMetaTSDB"); + tableMetaTSDB.init(destination); + logger.info("{} init TableMetaTSDB with {}", destination, springXml); + return tableMetaTSDB; + } else { + return null; + } + } + + public static void destory(String destination) { + ClassPathXmlApplicationContext context = contexts.remove(destination); + if (context != null) { + logger.info("{} destory TableMetaTSDB", destination); + context.close(); + } + } +} diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaHistoryDAO.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaHistoryDAO.java index 46ffafbf..e43c5e62 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaHistoryDAO.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaHistoryDAO.java @@ -1,11 +1,19 @@ package com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao; +import java.io.InputStream; +import java.sql.Connection; +import java.sql.Statement; +import java.text.SimpleDateFormat; +import java.util.Date; import java.util.HashMap; import java.util.List; +import javax.sql.DataSource; + +import org.apache.commons.io.IOUtils; +import org.apache.commons.lang.StringUtils; import org.springframework.orm.ibatis.support.SqlMapClientDaoSupport; -import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.model.MetaHistoryDO; import com.google.common.collect.Maps; /** @@ -14,22 +22,25 @@ import com.google.common.collect.Maps; * @author wanshao 2017年7月27日 下午10:51:55 * @since 3.2.5 */ +@SuppressWarnings("deprecation") public class MetaHistoryDAO extends SqlMapClientDaoSupport { - public List getAll() { - return getSqlMapClientTemplate().queryForList("table_meta_history.getAll"); - } - public Long insert(MetaHistoryDO metaDO) { - return (Long) getSqlMapClientTemplate().insert("table_meta_history.insert", metaDO); + return (Long) getSqlMapClientTemplate().insert("meta_history.insert", metaDO); } - public List findByTimestamp(long snapshotTimestamp, long timestamp) { + public List findByTimestamp(String destination, Long snapshotTimestamp, Long timestamp) { HashMap params = Maps.newHashMapWithExpectedSize(2); - params.put("snapshotTimestamp", snapshotTimestamp); - params.put("timestamp", timestamp); - return (List) getSqlMapClientTemplate().queryForList("table_meta_history.findByTimestamp", - params); + params.put("destination", destination); + params.put("snapshotTimestamp", snapshotTimestamp == null ? 0L : snapshotTimestamp); + params.put("timestamp", timestamp == null ? 0L : timestamp); + return (List) getSqlMapClientTemplate().queryForList("meta_history.findByTimestamp", params); + } + + public Integer deleteByName(String destination) { + HashMap params = Maps.newHashMapWithExpectedSize(2); + params.put("destination", destination); + return getSqlMapClientTemplate().delete("meta_history.deleteByName", params); } /** @@ -37,7 +48,35 @@ public class MetaHistoryDAO extends SqlMapClientDaoSupport { */ public Integer deleteByGmtModified(int interval) { HashMap params = Maps.newHashMapWithExpectedSize(2); - params.put("interval", interval); - return getSqlMapClientTemplate().delete("table_meta_history.deleteByGmtModified", params); + long timestamp = System.currentTimeMillis() - interval * 1000; + Date date = new Date(timestamp); + SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); + params.put("timestamp", format.format(date)); + return getSqlMapClientTemplate().delete("meta_history.deleteByGmtModified", params); + } + + protected void initDao() throws Exception { + Connection conn = null; + InputStream input = null; + try { + DataSource dataSource = getDataSource(); + conn = dataSource.getConnection(); + input = Thread.currentThread().getContextClassLoader().getResourceAsStream("ddl/mysql/meta_history.sql"); + if (input == null) { + return; + } + + String sql = StringUtils.join(IOUtils.readLines(input), "\n"); + Statement stmt = conn.createStatement(); + stmt.execute(sql); + stmt.close(); + } catch (Throwable e) { + logger.warn("init meta_history failed", e); + } finally { + IOUtils.closeQuietly(input); + if (conn != null) { + conn.close(); + } + } } } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/model/MetaHistoryDO.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaHistoryDO.java similarity index 55% rename from parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/model/MetaHistoryDO.java rename to parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaHistoryDO.java index 96108515..6ac548f1 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/model/MetaHistoryDO.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaHistoryDO.java @@ -1,37 +1,26 @@ -package com.alibaba.otter.canal.parse.inbound.mysql.tsdb.model; +package com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao; import java.util.Date; /** * @author agapple 2017年7月27日 下午11:09:41 - * @since 3.2.5 + * @since 1.0.25 */ public class MetaHistoryDO { - /** - * 主键 - */ private Long id; - - /** - * 创建时间 - */ private Date gmtCreate; - - /** - * 修改时间 - */ private Date gmtModified; - + private String destination; private String binlogFile; private Long binlogOffest; private String binlogMasterId; private Long binlogTimestamp; private String useSchema; - private String schema; - private String table; - private String sql; - private String type; + private String sqlSchema; + private String sqlTable; + private String sqlText; + private String sqlType; private String extra; public Long getId() { @@ -98,37 +87,6 @@ public class MetaHistoryDO { this.useSchema = useSchema; } - public String getSchema() { - return schema; - } - - public void setSchema(String schema) { - this.schema = schema; - } - - public String getTable() { - return table; - } - - public void setTable(String table) { - this.table = table; - } - - public String getSql() { - return sql; - } - - public void setSql(String sql) { - this.sql = sql; - } - - public String getType() { - return type; - } - - public void setType(String type) { - this.type = type; - } public String getExtra() { return extra; @@ -138,4 +96,53 @@ public class MetaHistoryDO { this.extra = extra; } + public String getDestination() { + return destination; + } + + public void setDestination(String destination) { + this.destination = destination; + } + + public String getSqlSchema() { + return sqlSchema; + } + + public void setSqlSchema(String sqlSchema) { + this.sqlSchema = sqlSchema; + } + + public String getSqlTable() { + return sqlTable; + } + + public void setSqlTable(String sqlTable) { + this.sqlTable = sqlTable; + } + + public String getSqlText() { + return sqlText; + } + + public void setSqlText(String sqlText) { + this.sqlText = sqlText; + } + + public String getSqlType() { + return sqlType; + } + + public void setSqlType(String sqlType) { + this.sqlType = sqlType; + } + + @Override + public String toString() { + return "MetaHistoryDO [id=" + id + ", gmtCreate=" + gmtCreate + ", gmtModified=" + gmtModified + + ", destination=" + destination + ", binlogFile=" + binlogFile + ", binlogOffest=" + binlogOffest + + ", binlogMasterId=" + binlogMasterId + ", binlogTimestamp=" + binlogTimestamp + ", useSchema=" + + useSchema + ", sqlSchema=" + sqlSchema + ", sqlTable=" + sqlTable + ", sqlText=" + sqlText + + ", sqlType=" + sqlType + ", extra=" + extra + "]"; + } + } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaSnapshotDAO.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaSnapshotDAO.java index ca32ebd3..1a3f3053 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaSnapshotDAO.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaSnapshotDAO.java @@ -1,10 +1,18 @@ package com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao; +import java.io.InputStream; +import java.sql.Connection; +import java.sql.Statement; +import java.text.SimpleDateFormat; +import java.util.Date; import java.util.HashMap; +import javax.sql.DataSource; + +import org.apache.commons.io.IOUtils; +import org.apache.commons.lang.StringUtils; import org.springframework.orm.ibatis.support.SqlMapClientDaoSupport; -import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.model.MetaSnapshotDO; import com.google.common.collect.Maps; /** @@ -13,27 +21,29 @@ import com.google.common.collect.Maps; * @author wanshao 2017年7月27日 下午10:51:55 * @since 3.2.5 */ - +@SuppressWarnings("deprecation") public class MetaSnapshotDAO extends SqlMapClientDaoSupport { public Long insert(MetaSnapshotDO snapshotDO) { - return (Long) getSqlMapClientTemplate().insert("table_meta_snapshot.insert", snapshotDO); + return (Long) getSqlMapClientTemplate().insert("meta_snapshot.insert", snapshotDO); } public Long update(MetaSnapshotDO snapshotDO) { - return (Long) getSqlMapClientTemplate().insert("table_meta_snapshot.update", snapshotDO); + return (Long) getSqlMapClientTemplate().insert("meta_snapshot.update", snapshotDO); } - public MetaSnapshotDO findByTimestamp(long timestamp) { + public MetaSnapshotDO findByTimestamp(String destination, Long timestamp) { HashMap params = Maps.newHashMapWithExpectedSize(2); - params.put("timestamp", timestamp); - return (MetaSnapshotDO) getSqlMapClientTemplate().queryForObject("table_meta_snapshot.findByTimestamp", params); + params.put("timestamp", timestamp == null ? 0L : timestamp); + params.put("destination", destination); + + return (MetaSnapshotDO) getSqlMapClientTemplate().queryForObject("meta_snapshot.findByTimestamp", params); } - public Integer deleteByTask(String taskName) { + public Integer deleteByName(String destination) { HashMap params = Maps.newHashMapWithExpectedSize(2); - params.put("taskName", taskName); - return getSqlMapClientTemplate().delete("table_meta_snapshot.deleteByTaskName", params); + params.put("destination", destination); + return getSqlMapClientTemplate().delete("meta_snapshot.deleteByName", params); } /** @@ -41,8 +51,35 @@ public class MetaSnapshotDAO extends SqlMapClientDaoSupport { */ public Integer deleteByGmtModified(int interval) { HashMap params = Maps.newHashMapWithExpectedSize(2); - params.put("interval", interval); - return getSqlMapClientTemplate().delete("table_meta_snapshot.deleteByGmtModified", params); + long timestamp = System.currentTimeMillis() - interval * 1000; + Date date = new Date(timestamp); + SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); + params.put("timestamp", format.format(date)); + return getSqlMapClientTemplate().delete("meta_snapshot.deleteByGmtModified", params); } + protected void initDao() throws Exception { + Connection conn = null; + InputStream input = null; + try { + DataSource dataSource = getDataSource(); + conn = dataSource.getConnection(); + input = Thread.currentThread().getContextClassLoader().getResourceAsStream("ddl/mysql/meta_snapshot.sql"); + if (input == null) { + return; + } + + String sql = StringUtils.join(IOUtils.readLines(input), "\n"); + Statement stmt = conn.createStatement(); + stmt.execute(sql); + stmt.close(); + } catch (Throwable e) { + logger.warn("init meta_history failed", e); + } finally { + IOUtils.closeQuietly(input); + if (conn != null) { + conn.close(); + } + } + } } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/model/MetaSnapshotDO.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaSnapshotDO.java similarity index 72% rename from parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/model/MetaSnapshotDO.java rename to parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaSnapshotDO.java index 8ddfbc64..693c5f8e 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/model/MetaSnapshotDO.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaSnapshotDO.java @@ -1,28 +1,17 @@ -package com.alibaba.otter.canal.parse.inbound.mysql.tsdb.model; +package com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao; import java.util.Date; /** * @author agapple 2017年7月27日 下午11:09:41 - * @since 3.2.5 + * @since 1.0.25 */ public class MetaSnapshotDO { - /** - * 主键 - */ private Long id; - - /** - * 创建时间 - */ private Date gmtCreate; - - /** - * 修改时间 - */ private Date gmtModified; - + private String destination; private String binlogFile; private Long binlogOffest; private String binlogMasterId; @@ -102,4 +91,20 @@ public class MetaSnapshotDO { this.extra = extra; } + public String getDestination() { + return destination; + } + + public void setDestination(String destination) { + this.destination = destination; + } + + @Override + public String toString() { + return "MetaSnapshotDO [id=" + id + ", gmtCreate=" + gmtCreate + ", gmtModified=" + gmtModified + + ", destination=" + destination + ", binlogFile=" + binlogFile + ", binlogOffest=" + binlogOffest + + ", binlogMasterId=" + binlogMasterId + ", binlogTimestamp=" + binlogTimestamp + ", data=" + data + + ", extra=" + extra + "]"; + } + } diff --git a/parse/src/main/resources/ddl/derby/meta_history.sql b/parse/src/main/resources/ddl/derby/meta_history.sql new file mode 100644 index 00000000..0529e1fa --- /dev/null +++ b/parse/src/main/resources/ddl/derby/meta_history.sql @@ -0,0 +1,22 @@ +CREATE TABLE meta_history ( + id bigint GENERATED ALWAYS AS IDENTITY NOT NULL, + gmt_create timestamp NOT NULL, + gmt_modified timestamp NOT NULL, + destination varchar(128) DEFAULT NULL, + binlog_file varchar(64) DEFAULT NULL, + binlog_offest bigint DEFAULT NULL, + binlog_master_id varchar(64) DEFAULT NULL, + binlog_timestamp bigint DEFAULT NULL, + use_schema varchar(1024) DEFAULT NULL, + sql_schema varchar(1024) DEFAULT NULL, + sql_table varchar(1024) DEFAULT NULL, + sql_text clob(16 M) DEFAULT NULL, + sql_type varchar(1024) DEFAULT NULL, + extra varchar(512) DEFAULT NULL, + PRIMARY KEY (id), + CONSTRAINT meta_history_binlog_file_offest UNIQUE (destination,binlog_master_id,binlog_file,binlog_offest) +); + +create index meta_history_destination on meta_history(destination); +create index meta_history_destination_timestamp on meta_history(destination,binlog_timestamp); +create index meta_history_gmt_modified on meta_history(gmt_modified); \ No newline at end of file diff --git a/parse/src/main/resources/ddl/derby/meta_snapshot.sql b/parse/src/main/resources/ddl/derby/meta_snapshot.sql new file mode 100644 index 00000000..61e6cd53 --- /dev/null +++ b/parse/src/main/resources/ddl/derby/meta_snapshot.sql @@ -0,0 +1,18 @@ +CREATE TABLE meta_snapshot ( + id bigint GENERATED ALWAYS AS IDENTITY NOT NULL, + gmt_create timestamp NOT NULL, + gmt_modified timestamp NOT NULL, + destination varchar(128) DEFAULT NULL, + binlog_file varchar(64) DEFAULT NULL, + binlog_offest bigint DEFAULT NULL, + binlog_master_id varchar(64) DEFAULT NULL, + binlog_timestamp bigint DEFAULT NULL, + data clob(16 M) DEFAULT NULL, + extra varchar(512) DEFAULT NULL, + PRIMARY KEY (id), + CONSTRAINT meta_snapshot_binlog_file_offest UNIQUE (destination,binlog_master_id,binlog_file,binlog_offest) +); + +create index meta_snapshot_destination on meta_snapshot(destination); +create index meta_snapshot_destination_timestamp on meta_snapshot(destination,binlog_timestamp); +create index meta_snapshot_gmt_modified on meta_snapshot(gmt_modified); \ No newline at end of file diff --git a/parse/src/main/resources/ddl/mysql/meta_history.sql b/parse/src/main/resources/ddl/mysql/meta_history.sql new file mode 100644 index 00000000..36425961 --- /dev/null +++ b/parse/src/main/resources/ddl/mysql/meta_history.sql @@ -0,0 +1,21 @@ +CREATE TABLE IF NOT EXISTS `meta_history` ( + `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键', + `gmt_create` datetime NOT NULL COMMENT '创建时间', + `gmt_modified` datetime NOT NULL COMMENT '修改时间', + `destination` varchar(128) DEFAULT NULL COMMENT '通道名称', + `binlog_file` varchar(64) DEFAULT NULL COMMENT 'binlog文件名', + `binlog_offest` bigint(20) DEFAULT NULL COMMENT 'binlog偏移量', + `binlog_master_id` varchar(64) DEFAULT NULL COMMENT 'binlog节点id', + `binlog_timestamp` bigint(20) DEFAULT NULL COMMENT 'binlog应用的时间戳', + `use_schema` varchar(1024) DEFAULT NULL COMMENT '执行sql时对应的schema', + `sql_schema` varchar(1024) DEFAULT NULL COMMENT '对应的schema', + `sql_table` varchar(1024) DEFAULT NULL COMMENT '对应的table', + `sql_text` longtext DEFAULT NULL COMMENT '执行的sql', + `sql_type` varchar(256) DEFAULT NULL COMMENT 'sql类型', + `extra` text DEFAULT NULL COMMENT '额外的扩展信息', + PRIMARY KEY (`id`), + UNIQUE KEY binlog_file_offest(`destination`,`binlog_master_id`,`binlog_file`,`binlog_offest`), + KEY `destination` (`destination`), + KEY `destination_timestamp` (`destination`,`binlog_timestamp`), + KEY `gmt_modified` (`gmt_modified`) +) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='表结构变化明细表'; \ No newline at end of file diff --git a/parse/src/main/resources/ddl/mysql/meta_snapshot.sql b/parse/src/main/resources/ddl/mysql/meta_snapshot.sql new file mode 100644 index 00000000..43313cb6 --- /dev/null +++ b/parse/src/main/resources/ddl/mysql/meta_snapshot.sql @@ -0,0 +1,17 @@ +CREATE TABLE IF NOT EXISTS `meta_snapshot` ( + `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键', + `gmt_create` datetime NOT NULL COMMENT '创建时间', + `gmt_modified` datetime NOT NULL COMMENT '修改时间', + `destination` varchar(128) DEFAULT NULL COMMENT '通道名称', + `binlog_file` varchar(64) DEFAULT NULL COMMENT 'binlog文件名', + `binlog_offest` bigint(20) DEFAULT NULL COMMENT 'binlog偏移量', + `binlog_master_id` varchar(64) DEFAULT NULL COMMENT 'binlog节点id', + `binlog_timestamp` bigint(20) DEFAULT NULL COMMENT 'binlog应用的时间戳', + `data` longtext DEFAULT NULL COMMENT '表结构数据', + `extra` text DEFAULT NULL COMMENT '额外的扩展信息', + PRIMARY KEY (`id`), + UNIQUE KEY binlog_file_offest(`destination`,`binlog_master_id`,`binlog_file`,`binlog_offest`), + KEY `destination` (`destination`), + KEY `destination_timestamp` (`destination`,`binlog_timestamp`), + KEY `gmt_modified` (`gmt_modified`) +) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COMMENT='表结构记录表快照表'; \ No newline at end of file diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParserTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParserTest.java index 7b996ba2..ba8854a7 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParserTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParserTest.java @@ -31,7 +31,7 @@ public class LocalBinlogEventParserTest { public void setUp() { URL url = Thread.currentThread().getContextClassLoader().getResource("dummy.txt"); File dummyFile = new File(url.getFile()); - directory = new File("/Users/wanshao/projects/canal/parse/src/test/resources/binlog/tsdb").getPath(); + directory = new File(dummyFile + "/binlog/tsdb").getPath(); } @Test diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlDumpTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlDumpTest.java index d8915e64..aef55822 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlDumpTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlDumpTest.java @@ -7,6 +7,7 @@ import java.util.List; import org.junit.Assert; import org.junit.Test; +import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter; import com.alibaba.otter.canal.parse.exception.CanalParseException; import com.alibaba.otter.canal.parse.index.AbstractLogPositionManager; import com.alibaba.otter.canal.parse.stub.AbstractCanalEventSinkTest; @@ -26,13 +27,18 @@ public class MysqlDumpTest { @Test public void testSimple() { final MysqlEventParser controller = new MysqlEventParser(); - final EntryPosition startPosition = new EntryPosition("mysql-bin.000003", 123L); + final EntryPosition startPosition = new EntryPosition("mysql-bin.000001", 104606L); controller.setConnectionCharset(Charset.forName("UTF-8")); controller.setSlaveId(3344L); controller.setDetectingEnable(false); controller.setMasterInfo(new AuthenticationInfo(new InetSocketAddress("127.0.0.1", 3306), "canal", "canal")); controller.setMasterPosition(startPosition); + controller.setEnableTsdb(true); + controller.setDestination("example"); + controller.setTsdbSpringXml("classpath:tsdb/mysql-tsdb.xml"); + controller.setEventFilter(new AviaterRegexFilter("test\\..*")); + controller.setEventBlackFilter(new AviaterRegexFilter("canal_tsdb\\..*")); controller.setEventSink(new AbstractCanalEventSinkTest>() { public boolean sink(List entrys, InetSocketAddress remoteAddress, String destination) @@ -100,7 +106,7 @@ public class MysqlDumpTest { controller.start(); try { - Thread.sleep(100 * 1000L); + Thread.sleep(100 * 1000 * 1000L); } catch (InterruptedException e) { Assert.fail(e.getMessage()); } diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMetaTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMetaTest.java index 8ead5e2a..9ef0dfc3 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMetaTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMetaTest.java @@ -4,32 +4,31 @@ import java.io.File; import java.io.FileInputStream; import java.net.URL; -import com.alibaba.otter.canal.parse.inbound.TableMeta; - import org.apache.commons.io.IOUtils; import org.apache.commons.lang.StringUtils; - import org.junit.Assert; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import com.alibaba.otter.canal.parse.inbound.TableMeta; + /** * @author agapple 2017年8月1日 下午7:15:54 */ @RunWith(SpringJUnit4ClassRunner.class) -@ContextConfiguration(locations = { "/dal-dao.xml" }) +@ContextConfiguration(locations = { "/tsdb/mysql-tsdb.xml" }) public class MemoryTableMetaTest { @Test public void testSimple() throws Throwable { - MemoryTableMeta memoryTableMeta = new MemoryTableMeta(null); + MemoryTableMeta memoryTableMeta = new MemoryTableMeta(); URL url = Thread.currentThread().getContextClassLoader().getResource("dummy.txt"); File dummyFile = new File(url.getFile()); File create = new File(dummyFile.getParent() + "/ddl", "create.sql"); String sql = StringUtils.join(IOUtils.readLines(new FileInputStream(create)), "\n"); - memoryTableMeta.apply(null, "test", sql); + memoryTableMeta.apply(null, "test", sql, null); TableMeta meta = memoryTableMeta.find("test", "test"); System.out.println(meta); diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MetaHistoryDAOTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MetaHistoryDAOTest.java index 29c02f7e..b199a326 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MetaHistoryDAOTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MetaHistoryDAOTest.java @@ -4,27 +4,27 @@ import java.util.List; import javax.annotation.Resource; -import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryDAO; -import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.model.MetaHistoryDO; - import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryDAO; +import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryDO; + /** * Created by wanshao Date: 2017/9/20 Time: 下午5:00 **/ @RunWith(SpringJUnit4ClassRunner.class) -@ContextConfiguration(locations = { "/dal-dao.xml" }) +@ContextConfiguration(locations = { "/tsdb/mysql-tsdb.xml" }) public class MetaHistoryDAOTest { @Resource MetaHistoryDAO metaHistoryDAO; @Test - public void testGetAll() { - List metaHistoryDOList = metaHistoryDAO.getAll(); + public void testSimple() { + List metaHistoryDOList = metaHistoryDAO.findByTimestamp("test", 0L, 0L); for (MetaHistoryDO metaHistoryDO : metaHistoryDOList) { System.out.println(metaHistoryDO.getId()); } diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaManagerBuilderTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaManagerBuilderTest.java new file mode 100644 index 00000000..76bd771e --- /dev/null +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaManagerBuilderTest.java @@ -0,0 +1,18 @@ +package com.alibaba.otter.canal.parse.inbound.mysql.tsdb; + +import org.junit.Test; +import org.springframework.util.Assert; + +/** + * @author agapple 2017年10月12日 上午10:50:00 + * @since 1.0.25 + */ +public class TableMetaManagerBuilderTest { + + @Test + public void testSimple() { + TableMetaTSDB tableMetaTSDB = TableMetaTSDBBuilder.build("test", "classpath:tsdb/mysql-tsdb.xml"); + Assert.notNull(tableMetaTSDB); + TableMetaTSDBBuilder.destory("test"); + } +} diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaManagerTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaManagerTest.java index 13413d81..b17dd8ea 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaManagerTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaManagerTest.java @@ -8,7 +8,6 @@ import java.net.URL; import javax.annotation.Resource; -import com.taobao.tddl.dbsync.binlog.BinlogPosition; import org.apache.commons.io.IOUtils; import org.apache.commons.lang.StringUtils; import org.junit.Test; @@ -16,12 +15,14 @@ import org.junit.runner.RunWith; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import com.alibaba.otter.canal.protocol.position.EntryPosition; + /** * @author wanshao 2017年8月2日 下午4:11:45 * @since 3.2.5 */ @RunWith(SpringJUnit4ClassRunner.class) -@ContextConfiguration(locations = { "/dal-dao.xml" }) +@ContextConfiguration(locations = { "/tsdb/mysql-tsdb.xml" }) public class TableMetaManagerTest { @Resource @@ -29,17 +30,18 @@ public class TableMetaManagerTest { @Test public void testSimple() throws FileNotFoundException, IOException { + tableMetaManager.init("test"); URL url = Thread.currentThread().getContextClassLoader().getResource("dummy.txt"); File dummyFile = new File(url.getFile()); File create = new File(dummyFile.getParent() + "/ddl", "create.sql"); - BinlogPosition position = BinlogPosition.parseFromString("001115:0139177334#3065927853.1501660815000"); + EntryPosition position = new EntryPosition("mysql-bin.001115", 139177334L, 3065927853L, 1501660815000L); String createSql = StringUtils.join(IOUtils.readLines(new FileInputStream(create)), "\n"); - tableMetaManager.apply(position, "tddl5_00", createSql); + tableMetaManager.apply(position, "tddl5_00", createSql, null); String alterSql = "alter table `test` add column name varchar(32) after c_varchar"; - position = BinlogPosition.parseFromString("001115:0139177334#3065927853.1501660816000"); - tableMetaManager.apply(position, "tddl5_00", alterSql); + position = new EntryPosition("mysql-bin.001115", 139177334L, 3065927853L, 1501660815000L); + tableMetaManager.apply(position, "tddl5_00", alterSql, null); } } diff --git a/parse/src/test/resources/ddl/create.sql b/parse/src/test/resources/ddl/create.sql index 273be888..6f64a3a6 100644 --- a/parse/src/test/resources/ddl/create.sql +++ b/parse/src/test/resources/ddl/create.sql @@ -1,5 +1,5 @@ CREATE TABLE `test` ( - `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT 'id', + `id` bigint(20) zerofill unsigNed NOT NULL AUTO_INCREMENT COMMENT 'id', `c_tinyint` tinyint(4) DEFAULT '1' COMMENT 'tinyint', `c_smallint` smallint(6) DEFAULT 0 COMMENT 'smallint', `c_mediumint` mediumint(9) DEFAULT NULL COMMENT 'mediumint', diff --git a/parse/src/test/resources/tsdb/dal-dao.xml b/parse/src/test/resources/tsdb/dal-dao.xml deleted file mode 100644 index 1842641f..00000000 --- a/parse/src/test/resources/tsdb/dal-dao.xml +++ /dev/null @@ -1,39 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/parse/src/test/resources/tsdb/derby-tsdb.xml b/parse/src/test/resources/tsdb/derby-tsdb.xml new file mode 100644 index 00000000..a6c19486 --- /dev/null +++ b/parse/src/test/resources/tsdb/derby-tsdb.xml @@ -0,0 +1,44 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/parse/src/test/resources/tsdb/mybatis-config.xml b/parse/src/test/resources/tsdb/mybatis-config.xml deleted file mode 100644 index 83ac0116..00000000 --- a/parse/src/test/resources/tsdb/mybatis-config.xml +++ /dev/null @@ -1,20 +0,0 @@ - - - - - - - - - - - - - - - - - - \ No newline at end of file diff --git a/parse/src/test/resources/tsdb/mysql-tsdb.xml b/parse/src/test/resources/tsdb/mysql-tsdb.xml new file mode 100644 index 00000000..e3743109 --- /dev/null +++ b/parse/src/test/resources/tsdb/mysql-tsdb.xml @@ -0,0 +1,47 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/parse/src/test/resources/tsdb/sqlmap-config.xml b/parse/src/test/resources/tsdb/sql-map/sqlmap-config.xml similarity index 68% rename from parse/src/test/resources/tsdb/sqlmap-config.xml rename to parse/src/test/resources/tsdb/sql-map/sqlmap-config.xml index 983e9663..2485a35b 100644 --- a/parse/src/test/resources/tsdb/sqlmap-config.xml +++ b/parse/src/test/resources/tsdb/sql-map/sqlmap-config.xml @@ -3,7 +3,6 @@ "http://www.ibatis.com/dtd/sql-map-config-2.dtd"> - - - + + diff --git a/parse/src/test/resources/tsdb/sql-map/sqlmap_history.xml b/parse/src/test/resources/tsdb/sql-map/sqlmap_history.xml index 00197194..7445bf69 100644 --- a/parse/src/test/resources/tsdb/sql-map/sqlmap_history.xml +++ b/parse/src/test/resources/tsdb/sql-map/sqlmap_history.xml @@ -1,57 +1,45 @@ - - - - + + - insert into `canal_table_meta_history` () - values(now(),now(),#binlogFile#,#binlogOffest#,#binlogMasterId#,#binlogTimestamp#,#useSchema#,#schema#,#table#,#sql#,#type#,#extra#); - - SELECT last_insert_id() - + insert into meta_history () + values(CURRENT_TIMESTAMP,CURRENT_TIMESTAMP,#destination#,#binlogFile#,#binlogOffest#,#binlogMasterId#,#binlogTimestamp#,#useSchema#,#sqlSchema#,#sqlTable#,#sqlText#,#sqlType#,#extra#) + + + delete from meta_history + where destination=#destination# + - - - - \ No newline at end of file diff --git a/parse/src/test/resources/tsdb/sql-map/sqlmap_snapshot.xml b/parse/src/test/resources/tsdb/sql-map/sqlmap_snapshot.xml index 9065a7e0..7ba0a435 100644 --- a/parse/src/test/resources/tsdb/sql-map/sqlmap_snapshot.xml +++ b/parse/src/test/resources/tsdb/sql-map/sqlmap_snapshot.xml @@ -1,66 +1,63 @@ - - - - - - + + + + - insert into `canal_table_meta_snapshot` () - values(now(),now(),#binlogFile#,#binlogOffest#,#binlogMasterId#,#binlogTimestamp#,#data#,#extra#); - - SELECT last_insert_id() - + insert into meta_snapshot () + values(CURRENT_TIMESTAMP,CURRENT_TIMESTAMP,#destination#,#binlogFile#,#binlogOffest#,#binlogMasterId#,#binlogTimestamp#,#data#,#extra#) - update `canal_table_meta_snapshot` set gmt_modified=now(), + update meta_snapshot set gmt_modified=now(), binlog_file=#binlogFile#,binlog_offest=#binlogOffest#,binlog_master_id=#binlogMasterId#,binlog_timestamp=#binlogTimestamp#,data=#data#,extra=#extra# - where binlog_timestamp=0 + where destination=#destination# and binlog_timestamp=0 + + delete from meta_snapshot + where destination=#destination# + - - - \ No newline at end of file diff --git a/pom.xml b/pom.xml index cb5f89cd..d1471058 100644 --- a/pom.xml +++ b/pom.xml @@ -109,6 +109,7 @@ 1.6 1.6 UTF-8 + 3.2.9.RELEASE @@ -131,14 +132,34 @@ - org.springframework - spring - 2.5.6 + org.springframework + spring-core + ${spring_version} + + + org.springframework + spring-aop + ${spring_version} + + + org.springframework + spring-context + ${spring_version} + + + org.springframework + spring-jdbc + ${spring_version} + + + org.springframework + spring-orm + ${spring_version} org.springframework spring-test - 2.5.6 + ${spring_version} test @@ -218,7 +239,7 @@ com.alibaba druid - 1.1.3 + 1.1.5-preview_14 @@ -251,9 +272,10 @@ mysql mysql-connector-java - 5.1.33 + 5.1.40 + diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/position/EntryPosition.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/position/EntryPosition.java index 2890e9ad..1f8ec38d 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/position/EntryPosition.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/position/EntryPosition.java @@ -1,5 +1,6 @@ package com.alibaba.otter.canal.protocol.position; + /** * 数据库对象的唯一标示 * @@ -121,4 +122,18 @@ public class EntryPosition extends TimePosition { return true; } + /** + * {@inheritDoc} + * + * @see java.lang.Comparable#compareTo(java.lang.Object) + */ + public int compareTo(EntryPosition o) { + final int val = journalName.compareTo(o.journalName); + + if (val == 0) { + return (int) (position - o.position); + } + return val; + } + }