From 02868d8fecb50e445d8b624ca70b25da36cda788 Mon Sep 17 00:00:00 2001 From: yfpeng Date: Tue, 31 Jul 2018 20:30:32 +0800 Subject: [PATCH] =?UTF-8?q?=E4=B8=BA=E4=BA=86=E8=A7=A3=E5=86=B3=E5=9B=9E?= =?UTF-8?q?=E8=BF=BD=E6=95=B0=E6=8D=AE=E6=97=B6Rowdata=E4=B8=8E=E5=BD=93?= =?UTF-8?q?=E5=89=8Dschema=E4=B8=8D=E5=8C=B9=E9=85=8D=20=E5=AF=B9TableMeta?= =?UTF-8?q?=E6=8F=90=E4=BE=9B=E6=8C=81=E4=B9=85=E5=8C=96=E6=8E=A5=E5=8F=A3?= =?UTF-8?q?=EF=BC=8C=E5=8F=AF=E4=BB=A5=E6=A0=B9=E6=8D=AEtimestamp=E5=8C=B9?= =?UTF-8?q?=E9=85=8D=E5=8E=86=E5=8F=B2schema?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../manager/CanalInstanceWithManager.java | 8 + .../manager/model/CanalParameter.java | 12 ++ .../parse/inbound/mysql/MysqlEventParser.java | 27 ++- .../inbound/mysql/dbsync/LogEventConvert.java | 11 +- .../inbound/mysql/dbsync/TableMetaCache.java | 7 +- .../tablemeta/HistoryTableMetaCache.java | 191 ++++++++++++++++++ .../tablemeta/TableMetaCacheInterface.java | 20 ++ .../tablemeta/TableMetaCacheWithStorage.java | 105 ++++++++++ .../mysql/tablemeta/TableMetaEntry.java | 55 +++++ .../mysql/tablemeta/TableMetaStorage.java | 18 ++ .../tablemeta/TableMetaStorageFactory.java | 9 + .../exception/CacheConnectionNull.java | 9 + .../exception/NoHistoryException.java | 21 ++ .../impl/mysql/MySqlTableMetaCallback.java | 14 ++ .../impl/mysql/MySqlTableMetaStorage.java | 48 +++++ .../mysql/MySqlTableMetaStorageFactory.java | 26 +++ .../mysql/tablemeta/NoStorageTest.java | 38 ++++ .../inbound/mysql/tablemeta/StorageTest.java | 77 +++++++ pom.xml | 2 +- 19 files changed, 687 insertions(+), 11 deletions(-) create mode 100644 parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/HistoryTableMetaCache.java create mode 100644 parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/TableMetaCacheInterface.java create mode 100644 parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/TableMetaCacheWithStorage.java create mode 100644 parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/TableMetaEntry.java create mode 100644 parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/TableMetaStorage.java create mode 100644 parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/TableMetaStorageFactory.java create mode 100644 parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/exception/CacheConnectionNull.java create mode 100644 parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/exception/NoHistoryException.java create mode 100644 parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/impl/mysql/MySqlTableMetaCallback.java create mode 100644 parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/impl/mysql/MySqlTableMetaStorage.java create mode 100644 parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/impl/mysql/MySqlTableMetaStorageFactory.java create mode 100644 parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/NoStorageTest.java create mode 100644 parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/StorageTest.java diff --git a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java index bca5ba13..f6cec13d 100644 --- a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java +++ b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java @@ -6,6 +6,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; +import com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.HistoryTableMetaCache; import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -242,6 +243,13 @@ public class CanalInstanceWithManager extends AbstractCanalInstance { mysqlEventParser.setDetectingIntervalInSeconds(parameters.getDetectingIntervalInSeconds()); // 数据库信息参数 mysqlEventParser.setSlaveId(parameters.getSlaveId()); + mysqlEventParser.setTableMetaStorageFactory(parameters.getTableMetaStorageFactory()); + // Ctrip callback +// mysqlEventParser.setCallback(parameters.getCallback()); +// HistoryTableMetaCache cache = new HistoryTableMetaCache(); +// cache.init(parameters.getEntries()); +// mysqlEventParser.setHistoryTableMetaCache(cache); + if (!CollectionUtils.isEmpty(dbAddresses)) { mysqlEventParser.setMasterInfo(new AuthenticationInfo(dbAddresses.get(0), parameters.getDbUsername(), diff --git a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/model/CanalParameter.java b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/model/CanalParameter.java index 16c929ca..94558590 100644 --- a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/model/CanalParameter.java +++ b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/model/CanalParameter.java @@ -5,6 +5,7 @@ import java.net.InetSocketAddress; import java.util.ArrayList; import java.util.List; +import com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.TableMetaStorageFactory; import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.builder.ToStringBuilder; @@ -106,6 +107,9 @@ public class CanalParameter implements Serializable { private Long standbyLogfileOffest = null; private Long standbyTimestamp = null; + // Ctrip Table Meta + TableMetaStorageFactory tableMetaStorageFactory; + public static enum RunMode { /** 嵌入式 */ @@ -859,6 +863,14 @@ public class CanalParameter implements Serializable { this.blackFilter = blackFilter; } + public TableMetaStorageFactory getTableMetaStorageFactory() { + return tableMetaStorageFactory; + } + + public void setTableMetaStorageFactory(TableMetaStorageFactory tableMetaStorageFactory) { + this.tableMetaStorageFactory = tableMetaStorageFactory; + } + public String toString() { return ToStringBuilder.reflectionToString(this, CanalToStringStyle.DEFAULT_STYLE); } 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 2efab0ee..a30af259 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 @@ -10,6 +10,11 @@ import java.util.Map; import java.util.TimerTask; import java.util.concurrent.atomic.AtomicLong; +import com.alibaba.otter.canal.parse.inbound.*; +import com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.TableMetaCacheInterface; +import com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.TableMetaCacheWithStorage; +import com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.TableMetaStorage; +import com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.TableMetaStorageFactory; import org.apache.commons.lang.StringUtils; import org.springframework.util.CollectionUtils; @@ -20,9 +25,6 @@ import com.alibaba.otter.canal.parse.driver.mysql.packets.server.FieldPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ResultSetPacket; import com.alibaba.otter.canal.parse.exception.CanalParseException; import com.alibaba.otter.canal.parse.ha.CanalHAController; -import com.alibaba.otter.canal.parse.inbound.ErosaConnection; -import com.alibaba.otter.canal.parse.inbound.HeartBeatCallback; -import com.alibaba.otter.canal.parse.inbound.SinkFunction; import com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection.BinlogFormat; import com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection.BinlogImage; import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert; @@ -62,7 +64,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE // 心跳检查信息 private String detectingSQL; // 心跳sql private MysqlConnection metaConnection; // 查询meta信息的链接 - private TableMetaCache tableMetaCache; // 对应meta + private TableMetaCacheInterface tableMetaCache; // 对应meta // cache private int fallbackIntervalInSeconds = 60; // 切换回退时间 private BinlogFormat[] supportBinlogFormats; // 支持的binlogFormat,如果设置会执行强校验 @@ -72,6 +74,8 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE private int dumpErrorCount = 0; // binlogDump失败异常计数 private int dumpErrorCountThreshold = 2; // binlogDump失败异常计数阀值 + private TableMetaStorageFactory tableMetaStorageFactory; + protected ErosaConnection buildErosaConnection() { return buildMysqlConnection(this.runningInfo); } @@ -125,7 +129,13 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE ((DatabaseTableMeta) tableMetaTSDB).setBlackFilter(eventBlackFilter); } - tableMetaCache = new TableMetaCache(metaConnection, tableMetaTSDB); + + TableMetaStorage storage = null; + if (tableMetaStorageFactory != null) { + storage = tableMetaStorageFactory.getTableMetaStorage(); + } + + tableMetaCache = new TableMetaCacheWithStorage(metaConnection, storage); ((LogEventConvert) binlogParser).setTableMetaCache(tableMetaCache); } } @@ -908,4 +918,11 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE this.dumpErrorCountThreshold = dumpErrorCountThreshold; } + public TableMetaStorageFactory getTableMetaStorageFactory() { + return tableMetaStorageFactory; + } + + public void setTableMetaStorageFactory(TableMetaStorageFactory tableMetaStorageFactory) { + this.tableMetaStorageFactory = tableMetaStorageFactory; + } } 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 fe34e199..67bb07b7 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 @@ -10,6 +10,9 @@ import java.util.Arrays; import java.util.BitSet; import java.util.List; +import com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.TableMetaCacheInterface; +import com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.TableMetaStorage; +import com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.exception.NoHistoryException; import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.exception.ExceptionUtils; import org.slf4j.Logger; @@ -88,7 +91,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar private volatile AviaterRegexFilter nameFilter; // 运行时引用可能会有变化，比如规则发生变化时 private volatile AviaterRegexFilter nameBlackFilter; - private TableMetaCache tableMetaCache; + private TableMetaCacheInterface tableMetaCache; private String binlogFileName = "mysql-bin.000001"; private Charset charset = Charset.defaultCharset(); private boolean filterQueryDcl = false; @@ -262,7 +265,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar if (!isSeek) { // 使用新的表结构元数据管理方式 EntryPosition position = createPosition(event.getHeader()); - tableMetaCache.apply(position, event.getDbName(), queryString, null); + String fulltbName = schemaName+"."+tableName; + tableMetaCache.apply(position, fulltbName, queryString, null); } Header header = createHeader(binlogFileName, event.getHeader(), schemaName, tableName, type); @@ -901,7 +905,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar this.nameBlackFilter = nameBlackFilter; } - public void setTableMetaCache(TableMetaCache tableMetaCache) { + public void setTableMetaCache(TableMetaCacheInterface tableMetaCache) { this.tableMetaCache = tableMetaCache; } @@ -928,5 +932,4 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar public void setGtidSet(GTIDSet gtidSet) { this.gtidSet = gtidSet; } - } 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 4c1cef3c..5c782aa1 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 @@ -6,6 +6,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.TableMetaCacheInterface; import org.apache.commons.lang.StringUtils; import com.alibaba.otter.canal.parse.driver.mysql.packets.server.FieldPacket; @@ -29,7 +30,7 @@ import com.google.common.cache.LoadingCache; * @author jianghang 2013-1-17 下午10:15:16 * @version 1.0.0 */ -public class TableMetaCache { +public class TableMetaCache implements TableMetaCacheInterface { public static final String COLUMN_NAME = "COLUMN_NAME"; public static final String COLUMN_TYPE = "COLUMN_TYPE"; @@ -99,6 +100,10 @@ public class TableMetaCache { String createDDL = packet.getFieldValues().get(1); MemoryTableMeta memoryTableMeta = new MemoryTableMeta(); memoryTableMeta.apply(DatabaseTableMeta.INIT_POSITION, schema, createDDL, null); + String[] strings = table.split("\\."); + if (strings.length > 1) { + table = strings[1]; + } TableMeta tableMeta = memoryTableMeta.find(schema, table); return tableMeta.getFields(); } else { diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/HistoryTableMetaCache.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/HistoryTableMetaCache.java new file mode 100644 index 00000000..0964627d --- /dev/null +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/HistoryTableMetaCache.java @@ -0,0 +1,191 @@ +package com.alibaba.otter.canal.parse.inbound.mysql.tablemeta; + +import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ResultSetPacket; +import com.alibaba.otter.canal.parse.inbound.TableMeta; +import com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection; +import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.TableMetaCache; +import com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.exception.CacheConnectionNull; +import com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.exception.NoHistoryException; +import com.google.common.cache.CacheBuilder; +import com.google.common.cache.CacheLoader; +import com.google.common.cache.LoadingCache; + +import java.io.IOException; +import java.util.*; + +public class HistoryTableMetaCache { + private TableMetaStorage tableMetaStorage; + private MysqlConnection metaConnection; + private LoadingCache> cache; // 第一层：数据库名.表名，第二层时间戳，TableMeta + + public HistoryTableMetaCache() { + cache = CacheBuilder.newBuilder().build(new CacheLoader>() { + @Override + public Map load(String tableName) throws Exception { + Long timestamp = new Date().getTime(); + String[] strs = tableName.split("\\."); + String schema = strs[0]; + if (tableMetaStorage != null) { + init(tableMetaStorage.fetchByTableName(tableName)); // 从存储中读取表的历史ddl + } + ResultSetPacket resultSetPacket = connectionQuery("show create table " + tableName); // 获取当前ddl + String currentDdl = resultSetPacket.getFieldValues().get(1); + if (cache.asMap().containsKey(tableName)) { + Map tableMetaMap = cache.getUnchecked(tableName); + if (tableMetaMap.isEmpty()) { + put(schema, tableName, currentDdl, timestamp - 1000L); // 放入当前schema，取时间为当前时间-1s + } else { // 如果table存在历史 + Iterator iterator = tableMetaMap.keySet().iterator(); + Long firstTimestamp = iterator.next(); + TableMeta first = tableMetaMap.get(firstTimestamp); // 拿第一条ddl + if (!first.getDdl().equalsIgnoreCase(currentDdl)) { // 当前ddl与历史第一条不一致，放入当前ddl + put(schema, tableName, currentDdl, calculateNewTimestamp(firstTimestamp)); // 计算放入的timestamp,设为第一条时间+1s + } + } + } else { + put(schema, tableName, currentDdl, timestamp - 1000L); // 放入当前schema + } + return cache.get(tableName); + } + }); + } + + public void init(List entries) throws IOException { + if (entries == null) { + return; + } + for (TableMetaEntry entry : entries) { + try { + put(entry.getSchema(), entry.getTable(), entry.getDdl(), entry.getTimestamp()); + } catch (CacheConnectionNull cacheConnectionNull) { + cacheConnectionNull.printStackTrace(); + } + } + } + + public TableMeta put(String schema, String table, String ddl, Long timestamp) throws CacheConnectionNull, IOException { + ResultSetPacket resultSetPacket; + if (!(ddl.contains("CREATE TABLE") || ddl.contains("create table"))) { // 尝试直接从数据库拉取CREATE TABLE的DDL + resultSetPacket = connectionQuery("show create table " + table); + ddl = resultSetPacket.getFieldValues().get(1); + } else { // CREATE TABLE 的 DDL + resultSetPacket = new ResultSetPacket(); + List fields = new ArrayList(); + String[] strings = table.split("\\."); + String shortTable = table; + if (strings.length > 1) { + shortTable = strings[1]; + } + fields.add(0, shortTable); + fields.add(1, ddl); + resultSetPacket.setFieldValues(fields); + if (metaConnection != null) { + resultSetPacket.setSourceAddress(metaConnection.getAddress()); + } + } + Map tableMetaMap; + if (!cache.asMap().containsKey(table)) { + tableMetaMap = new TreeMap(new Comparator() { + @Override + public int compare(Long o1, Long o2) { + return o2.compareTo(o1); + } + }); + cache.put(table, tableMetaMap); + } else { + tableMetaMap = cache.getUnchecked(table); + } + eliminate(tableMetaMap); // 淘汰旧的TableMeta + TableMeta tableMeta = new TableMeta(schema, table, TableMetaCache.parseTableMeta(schema, table, resultSetPacket)); + if (tableMeta.getDdl() == null) { // 生成的TableMeta有时DDL为null + tableMeta.setDdl(ddl); + } + tableMetaMap.put(timestamp, tableMeta); + return tableMeta; + } + + public TableMeta get(String schema, String table, Long timestamp) throws NoHistoryException, CacheConnectionNull { + Map tableMetaMap = cache.getUnchecked(table); + Iterator iterator = tableMetaMap.keySet().iterator(); + Long selected = null; + while(iterator.hasNext()) { + Long temp = iterator.next(); + if (timestamp > temp) { + selected = temp; + break; + } + } + + if (selected == null) { + iterator = tableMetaMap.keySet().iterator(); + if (iterator.hasNext()) { + selected = iterator.next(); + } else { + throw new NoHistoryException(schema, table); + } + } + + return tableMetaMap.get(selected); + } + + public void clearTableMeta() { + cache.invalidateAll(); + } + + public void clearTableMetaWithSchemaName(String schema) { + for (String tableName : cache.asMap().keySet()) { + String[] strs = tableName.split("\\."); + if (schema.equalsIgnoreCase(strs[0])) { + cache.invalidate(tableName); + } + } + } + + public void clearTableMeta(String schema, String table) { + if (!table.contains(".")) { + table = schema+"."+table; + } + cache.invalidate(table); + } + + // eliminate older table meta in cache + private void eliminate(Map tableMetaMap) { + int MAX_CAPABILITY = 20; + if (tableMetaMap.keySet().size() < MAX_CAPABILITY) { + return; + } + Iterator iterator = tableMetaMap.keySet().iterator(); + while(iterator.hasNext()) { + iterator.next(); + } + iterator.remove(); + } + + private Long calculateNewTimestamp(Long oldTimestamp) { + return oldTimestamp + 1000; + } + + private ResultSetPacket connectionQuery(String query) throws CacheConnectionNull, IOException { + if (metaConnection == null) { + throw new CacheConnectionNull(); + } + try { + return metaConnection.query(query); + } catch (IOException e) { + try { + metaConnection.reconnect(); + return metaConnection.query(query); + } catch (IOException e1) { + throw e1; + } + } + } + + public void setMetaConnection(MysqlConnection metaConnection) { + this.metaConnection = metaConnection; + } + + public void setTableMetaStorage(TableMetaStorage tableMetaStorage) { + this.tableMetaStorage = tableMetaStorage; + } +} diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/TableMetaCacheInterface.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/TableMetaCacheInterface.java new file mode 100644 index 00000000..5196ee08 --- /dev/null +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/TableMetaCacheInterface.java @@ -0,0 +1,20 @@ +package com.alibaba.otter.canal.parse.inbound.mysql.tablemeta; + +import com.alibaba.otter.canal.parse.inbound.TableMeta; +import com.alibaba.otter.canal.protocol.position.EntryPosition; + +public interface TableMetaCacheInterface { + + TableMeta getTableMeta(String schema, String table, boolean useCache, EntryPosition position); + + void clearTableMeta(); + + void clearTableMetaWithSchemaName(String schema); + + void clearTableMeta(String schema, String table); + + boolean apply(EntryPosition position, String schema, String ddl, String extra); + + boolean isOnRDS(); + +} diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/TableMetaCacheWithStorage.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/TableMetaCacheWithStorage.java new file mode 100644 index 00000000..eb43158b --- /dev/null +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/TableMetaCacheWithStorage.java @@ -0,0 +1,105 @@ +package com.alibaba.otter.canal.parse.inbound.mysql.tablemeta; + +import com.alibaba.otter.canal.parse.inbound.TableMeta; +import com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection; +import com.alibaba.otter.canal.protocol.position.EntryPosition; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.net.InetSocketAddress; + +public class TableMetaCacheWithStorage implements TableMetaCacheInterface { + + private static Logger logger = LoggerFactory.getLogger(TableMetaCacheWithStorage.class); + private TableMetaStorage tableMetaStorage; // TableMeta存储 + private HistoryTableMetaCache cache = new HistoryTableMetaCache(); // cache + + public TableMetaCacheWithStorage(MysqlConnection con, TableMetaStorage tableMetaStorage) { + this.tableMetaStorage = tableMetaStorage; + InetSocketAddress address = con.getAddress(); + this.tableMetaStorage.setDbAddress(address.getHostName()+":"+address.getPort()); + cache.setMetaConnection(con); + cache.setTableMetaStorage(tableMetaStorage); + if (tableMetaStorage != null) { + try { + cache.init(tableMetaStorage.fetch()); // 初始化，从存储拉取TableMeta + } catch (IOException e) { + logger.error(e.getMessage()); + } + } + } + + @Override + public boolean apply(EntryPosition position, String fullTableName, String ddl, String extra) { + String[] strs = fullTableName.split("\\."); + String schema = strs[0]; + if (schema.equalsIgnoreCase("null")) { // ddl schema为null，放弃处理 + return false; + } + try { + TableMeta tableMeta = cache.get(schema, fullTableName, position.getTimestamp()); + if (!compare(tableMeta, ddl)) { // 获取最近的TableMeta，进行比对 + TableMeta result = cache.put(schema, fullTableName, ddl, calTimestamp(position.getTimestamp())); + if (tableMetaStorage != null && result != null) { // 储存 + tableMetaStorage.store(schema, fullTableName, result.getDdl(), calTimestamp(position.getTimestamp())); + } + } + return true; + } catch (Exception e) { + logger.error(e.toString()); + } + + return false; + } + + @Override + public boolean isOnRDS() { + return false; + } + + /*** + * + * @param schema dbname + * @param table tablename + * @param useCache unused + * @param position timestamp + * @return + */ + @Override + public TableMeta getTableMeta(String schema, String table, boolean useCache, EntryPosition position) { + String fulltbName = schema + "." + table; + try { + return cache.get(schema, fulltbName, position.getTimestamp()); + } catch (Exception e) { + logger.error(e.toString()); + } + return null; + } + + @Override + public void clearTableMeta() { + cache.clearTableMeta(); + } + + @Override + public void clearTableMetaWithSchemaName(String schema) { + cache.clearTableMetaWithSchemaName(schema); + } + + @Override + public void clearTableMeta(String schema, String table) { + cache.clearTableMeta(schema, table); + } + + private boolean compare(TableMeta tableMeta, String ddl) { + if (tableMeta == null) { + return false; + } + return tableMeta.getDdl().equalsIgnoreCase(ddl); + } + + private Long calTimestamp(Long timestamp) { + return timestamp; + } +} diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/TableMetaEntry.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/TableMetaEntry.java new file mode 100644 index 00000000..03e63ed7 --- /dev/null +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/TableMetaEntry.java @@ -0,0 +1,55 @@ +package com.alibaba.otter.canal.parse.inbound.mysql.tablemeta; + +import java.io.Serializable; + +public class TableMetaEntry implements Serializable { + + private static final long serialVersionUID = -1350200637109107904L; + + private String dbAddress; + private String schema; + private String table; + private String ddl; + private Long timestamp; + + + 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 getDdl() { + return ddl; + } + + public void setDdl(String ddl) { + this.ddl = ddl; + } + + public Long getTimestamp() { + return timestamp; + } + + public void setTimestamp(Long timestamp) { + this.timestamp = timestamp; + } + + public String getDbAddress() { + return dbAddress; + } + + public void setDbAddress(String dbAddress) { + this.dbAddress = dbAddress; + } +} diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/TableMetaStorage.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/TableMetaStorage.java new file mode 100644 index 00000000..ac9d7a0a --- /dev/null +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/TableMetaStorage.java @@ -0,0 +1,18 @@ +package com.alibaba.otter.canal.parse.inbound.mysql.tablemeta; + +import java.util.List; + +public interface TableMetaStorage { + + void store(String schema, String table, String ddl, Long timestamp); + + List fetch(); + + List fetchByTableName(String tableName); + + String getDbName(); + + String getDbAddress(); + + void setDbAddress(String address); +} diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/TableMetaStorageFactory.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/TableMetaStorageFactory.java new file mode 100644 index 00000000..0e532ebb --- /dev/null +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/TableMetaStorageFactory.java @@ -0,0 +1,9 @@ +package com.alibaba.otter.canal.parse.inbound.mysql.tablemeta; + +public interface TableMetaStorageFactory { + + TableMetaStorage getTableMetaStorage(); + + String getDbName(); + +} diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/exception/CacheConnectionNull.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/exception/CacheConnectionNull.java new file mode 100644 index 00000000..1dadd78a --- /dev/null +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/exception/CacheConnectionNull.java @@ -0,0 +1,9 @@ +package com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.exception; + +public class CacheConnectionNull extends Exception{ + + @Override + public String toString() { + return "CacheConnectionNull"; + } +} diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/exception/NoHistoryException.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/exception/NoHistoryException.java new file mode 100644 index 00000000..d8f1d16b --- /dev/null +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/exception/NoHistoryException.java @@ -0,0 +1,21 @@ +package com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.exception; + +public class NoHistoryException extends Exception{ + + private String dbName; + private String tbName; + + public NoHistoryException(String dbName, String tbName) { + this.dbName = dbName; + this.tbName = tbName; + } + + public void printTableName() { + System.out.println(dbName+"."+tbName); + } + + @Override + public String toString() { + return "NioHistoryException: " + dbName + " " + tbName; + } +} diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/impl/mysql/MySqlTableMetaCallback.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/impl/mysql/MySqlTableMetaCallback.java new file mode 100644 index 00000000..661986a1 --- /dev/null +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/impl/mysql/MySqlTableMetaCallback.java @@ -0,0 +1,14 @@ +package com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.impl.mysql; + +import com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.TableMetaEntry; + +import java.util.List; + +public interface MySqlTableMetaCallback { + + void save(String dbAddress, String schema, String table,String ddl, Long timestamp); + + List fetch(String dbAddress, String dbName); + + List fetch(String dbAddress, String dbName, String tableName); +} diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/impl/mysql/MySqlTableMetaStorage.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/impl/mysql/MySqlTableMetaStorage.java new file mode 100644 index 00000000..f120ef80 --- /dev/null +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/impl/mysql/MySqlTableMetaStorage.java @@ -0,0 +1,48 @@ +package com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.impl.mysql; + +import com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.TableMetaEntry; +import com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.TableMetaStorage; + +import java.util.List; + +public class MySqlTableMetaStorage implements TableMetaStorage { + private MySqlTableMetaCallback mySqlTableMetaCallback; + private String dbName; + private String dbAddress; + + MySqlTableMetaStorage(MySqlTableMetaCallback callback, String dbName) { + mySqlTableMetaCallback = callback; + this.dbName = dbName; + } + + + @Override + public void store(String schema, String table, String ddl, Long timestamp) { + mySqlTableMetaCallback.save(dbAddress, schema, table, ddl, timestamp); + } + + @Override + public List fetch() { + return mySqlTableMetaCallback.fetch(dbAddress, dbName); + } + + @Override + public List fetchByTableName(String tableName) { + return mySqlTableMetaCallback.fetch(dbAddress, dbName, tableName); + } + + @Override + public String getDbName() { + return dbName; + } + + @Override + public String getDbAddress() { + return dbAddress; + } + + @Override + public void setDbAddress(String address) { + this.dbAddress = address; + } +} diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/impl/mysql/MySqlTableMetaStorageFactory.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/impl/mysql/MySqlTableMetaStorageFactory.java new file mode 100644 index 00000000..0eee235d --- /dev/null +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/impl/mysql/MySqlTableMetaStorageFactory.java @@ -0,0 +1,26 @@ +package com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.impl.mysql; + +import com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.TableMetaStorage; +import com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.TableMetaStorageFactory; + +public class MySqlTableMetaStorageFactory implements TableMetaStorageFactory { + + private MySqlTableMetaCallback mySQLTableMetaCallback; + private String dbName; + + public MySqlTableMetaStorageFactory(MySqlTableMetaCallback callback, String dbName) { + mySQLTableMetaCallback = callback; + this.dbName = dbName; + } + + @Override + public TableMetaStorage getTableMetaStorage() { + return new MySqlTableMetaStorage(mySQLTableMetaCallback, dbName); + } + + @Override + public String getDbName() { + return dbName; + } + +} diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/NoStorageTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/NoStorageTest.java new file mode 100644 index 00000000..04fd95ea --- /dev/null +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/NoStorageTest.java @@ -0,0 +1,38 @@ +package com.alibaba.otter.canal.parse.inbound.mysql.tablemeta; + +import com.alibaba.otter.canal.parse.inbound.TableMeta; +import com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection; +import com.alibaba.otter.canal.protocol.position.EntryPosition; +import org.junit.Test; + +import java.net.InetSocketAddress; +import java.util.Date; + +public class NoStorageTest { + final String DBNAME = "testdb"; + final String TBNAME = "testtb"; + final String DDL = "CREATE TABLE `testtb` (\n" + + " `id` int(11) NOT NULL AUTO_INCREMENT,\n" + + " `name` varchar(2048) DEFAULT NULL,\n" + + " `datachange_lasttime` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '最晚更新时间',\n" + + " `otter_testcol` varchar(45) DEFAULT NULL,\n" + + " `otter_testcol1` varchar(45) DEFAULT NULL,\n" + + " `otter_testcol2` varchar(45) DEFAULT NULL,\n" + + " `otter_testcol3` varchar(45) DEFAULT NULL,\n" + + " `otter_testcol4` varchar(45) DEFAULT NULL,\n" + + " `otter_testcol5` varchar(45) DEFAULT NULL,\n" + + " PRIMARY KEY (`id`)\n" + + " ) ENGINE=InnoDB AUTO_INCREMENT=58333898 DEFAULT CHARSET=utf8mb4"; + @Test + public void nostorage() { + MysqlConnection connection = new MysqlConnection(new InetSocketAddress("127.0.0.1", 3306), "root", "hello"); + TableMetaCacheWithStorage tableMetaCacheWithStorage = new TableMetaCacheWithStorage(connection, null); + EntryPosition entryPosition = new EntryPosition(); + entryPosition.setTimestamp(new Date().getTime()); + String fullTableName = DBNAME + "." + TBNAME; + tableMetaCacheWithStorage.apply(entryPosition, fullTableName, DDL, null); + entryPosition.setTimestamp(new Date().getTime() + 1000L); + TableMeta result = tableMetaCacheWithStorage.getTableMeta(DBNAME, TBNAME, false, entryPosition); + assert result.getDdl().equalsIgnoreCase(DDL); + } +} diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/StorageTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/StorageTest.java new file mode 100644 index 00000000..0ec16607 --- /dev/null +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/StorageTest.java @@ -0,0 +1,77 @@ +package com.alibaba.otter.canal.parse.inbound.mysql.tablemeta; + +import com.alibaba.otter.canal.parse.inbound.TableMeta; +import com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection; +import com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.impl.mysql.MySqlTableMetaCallback; +import com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.impl.mysql.MySqlTableMetaStorageFactory; +import com.alibaba.otter.canal.protocol.position.EntryPosition; +import com.alibaba.otter.canal.protocol.position.Position; +import org.junit.Test; + +import java.net.InetSocketAddress; +import java.util.ArrayList; +import java.util.Date; +import java.util.List; + +public class StorageTest { + + final String DBNAME = "testdb"; + final String TBNAME = "testtb"; + final String DDL = "CREATE TABLE `testtb` (\n" + + " `id` int(11) NOT NULL AUTO_INCREMENT,\n" + + " `name` varchar(2048) DEFAULT NULL,\n" + + " `datachange_lasttime` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '最晚更新时间',\n" + + " `otter_testcol` varchar(45) DEFAULT NULL,\n" + + " `otter_testcol1` varchar(45) DEFAULT NULL,\n" + + " `otter_testcol2` varchar(45) DEFAULT NULL,\n" + + " `otter_testcol3` varchar(45) DEFAULT NULL,\n" + + " `otter_testcol4` varchar(45) DEFAULT NULL,\n" + + " `otter_testcol5` varchar(45) DEFAULT NULL,\n" + + " PRIMARY KEY (`id`)\n" + + " ) ENGINE=InnoDB AUTO_INCREMENT=58333898 DEFAULT CHARSET=utf8mb4"; + + @Test + public void storage() { + + MySqlTableMetaStorageFactory factory = new MySqlTableMetaStorageFactory(new MySqlTableMetaCallback() { + @Override + public void save(String dbAddress, String schema, String table, String ddl, Long timestamp) { + + } + + @Override + public List fetch(String dbAddress, String dbName) { + TableMetaEntry tableMeta = new TableMetaEntry(); + tableMeta.setSchema(DBNAME); + tableMeta.setTable(TBNAME); + tableMeta.setDdl(DDL); + tableMeta.setTimestamp(new Date().getTime()); + List entries = new ArrayList(); + entries.add(tableMeta); + return entries; + } + + @Override + public List fetch(String dbAddress, String dbName, String tableName) { + TableMetaEntry tableMeta = new TableMetaEntry(); + tableMeta.setSchema(DBNAME); + tableMeta.setTable(TBNAME); + tableMeta.setDdl(DDL); + tableMeta.setTimestamp(new Date().getTime()); + List entries = new ArrayList(); + entries.add(tableMeta); + return entries; + } + }, DBNAME); + MysqlConnection connection = new MysqlConnection(new InetSocketAddress("127.0.0.1", 3306), "root", "hello"); + TableMetaCacheWithStorage tableMetaCacheWithStorage = new TableMetaCacheWithStorage(connection, factory.getTableMetaStorage()); + EntryPosition entryPosition = new EntryPosition(); + entryPosition.setTimestamp(new Date().getTime()); + String fullTableName = DBNAME + "." + TBNAME; + tableMetaCacheWithStorage.apply(entryPosition, fullTableName, DDL, null); + + entryPosition.setTimestamp(new Date().getTime() + 1000L); + TableMeta result = tableMetaCacheWithStorage.getTableMeta(DBNAME, TBNAME, false, entryPosition); + assert result.getDdl().equalsIgnoreCase(DDL); + } +} diff --git a/pom.xml b/pom.xml index f48a488a..e9c2623b 100644 --- a/pom.xml +++ b/pom.xml @@ -254,7 +254,7 @@ com.alibaba.fastsql fastsql - 2.0.0_preview_366 + 2.0.0_preview_186 com.alibaba