support otter use tableMetaTSDB

dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java

@@ -490,7 +490,6 @@ public final class RowsLogBuffer {
                     // t % 100);
 
                     StringBuilder builder = new StringBuilder();
-                    builder.append(26);
                     appendNumber4(builder, d / 10000);
                     builder.append('-');
                     appendNumber2(builder, (d % 10000) / 100);
@@ -615,7 +614,13 @@ public final class RowsLogBuffer {
                     if (i32 < 0) {
                         builder.append('-');
                     }
-                    appendNumber2(builder, u32 / 10000);
+
+                    int d = u32 / 10000;
+                    if (d > 100) {
+                        builder.append(String.valueOf(d));
+                    } else {
+                        appendNumber2(builder, d);
+                    }
                     builder.append(':');
                     appendNumber2(builder, (u32 % 10000) / 100);
                     builder.append(':');
@@ -724,7 +729,13 @@ public final class RowsLogBuffer {
                     if (ltime < 0) {
                         builder.append('-');
                     }
-                    appendNumber2(builder, (int) ((intpart >> 12) % (1 << 10)));
+
+                    int d = (int) ((intpart >> 12) % (1 << 10));
+                    if (d > 100) {
+                        builder.append(String.valueOf(d));
+                    } else {
+                        appendNumber2(builder, d);
+                    }
                     builder.append(':');
                     appendNumber2(builder, (int) ((intpart >> 6) % (1 << 6)));
                     builder.append(':');
@@ -1134,7 +1145,7 @@ public final class RowsLogBuffer {
                 .append(digits[(d / 100) % 10])
                 .append(digits[(d / 10) % 10])
                 .append(digits[d % 10]);
-        } else if (d >= 100) {
+        } else {
             builder.append('0');
             appendNumber3(builder, d);
         }
@@ -1142,9 +1153,7 @@ public final class RowsLogBuffer {
 
     private void appendNumber3(StringBuilder builder, int d) {
         if (d >= 100) {
-            builder.append(digits[d / 100])
+            builder.append(digits[d / 100]).append(digits[(d / 10) % 10]).append(digits[d % 10]);
-                .append(digits[(d / 10) % 10])
-                .append(digits[d % 10]);
         } else {
             builder.append('0');
             appendNumber2(builder, d);

instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java

@@ -6,11 +6,6 @@ import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 
-
-import com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.HistoryTableMetaCache;
-
-import com.alibaba.otter.canal.meta.FileMixedMetaManager;
-
 import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -25,7 +20,14 @@ import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter;
 import com.alibaba.otter.canal.instance.core.AbstractCanalInstance;
 import com.alibaba.otter.canal.instance.manager.model.Canal;
 import com.alibaba.otter.canal.instance.manager.model.CanalParameter;
-import com.alibaba.otter.canal.instance.manager.model.CanalParameter.*;
+import com.alibaba.otter.canal.instance.manager.model.CanalParameter.DataSourcing;
+import com.alibaba.otter.canal.instance.manager.model.CanalParameter.HAMode;
+import com.alibaba.otter.canal.instance.manager.model.CanalParameter.IndexMode;
+import com.alibaba.otter.canal.instance.manager.model.CanalParameter.MetaMode;
+import com.alibaba.otter.canal.instance.manager.model.CanalParameter.SourcingType;
+import com.alibaba.otter.canal.instance.manager.model.CanalParameter.StorageMode;
+import com.alibaba.otter.canal.instance.manager.model.CanalParameter.StorageScavengeMode;
+import com.alibaba.otter.canal.meta.FileMixedMetaManager;
 import com.alibaba.otter.canal.meta.MemoryMetaManager;
 import com.alibaba.otter.canal.meta.PeriodMixedMetaManager;
 import com.alibaba.otter.canal.meta.ZooKeeperMetaManager;
@@ -36,7 +38,12 @@ import com.alibaba.otter.canal.parse.inbound.AbstractEventParser;
 import com.alibaba.otter.canal.parse.inbound.group.GroupEventParser;
 import com.alibaba.otter.canal.parse.inbound.mysql.LocalBinlogEventParser;
 import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser;
-import com.alibaba.otter.canal.parse.index.*;
+import com.alibaba.otter.canal.parse.index.CanalLogPositionManager;
+import com.alibaba.otter.canal.parse.index.FailbackLogPositionManager;
+import com.alibaba.otter.canal.parse.index.MemoryLogPositionManager;
+import com.alibaba.otter.canal.parse.index.MetaLogPositionManager;
+import com.alibaba.otter.canal.parse.index.PeriodMixedLogPositionManager;
+import com.alibaba.otter.canal.parse.index.ZooKeeperLogPositionManager;
 import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
 import com.alibaba.otter.canal.protocol.position.EntryPosition;
 import com.alibaba.otter.canal.sink.entry.EntryEventSink;
@@ -114,7 +121,7 @@ public class CanalInstanceWithManager extends AbstractCanalInstance {
             ZooKeeperMetaManager zooKeeperMetaManager = new ZooKeeperMetaManager();
             zooKeeperMetaManager.setZkClientx(getZkclientx());
             ((PeriodMixedMetaManager) metaManager).setZooKeeperMetaManager(zooKeeperMetaManager);
-        } else if (mode.isLocalFile()){
+        } else if (mode.isLocalFile()) {
             FileMixedMetaManager fileMixedMetaManager = new FileMixedMetaManager();
             fileMixedMetaManager.setDataDir(parameters.getDataDir());
             fileMixedMetaManager.setPeriod(parameters.getMetaFileFlushPeriod());
@@ -241,13 +248,6 @@ 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(),

instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/model/CanalParameter.java

@@ -5,7 +5,6 @@ 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;
 
@@ -94,6 +93,10 @@ public class CanalParameter implements Serializable {
     private Boolean                  filterTableError                   = Boolean.FALSE;             // 是否忽略表解析异常
     private String                   blackFilter                        = null;                      // 匹配黑名单,忽略解析
 
+    private Boolean                  tsdbEnable                         = Boolean.FALSE;             // 是否开启tableMetaTSDB
+    private String                   tsdbJdbcUrl;
+    private String                   tsdbJdbcUserName;
+    private String                   tsdbJdbcPassword;
     // ================================== 兼容字段处理
     private InetSocketAddress        masterAddress;                                                  // 主库信息
     private String                   masterUsername;                                                 // 帐号
@@ -109,9 +112,6 @@ public class CanalParameter implements Serializable {
     private Long                     standbyLogfileOffest               = null;
     private Long                     standbyTimestamp                   = null;
 
-    // Ctrip Table Meta
-    TableMetaStorageFactory tableMetaStorageFactory;
-
     public static enum RunMode {
 
         /** 嵌入式 */
@@ -250,7 +250,7 @@ public class CanalParameter implements Serializable {
         ZOOKEEPER,
         /** 混合模式，内存+文件 */
         MIXED,
-        /** 本地文件存储模式*/
+        /** 本地文件存储模式 */
         LOCAL_FILE;
 
         public boolean isMemory() {
@@ -265,7 +265,7 @@ public class CanalParameter implements Serializable {
             return this.equals(MetaMode.MIXED);
         }
 
-        public boolean isLocalFile(){
+        public boolean isLocalFile() {
             return this.equals(MetaMode.LOCAL_FILE);
         }
     }
@@ -887,12 +887,36 @@ public class CanalParameter implements Serializable {
         this.blackFilter = blackFilter;
     }
 
-    public TableMetaStorageFactory getTableMetaStorageFactory() {
+    public Boolean getTsdbEnable() {
-        return tableMetaStorageFactory;
+        return tsdbEnable;
     }
 
-    public void setTableMetaStorageFactory(TableMetaStorageFactory tableMetaStorageFactory) {
+    public void setTsdbEnable(Boolean tsdbEnable) {
-        this.tableMetaStorageFactory = tableMetaStorageFactory;
+        this.tsdbEnable = tsdbEnable;
+    }
+
+    public String getTsdbJdbcUrl() {
+        return tsdbJdbcUrl;
+    }
+
+    public void setTsdbJdbcUrl(String tsdbJdbcUrl) {
+        this.tsdbJdbcUrl = tsdbJdbcUrl;
+    }
+
+    public String getTsdbJdbcUserName() {
+        return tsdbJdbcUserName;
+    }
+
+    public void setTsdbJdbcUserName(String tsdbJdbcUserName) {
+        this.tsdbJdbcUserName = tsdbJdbcUserName;
+    }
+
+    public String getTsdbJdbcPassword() {
+        return tsdbJdbcPassword;
+    }
+
+    public void setTsdbJdbcPassword(String tsdbJdbcPassword) {
+        this.tsdbJdbcPassword = tsdbJdbcPassword;
     }
 
     public String toString() {

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

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

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

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

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

@@ -10,9 +10,6 @@ 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;
@@ -90,10 +87,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;
     private boolean                     filterQueryDml      = false;
@@ -268,8 +262,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
             if (!isSeek) {
                 // 使用新的表结构元数据管理方式
                 EntryPosition position = createPosition(event.getHeader());
-                String fulltbName = schemaName+"."+tableName;
+                tableMetaCache.apply(position, event.getDbName(), queryString, null);
-                tableMetaCache.apply(position, fulltbName, queryString, null);
             }
 
             Header header = createHeader(event.getHeader(), schemaName, tableName, type);
@@ -548,12 +541,16 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
                     tableError |= parseOneRow(rowDataBuilder, event, buffer, changeColumns, true, tableMeta);
                 }
 
-                rowsCount ++;
+                rowsCount++;
                 rowChangeBuider.addRowDatas(rowDataBuilder.build());
             }
 
             TableMapLogEvent table = event.getTable();
-            Header header = createHeader(event.getHeader(), table.getDbName(), table.getTableName(), eventType, rowsCount);
+            Header header = createHeader(event.getHeader(),
+                table.getDbName(),
+                table.getTableName(),
+                eventType,
+                rowsCount);
 
             RowChange rowChange = rowChangeBuider.build();
             if (tableError) {
@@ -808,12 +805,12 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
         return createEntry(header, EntryType.ROWDATA, rowChangeBuider.build().toByteString());
     }
 
-
     private Header createHeader(LogHeader logHeader, String schemaName, String tableName, EventType eventType) {
         return createHeader(logHeader, schemaName, tableName, eventType, -1);
     }
 
-    private Header createHeader(LogHeader logHeader, String schemaName, String tableName, EventType eventType, Integer rowsCount) {
+    private Header createHeader(LogHeader logHeader, String schemaName, String tableName, EventType eventType,
+                                Integer rowsCount) {
         // header会做信息冗余,方便以后做检索或者过滤
         Header.Builder headerBuilder = Header.newBuilder();
         headerBuilder.setVersion(version);
@@ -940,7 +937,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
         this.nameBlackFilter = nameBlackFilter;
     }
 
-    public void setTableMetaCache(TableMetaCacheInterface tableMetaCache) {
+    public void setTableMetaCache(TableMetaCache tableMetaCache) {
         this.tableMetaCache = tableMetaCache;
     }
 

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java

@@ -6,7 +6,6 @@ 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;
@@ -30,7 +29,7 @@ import com.google.common.cache.LoadingCache;
  * @author jianghang 2013-1-17 下午10:15:16
  * @version 1.0.0
  */
-public class TableMetaCache implements TableMetaCacheInterface {
+public class TableMetaCache {
 
     public static final String              COLUMN_NAME    = "COLUMN_NAME";
     public static final String              COLUMN_TYPE    = "COLUMN_TYPE";
@@ -100,10 +99,6 @@ public class TableMetaCache implements TableMetaCacheInterface {
             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 {

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/HistoryTableMetaCache.java

@@ -1,191 +0,0 @@
-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<String, Map<Long, TableMeta>> cache; // 第一层：数据库名.表名，第二层时间戳，TableMeta
-
-    public HistoryTableMetaCache() {
-        cache = CacheBuilder.newBuilder().build(new CacheLoader<String, Map<Long, TableMeta>>() {
-            @Override
-            public Map<Long, TableMeta> 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<Long, TableMeta> tableMetaMap = cache.getUnchecked(tableName);
-                    if (tableMetaMap.isEmpty()) {
-                        put(schema, tableName, currentDdl, timestamp - 1000L); // 放入当前schema，取时间为当前时间-1s
-                    } else {                                               // 如果table存在历史
-                        Iterator<Long> 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<TableMetaEntry> 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<String> fields = new ArrayList<String>();
-            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<Long, TableMeta> tableMetaMap;
-        if (!cache.asMap().containsKey(table)) {
-            tableMetaMap = new TreeMap<Long, TableMeta>(new Comparator<Long>() {
-                @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<Long, TableMeta> tableMetaMap = cache.getUnchecked(table);
-        Iterator<Long> 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<Long, TableMeta> tableMetaMap) {
-        int MAX_CAPABILITY = 20;
-        if (tableMetaMap.keySet().size() < MAX_CAPABILITY) {
-            return;
-        }
-        Iterator<Long> 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;
-    }
-}

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/TableMetaCacheInterface.java

@@ -1,20 +0,0 @@
-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();
-
-}

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/TableMetaCacheWithStorage.java

@@ -1,105 +0,0 @@
-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;
-    }
-}

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/TableMetaEntry.java

@@ -1,55 +0,0 @@
-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;
-    }
-}

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/TableMetaStorage.java

@@ -1,18 +0,0 @@
-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<TableMetaEntry> fetch();
-
-    List<TableMetaEntry> fetchByTableName(String tableName);
-
-    String getDbName();
-
-    String getDbAddress();
-
-    void setDbAddress(String address);
-}

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/TableMetaStorageFactory.java

@@ -1,9 +0,0 @@
-package com.alibaba.otter.canal.parse.inbound.mysql.tablemeta;
-
-public interface TableMetaStorageFactory {
-
-    TableMetaStorage getTableMetaStorage();
-
-    String getDbName();
-
-}

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/exception/CacheConnectionNull.java

@@ -1,9 +0,0 @@
-package com.alibaba.otter.canal.parse.inbound.mysql.tablemeta.exception;
-
-public class CacheConnectionNull extends Exception{
-
-    @Override
-    public String toString() {
-        return "CacheConnectionNull";
-    }
-}

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/exception/NoHistoryException.java

@@ -1,21 +0,0 @@
-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;
-    }
-}

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/impl/mysql/MySqlTableMetaCallback.java

@@ -1,14 +0,0 @@
-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<TableMetaEntry> fetch(String dbAddress, String dbName);
-
-    List<TableMetaEntry> fetch(String dbAddress, String dbName, String tableName);
-}

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/impl/mysql/MySqlTableMetaStorage.java

@@ -1,48 +0,0 @@
-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<TableMetaEntry> fetch() {
-        return mySqlTableMetaCallback.fetch(dbAddress, dbName);
-    }
-
-    @Override
-    public List<TableMetaEntry> 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;
-    }
-}

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tablemeta/impl/mysql/MySqlTableMetaStorageFactory.java

@@ -1,26 +0,0 @@
-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;
-    }
-
-}

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DefaultTableMetaTSDBFactory.java

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

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

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

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaTSDBFactory.java

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