diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index 09616194..033a3f89 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -60,7 +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.tsdb.spring.xml=classpath:spring/tsdb/h2-tsdb.xml +#canal.instance.tsdb.spring.xml=classpath:spring/tsdb/mysql-tsdb.xml canal.instance.global.mode = spring canal.instance.global.lazy = false diff --git a/deployer/src/main/resources/example/instance.properties b/deployer/src/main/resources/example/instance.properties index 50640d99..bef5cac3 100644 --- a/deployer/src/main/resources/example/instance.properties +++ b/deployer/src/main/resources/example/instance.properties @@ -4,13 +4,15 @@ canal.instance.mysql.slaveId=1234 # position info canal.instance.master.address=127.0.0.1:3306 canal.instance.master.journal.name=mysql-bin.000001 -canal.instance.master.position=68240 +canal.instance.master.position=104606 canal.instance.master.timestamp= # table meta tsdb info canal.instance.tsdb.enable=true -canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb +canal.instance.tsdb.dir=${canal.file.data.dir:../conf}/${canal.instance.destination:} +canal.instance.tsdb.url=jdbc:h2:${canal.instance.tsdb.dir}/h2;CACHE_SIZE=1000;MODE=MYSQL; +#canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/canal_tsdb canal.instance.tsdb.dbUsername=canal canal.instance.tsdb.dbPassword=canal diff --git a/deployer/src/main/resources/spring/tsdb/h2-tsdb.xml b/deployer/src/main/resources/spring/tsdb/h2-tsdb.xml new file mode 100644 index 00000000..5c3b869d --- /dev/null +++ b/deployer/src/main/resources/spring/tsdb/h2-tsdb.xml @@ -0,0 +1,60 @@ + + + + + + + + + + classpath:canal.properties + classpath:${canal.instance.destination:}/instance.properties + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/parse/pom.xml b/parse/pom.xml index 66e4bb60..0dd84d6f 100644 --- a/parse/pom.xml +++ b/parse/pom.xml @@ -58,6 +58,10 @@ org.apache.ibatis ibatis-sqlmap + + com.h2database + h2 + junit 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 c8fe89e1..4e6a392d 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 @@ -46,10 +46,11 @@ 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 Pattern h2Pattern = Pattern.compile("Unique index or primary key violation"); private static final EntryPosition INIT_POSITION = new EntryPosition("0", 0L, -2L, -1L); private String destination; private MemoryTableMeta memoryTableMeta; - private MysqlConnection connection; // 查询meta信息的链接 + private MysqlConnection connection; // 查询meta信息的链接 private CanalEventFilter filter; private CanalEventFilter blackFilter; private EntryPosition lastPosition; @@ -466,7 +467,7 @@ public class TableMetaManager implements TableMetaTSDB { } public boolean isUkDuplicateException(Throwable t) { - if (pattern.matcher(t.getMessage()).find()) { + if (pattern.matcher(t.getMessage()).find() || h2Pattern.matcher(t.getMessage()).find()) { // 违反外键约束时也抛出这种异常，所以这里还要判断包含字符串Duplicate entry return true; } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaBaseDAO.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaBaseDAO.java new file mode 100644 index 00000000..968e83b8 --- /dev/null +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/dao/MetaBaseDAO.java @@ -0,0 +1,59 @@ +package com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao; + +import java.io.InputStream; +import java.sql.Connection; +import java.sql.SQLException; +import java.sql.Statement; + +import javax.sql.DataSource; + +import org.apache.commons.io.IOUtils; +import org.apache.commons.lang.StringUtils; +import org.springframework.orm.ibatis.support.SqlMapClientDaoSupport; + +/** + * @author agapple 2017年10月14日 上午1:05:22 + * @since 1.0.25 + */ +@SuppressWarnings("deprecation") +public class MetaBaseDAO extends SqlMapClientDaoSupport { + + protected boolean isH2 = false; + + protected void initTable(String tableName) throws Exception { + Connection conn = null; + InputStream input = null; + try { + DataSource dataSource = getDataSource(); + conn = dataSource.getConnection(); + String name = "mysql"; + isH2 = isH2(conn); + if (isH2) { + name = "h2"; + } + input = Thread.currentThread() + .getContextClassLoader() + .getResourceAsStream("ddl/" + name + "/" + tableName + ".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 " + tableName + " failed", e); + } finally { + IOUtils.closeQuietly(input); + if (conn != null) { + conn.close(); + } + } + } + + private boolean isH2(Connection conn) throws SQLException { + String product = conn.getMetaData().getDatabaseProductName(); + return StringUtils.containsIgnoreCase(product, "H2"); + } +} 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 e43c5e62..6ec51614 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,19 +1,10 @@ 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.google.common.collect.Maps; /** @@ -23,7 +14,7 @@ import com.google.common.collect.Maps; * @since 3.2.5 */ @SuppressWarnings("deprecation") -public class MetaHistoryDAO extends SqlMapClientDaoSupport { +public class MetaHistoryDAO extends MetaBaseDAO { public Long insert(MetaHistoryDO metaDO) { return (Long) getSqlMapClientTemplate().insert("meta_history.insert", metaDO); @@ -56,27 +47,6 @@ public class MetaHistoryDAO extends SqlMapClientDaoSupport { } 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(); - } - } + initTable("meta_history"); } } 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 1a3f3053..f525031a 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,18 +1,9 @@ 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.google.common.collect.Maps; /** @@ -22,7 +13,7 @@ import com.google.common.collect.Maps; * @since 3.2.5 */ @SuppressWarnings("deprecation") -public class MetaSnapshotDAO extends SqlMapClientDaoSupport { +public class MetaSnapshotDAO extends MetaBaseDAO { public Long insert(MetaSnapshotDO snapshotDO) { return (Long) getSqlMapClientTemplate().insert("meta_snapshot.insert", snapshotDO); @@ -59,27 +50,7 @@ public class MetaSnapshotDAO extends SqlMapClientDaoSupport { } 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(); - } - } + initTable("meta_snapshot"); } + } diff --git a/parse/src/main/resources/ddl/h2/meta_history.sql b/parse/src/main/resources/ddl/h2/meta_history.sql new file mode 100644 index 00000000..4c7bbff4 --- /dev/null +++ b/parse/src/main/resources/ddl/h2/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 meta_history_binlog_file_offest(`destination`,`binlog_master_id`,`binlog_file`,`binlog_offest`), + KEY `meta_history_destination` (`destination`), + KEY `meta_history_destination_timestamp` (`destination`,`binlog_timestamp`), + KEY `meta_history_gmt_modified` (`gmt_modified`) +); \ No newline at end of file diff --git a/parse/src/main/resources/ddl/h2/meta_snapshot.sql b/parse/src/main/resources/ddl/h2/meta_snapshot.sql new file mode 100644 index 00000000..c14d7358 --- /dev/null +++ b/parse/src/main/resources/ddl/h2/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 meta_snapshot_binlog_file_offest(`destination`,`binlog_master_id`,`binlog_file`,`binlog_offest`), + KEY `meta_snapshot_destination` (`destination`), + KEY `meta_snapshot_destination_timestamp` (`destination`,`binlog_timestamp`), + KEY `meta_snapshot_gmt_modified` (`gmt_modified`) +); \ No newline at end of file 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 aef55822..a0754635 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 @@ -36,7 +36,7 @@ public class MysqlDumpTest { controller.setMasterPosition(startPosition); controller.setEnableTsdb(true); controller.setDestination("example"); - controller.setTsdbSpringXml("classpath:tsdb/mysql-tsdb.xml"); + controller.setTsdbSpringXml("classpath:tsdb/h2-tsdb.xml"); controller.setEventFilter(new AviaterRegexFilter("test\\..*")); controller.setEventBlackFilter(new AviaterRegexFilter("canal_tsdb\\..*")); controller.setEventSink(new AbstractCanalEventSinkTest>() { 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 b17dd8ea..af8c9dc5 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 @@ -22,7 +22,7 @@ import com.alibaba.otter.canal.protocol.position.EntryPosition; * @since 3.2.5 */ @RunWith(SpringJUnit4ClassRunner.class) -@ContextConfiguration(locations = { "/tsdb/mysql-tsdb.xml" }) +@ContextConfiguration(locations = { "/tsdb/h2-tsdb.xml" }) public class TableMetaManagerTest { @Resource @@ -40,8 +40,7 @@ public class TableMetaManagerTest { tableMetaManager.apply(position, "tddl5_00", createSql, null); String alterSql = "alter table `test` add column name varchar(32) after c_varchar"; - position = new EntryPosition("mysql-bin.001115", 139177334L, 3065927853L, 1501660815000L); + position = new EntryPosition("mysql-bin.001115", 139177334L, 3065927854L, 1501660815000L); tableMetaManager.apply(position, "tddl5_00", alterSql, null); - } } diff --git a/parse/src/test/resources/tsdb/h2-tsdb.xml b/parse/src/test/resources/tsdb/h2-tsdb.xml new file mode 100644 index 00000000..452e7ee5 --- /dev/null +++ b/parse/src/test/resources/tsdb/h2-tsdb.xml @@ -0,0 +1,44 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/pom.xml b/pom.xml index d1471058..649abb36 100644 --- a/pom.xml +++ b/pom.xml @@ -262,6 +262,11 @@ slf4j-api 1.7.12 + + com.h2database + h2 + 1.4.196 + junit