From 19707f92d4e756585488b2fc7613ff424f7f0baa Mon Sep 17 00:00:00 2001 From: agapple Date: Sat, 30 Sep 2017 23:21:17 +0800 Subject: [PATCH] merge tsdb meta support and refactor --- .../tddl/dbsync/binlog/BinlogPosition.java | 12 +- .../resources/example/instance.properties | 9 +- .../main/resources/spring/file-instance.xml | 6 +- .../resources/spring/{ => tsdb}/dal-dao.xml | 5 +- .../spring/{ => tsdb}/mybatis-config.xml | 0 .../spring/tsdb}/sql-map/sqlmap_history.xml | 6 - .../{ => tsdb}/sql-map/sqlmap_snapshot.xml | 0 .../spring/{ => tsdb}/sqlmap-config.xml | 0 .../parse/driver/mysql/MysqlConnector.java | 2 +- .../driver/mysql/MysqlQueryExecutor.java | 57 ++- .../client/ClientAuthenticationPacket.java | 7 +- .../parse/inbound/AbstractEventParser.java | 6 +- .../canal/parse/inbound/BinlogParser.java | 2 +- .../otter/canal/parse/inbound/TableMeta.java | 24 +- .../mysql/AbstractMysqlEventParser.java | 8 +- .../inbound/mysql/LocalBinlogEventParser.java | 2 +- .../parse/inbound/mysql/MysqlConnection.java | 74 +--- .../parse/inbound/mysql/MysqlEventParser.java | 20 +- .../inbound/mysql/dbsync/DBMSAction.java | 77 ---- .../inbound/mysql/dbsync/LogEventConvert.java | 210 ++++++---- .../inbound/mysql/dbsync/TableMetaCache.java | 165 ++++---- .../parse/inbound/mysql/ddl/DdlResult.java | 39 +- .../inbound/mysql/ddl/DruidDdlParser.java | 34 +- .../{dbsync => ddl}/SimpleDdlParser.java | 100 +---- .../mysql/tsdb/DataSourceFactoryTSDB.java | 52 ++- .../inbound/mysql/tsdb/MemoryTableMeta.java | 6 +- .../inbound/mysql/tsdb/TableMetaManager.java | 390 +++++++----------- .../mysql/tsdb/dao/MetaHistoryDAO.java | 11 +- .../mysql/tsdb/dao/MetaSnapshotDAO.java | 10 +- .../mysql/tsdb/model/MetaHistoryDO.java | 10 +- .../mysql/tsdb/model/MetaSnapshotDO.java | 10 +- .../index/FileMixedLogPositionManager.java | 1 + .../index/PeriodMixedLogPositionManager.java | 1 + .../parse/inbound/TableMetaCacheTest.java | 53 ++- .../inbound/group/GroupEventPaserTest.java | 1 - .../inbound/mysql/SimpleDdlParserTest.java | 4 +- .../mysql/tsdb/MemoryTableMetaTest.java | 4 +- .../mysql/tsdb/MetaHistoryDAOTest.java | 9 +- .../mysql/tsdb/TableMetaManagerTest.java | 2 - parse/src/test/resources/dal-dao.xml | 46 --- .../resources/tsdb}/dal-dao.xml | 9 +- .../resources/tsdb}/mybatis-config.xml | 0 .../tsdb}/sql-map/sqlmap_history.xml | 0 .../tsdb}/sql-map/sqlmap_snapshot.xml | 0 .../resources/tsdb}/sqlmap-config.xml | 0 45 files changed, 585 insertions(+), 899 deletions(-) rename deployer/src/main/resources/spring/{ => tsdb}/dal-dao.xml (89%) rename deployer/src/main/resources/spring/{ => tsdb}/mybatis-config.xml (100%) rename {parse/src/main/resources => deployer/src/main/resources/spring/tsdb}/sql-map/sqlmap_history.xml (99%) rename deployer/src/main/resources/spring/{ => tsdb}/sql-map/sqlmap_snapshot.xml (100%) rename deployer/src/main/resources/spring/{ => tsdb}/sqlmap-config.xml (100%) delete mode 100644 parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DBMSAction.java rename parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/{dbsync => ddl}/SimpleDdlParser.java (73%) delete mode 100644 parse/src/test/resources/dal-dao.xml rename parse/src/{main/resources => test/resources/tsdb}/dal-dao.xml (83%) rename parse/src/{main/resources => test/resources/tsdb}/mybatis-config.xml (100%) rename {deployer/src/main/resources/spring => parse/src/test/resources/tsdb}/sql-map/sqlmap_history.xml (100%) rename parse/src/{main/resources => test/resources/tsdb}/sql-map/sqlmap_snapshot.xml (100%) rename parse/src/{main/resources => test/resources/tsdb}/sqlmap-config.xml (100%) diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/BinlogPosition.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/BinlogPosition.java index 92bba817..a602c05f 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/BinlogPosition.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/BinlogPosition.java @@ -1,6 +1,5 @@ package com.taobao.tddl.dbsync.binlog; -import com.taobao.tddl.dbsync.binlog.LogPosition; /** * Position inside binlog file @@ -35,9 +34,8 @@ public class BinlogPosition extends LogPosition { } private final static long[] pow10 = { 1, 10, 100, 1000, 10000, 100000, 1000000, 10000000, 100000000, 1000000000, - 10000000000L, 100000000000L, 1000000000000L, 10000000000000L, - 100000000000000L, 1000000000000000L, 10000000000000000L, 100000000000000000L, - 1000000000000000000L }; + 10000000000L, 100000000000L, 1000000000000L, 10000000000000L, 100000000000000L, 1000000000000000L, + 10000000000000000L, 100000000000000000L, 1000000000000000000L }; public static String placeHolder(int bit, long number) { if (bit > 18) { @@ -53,9 +51,6 @@ public class BinlogPosition extends LogPosition { return String.valueOf(max + number).substring(1); } - /** - * Return BinlogPosition in String representation. This serves as EventId for DBMSEvent. - */ public String format2String(final int positionMaxLen) { String binlogSuffix = fileName; String binlogOffset = placeHolder((int) positionMaxLen, position); @@ -90,7 +85,8 @@ public class BinlogPosition extends LogPosition { if (sharpIndex != -1) { binlogPosition = Long.parseLong(source.substring(miscIndex, sharpIndex)); } else if (semicolonIndex != -1) { - binlogPosition = Long.parseLong(source.substring(miscIndex, semicolonIndex)); // NOTE: 向后兼容 + binlogPosition = Long.parseLong(source.substring(miscIndex, semicolonIndex)); // NOTE: + // 向后兼容 } else if (dotIndex != -1) { binlogPosition = Long.parseLong(source.substring(miscIndex, dotIndex)); } else { diff --git a/deployer/src/main/resources/example/instance.properties b/deployer/src/main/resources/example/instance.properties index 3c5e0a8e..7f00f079 100644 --- a/deployer/src/main/resources/example/instance.properties +++ b/deployer/src/main/resources/example/instance.properties @@ -9,15 +9,10 @@ canal.instance.master.timestamp=1506088042 # tsdb info -canal.instance.tsdb.address=127.0.0.1:3306 canal.instance.tsdb.enable=true +canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/tsdb canal.instance.tsdb.dbUsername=canal canal.instance.tsdb.dbPassword=canal -# tsdb关联的目标db，和canal.instance.defaultDatabaseName意义相同 -canal.instance.tsdb.defaultDatabaseName=test -canal.instance.tsdb.connectionCharset=UTF-8 -canal.instance.tsdb.driverClassName=com.mysql.jdbc.Driver -canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/tsdb #canal.instance.standby.address = @@ -31,6 +26,6 @@ canal.instance.defaultDatabaseName=test canal.instance.connectionCharset=UTF-8 # table regex canal.instance.filter.regex=.*\\..* -# table black regex , 默认过滤表结构管理的数据 +# table black regex canal.instance.filter.black.regex=tsdb\\..* ################################################# \ No newline at end of file diff --git a/deployer/src/main/resources/spring/file-instance.xml b/deployer/src/main/resources/spring/file-instance.xml index a2563119..a9f2823b 100644 --- a/deployer/src/main/resources/spring/file-instance.xml +++ b/deployer/src/main/resources/spring/file-instance.xml @@ -180,13 +180,10 @@ - + - - - @@ -210,5 +207,4 @@ - diff --git a/deployer/src/main/resources/spring/dal-dao.xml b/deployer/src/main/resources/spring/tsdb/dal-dao.xml similarity index 89% rename from deployer/src/main/resources/spring/dal-dao.xml rename to deployer/src/main/resources/spring/tsdb/dal-dao.xml index 644979fc..1842641f 100644 --- a/deployer/src/main/resources/spring/dal-dao.xml +++ b/deployer/src/main/resources/spring/tsdb/dal-dao.xml @@ -9,13 +9,10 @@ - + - - - diff --git a/deployer/src/main/resources/spring/mybatis-config.xml b/deployer/src/main/resources/spring/tsdb/mybatis-config.xml similarity index 100% rename from deployer/src/main/resources/spring/mybatis-config.xml rename to deployer/src/main/resources/spring/tsdb/mybatis-config.xml diff --git a/parse/src/main/resources/sql-map/sqlmap_history.xml b/deployer/src/main/resources/spring/tsdb/sql-map/sqlmap_history.xml similarity index 99% rename from parse/src/main/resources/sql-map/sqlmap_history.xml rename to deployer/src/main/resources/spring/tsdb/sql-map/sqlmap_history.xml index 00197194..a983e222 100644 --- a/parse/src/main/resources/sql-map/sqlmap_history.xml +++ b/deployer/src/main/resources/spring/tsdb/sql-map/sqlmap_history.xml @@ -6,18 +6,14 @@ @@ -42,10 +38,8 @@ diff --git a/deployer/src/main/resources/spring/sql-map/sqlmap_snapshot.xml b/deployer/src/main/resources/spring/tsdb/sql-map/sqlmap_snapshot.xml similarity index 100% rename from deployer/src/main/resources/spring/sql-map/sqlmap_snapshot.xml rename to deployer/src/main/resources/spring/tsdb/sql-map/sqlmap_snapshot.xml diff --git a/deployer/src/main/resources/spring/sqlmap-config.xml b/deployer/src/main/resources/spring/tsdb/sqlmap-config.xml similarity index 100% rename from deployer/src/main/resources/spring/sqlmap-config.xml rename to deployer/src/main/resources/spring/tsdb/sqlmap-config.xml diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnector.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnector.java index 49f6edea..0e488162 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnector.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnector.java @@ -34,7 +34,7 @@ public class MysqlConnector { private byte charsetNumber = 33; private String defaultSchema = "retl"; private int soTimeout = 30 * 1000; - private int connTimeout = 5 * 1000; + private int connTimeout = 5 * 1000; private int receiveBufferSize = 16 * 1024; private int sendBufferSize = 16 * 1024; diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlQueryExecutor.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlQueryExecutor.java index a389623e..6d30bd4f 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlQueryExecutor.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlQueryExecutor.java @@ -6,6 +6,7 @@ import java.util.List; import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.client.QueryCommandPacket; +import com.alibaba.otter.canal.parse.driver.mysql.packets.server.EOFPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ErrorPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.server.FieldPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ResultSetHeaderPacket; @@ -93,11 +94,65 @@ public class MysqlQueryExecutor { return resultSet; } - private void readEofPacket() throws IOException { + public List queryMulti(String queryString) throws IOException { + QueryCommandPacket cmd = new QueryCommandPacket(); + cmd.setQueryString(queryString); + byte[] bodyBytes = cmd.toBytes(); + PacketManager.writeBody(channel, bodyBytes); + List resultSets = new ArrayList(); + boolean moreResult = true; + while (moreResult) { + byte[] body = readNextPacket(); + if (body[0] < 0) { + ErrorPacket packet = new ErrorPacket(); + packet.fromBytes(body); + throw new IOException(packet + "\n with command: " + queryString); + } + + ResultSetHeaderPacket rsHeader = new ResultSetHeaderPacket(); + rsHeader.fromBytes(body); + + List fields = new ArrayList(); + for (int i = 0; i < rsHeader.getColumnCount(); i++) { + FieldPacket fp = new FieldPacket(); + fp.fromBytes(readNextPacket()); + fields.add(fp); + } + + moreResult = readEofPacket(); + + List rowData = new ArrayList(); + while (true) { + body = readNextPacket(); + if (body[0] == -2) { + break; + } + RowDataPacket rowDataPacket = new RowDataPacket(); + rowDataPacket.fromBytes(body); + rowData.add(rowDataPacket); + } + + ResultSetPacket resultSet = new ResultSetPacket(); + resultSet.getFieldDescriptors().addAll(fields); + for (RowDataPacket r : rowData) { + resultSet.getFieldValues().addAll(r.getColumns()); + } + resultSet.setSourceAddress(channel.getRemoteSocketAddress()); + resultSets.add(resultSet); + } + + return resultSets; + } + + private boolean readEofPacket() throws IOException { byte[] eofBody = readNextPacket(); + EOFPacket packet = new EOFPacket(); + packet.fromBytes(eofBody); if (eofBody[0] != -2) { throw new IOException("EOF Packet is expected, but packet with field_count=" + eofBody[0] + " is found."); } + + return (packet.statusFlag & 0x0008) != 0; } protected byte[] readNextPacket() throws IOException { diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/ClientAuthenticationPacket.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/ClientAuthenticationPacket.java index 04975a31..fb04d7dc 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/ClientAuthenticationPacket.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/client/ClientAuthenticationPacket.java @@ -47,10 +47,11 @@ public class ClientAuthenticationPacket extends PacketWithHeaderPacket { /** * CLIENT_LONG_PASSWORD CLIENT_LONG_FLAG CLIENT_PROTOCOL_41 * CLIENT_INTERACTIVE CLIENT_TRANSACTIONS CLIENT_SECURE_CONNECTION + * CLIENT_MULTI_STATEMENTS; */ - ByteHelper.writeUnsignedIntLittleEndian(1 | 4 | 512 | 8192 | 32768, out); // remove - // client_interactive - // feature + ByteHelper.writeUnsignedIntLittleEndian(1 | 4 | 512 | 8192 | 32768 | 0x00010000, out); // remove + // client_interactive + // feature // 2. write max_packet_size ByteHelper.writeUnsignedIntLittleEndian(MSC.MAX_PACKET_LENGTH, out); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java index b5e1b090..32d3aee5 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java @@ -173,7 +173,7 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle public boolean sink(EVENT event) { try { - CanalEntry.Entry entry = parseAndProfilingIfNecessary(event,false); + CanalEntry.Entry entry = parseAndProfilingIfNecessary(event, false); if (!running) { return false; @@ -320,13 +320,13 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle return result; } - protected CanalEntry.Entry parseAndProfilingIfNecessary(EVENT bod,boolean isSeek) throws Exception { + protected CanalEntry.Entry parseAndProfilingIfNecessary(EVENT bod, boolean isSeek) throws Exception { long startTs = -1; boolean enabled = getProfilingEnabled(); if (enabled) { startTs = System.currentTimeMillis(); } - CanalEntry.Entry event = binlogParser.parse(bod,isSeek); + CanalEntry.Entry event = binlogParser.parse(bod, isSeek); if (enabled) { this.parsingInterval = System.currentTimeMillis() - startTs; } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/BinlogParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/BinlogParser.java index 01c55a51..034879e0 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/BinlogParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/BinlogParser.java @@ -11,7 +11,7 @@ import com.alibaba.otter.canal.protocol.CanalEntry; */ public interface BinlogParser extends CanalLifeCycle { - CanalEntry.Entry parse(T event,boolean isSeek) throws CanalParseException; + CanalEntry.Entry parse(T event, boolean isSeek) throws CanalParseException; void reset(); } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/TableMeta.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/TableMeta.java index d3009624..7f56e440 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/TableMeta.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/TableMeta.java @@ -20,16 +20,16 @@ import org.apache.commons.lang.StringUtils; */ public class TableMeta { - private String schema; - private String table; + private String schema; + private String table; private List fields = new ArrayList(); - private String ddl; // 表结构的DDL语句 + private String ddl; // 表结构的DDL语句 - public TableMeta() { + public TableMeta(){ } - public TableMeta(String schema, String table, List fields) { + public TableMeta(String schema, String table, List fields){ this.schema = schema; this.table = table; this.fields = fields; @@ -109,11 +109,11 @@ public class TableMeta { public static class FieldMeta { - public FieldMeta() { + public FieldMeta(){ } - public FieldMeta(String columnName, String columnType, boolean nullable, boolean key, String defaultValue) { + public FieldMeta(String columnName, String columnType, boolean nullable, boolean key, String defaultValue){ this.columnName = columnName; this.columnType = columnType; this.nullable = nullable; @@ -121,12 +121,12 @@ public class TableMeta { this.defaultValue = defaultValue; } - private String columnName; - private String columnType; + private String columnName; + private String columnType; private boolean nullable; private boolean key; - private String defaultValue; - private String extra; + private String defaultValue; + private String extra; public String getColumnName() { return columnName; @@ -182,7 +182,7 @@ public class TableMeta { public String toString() { return "FieldMeta [columnName=" + columnName + ", columnType=" + columnType + ", defaultValue=" - + defaultValue + ", nullable=" + nullable + ", key=" + key + "]"; + + defaultValue + ", nullable=" + nullable + ", key=" + key + "]"; } } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java index 4510159b..094d0b1d 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java @@ -30,8 +30,8 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { protected boolean filterTableError = false; @Resource - protected TableMetaManager tableMetaManager; - protected boolean useDruidDdlFilter = true; + protected TableMetaManager tableMetaManager; + protected boolean useDruidDdlFilter = true; protected BinlogParser buildParser() { LogEventConvert convert = new LogEventConvert(); @@ -50,7 +50,7 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { convert.setFilterRows(filterRows); convert.setFilterTableError(filterTableError); - //初始化parser的时候也初始化管理mysql 表结构的管理器 + // 初始化parser的时候也初始化管理mysql 表结构的管理器 tableMetaManager.init(); return convert; } @@ -66,6 +66,7 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { /** * 回滚到指定位点 + * * @param position * @return */ @@ -77,7 +78,6 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { return true; } - public void setEventBlackFilter(CanalEventFilter eventBlackFilter) { super.setEventBlackFilter(eventBlackFilter); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java index 87f31208..d45c5615 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinlogEventParser.java @@ -50,7 +50,7 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements throw new CanalParseException(e); } - tableMetaCache = new TableMetaCache(metaConnection,tableMetaManager); + tableMetaCache = new TableMetaCache(metaConnection, tableMetaManager); ((LogEventConvert) binlogParser).setTableMetaCache(tableMetaCache); } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java index e399a356..94625b10 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java @@ -3,14 +3,8 @@ package com.alibaba.otter.canal.parse.inbound.mysql; import java.io.IOException; import java.net.InetSocketAddress; import java.nio.charset.Charset; -import java.sql.Connection; -import java.sql.ResultSet; -import java.sql.SQLException; -import java.sql.Statement; import java.util.List; -import java.util.Properties; -import com.mysql.jdbc.Driver; import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -27,28 +21,24 @@ import com.alibaba.otter.canal.parse.inbound.ErosaConnection; import com.alibaba.otter.canal.parse.inbound.SinkFunction; import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.DirectLogFetcher; import com.alibaba.otter.canal.parse.support.AuthenticationInfo; - import com.taobao.tddl.dbsync.binlog.LogContext; import com.taobao.tddl.dbsync.binlog.LogDecoder; import com.taobao.tddl.dbsync.binlog.LogEvent; public class MysqlConnection implements ErosaConnection { - private static final Logger logger = LoggerFactory.getLogger(MysqlConnection.class); + private static final Logger logger = LoggerFactory.getLogger(MysqlConnection.class); private MysqlConnector connector; private long slaveId; - private Charset charset = Charset.forName("UTF-8"); + private Charset charset = Charset.forName("UTF-8"); private BinlogFormat binlogFormat; private BinlogImage binlogImage; // tsdb releated - private AuthenticationInfo authInfo; - private Connection conn; - protected int connTimeout = 5 * 1000; // 5秒 - protected int soTimeout = 60 * 60 * 1000; // 1小时 - protected int bufferSize = 16 * 1024; - + private AuthenticationInfo authInfo; + protected int connTimeout = 5 * 1000; // 5秒 + protected int soTimeout = 60 * 60 * 1000; // 1小时 public MysqlConnection(){ } @@ -59,7 +49,7 @@ public class MysqlConnection implements ErosaConnection { authInfo.setUsername(username); authInfo.setPassword(password); connector = new MysqlConnector(address, username, password); - //将connection里面的参数透传下 + // 将connection里面的参数透传下 connector.setSoTimeout(soTimeout); connector.setConnTimeout(connTimeout); } @@ -72,29 +62,13 @@ public class MysqlConnection implements ErosaConnection { authInfo.setPassword(password); authInfo.setDefaultDatabaseName(defaultSchema); connector = new MysqlConnector(address, username, password, charsetNumber, defaultSchema); - //将connection里面的参数透传下 + // 将connection里面的参数透传下 connector.setSoTimeout(soTimeout); connector.setConnTimeout(connTimeout); } public void connect() throws IOException { connector.connect(); - //准备一个connection连接给tsdb查询用 - Properties info = new Properties(); - info.put("user", authInfo.getUsername()); - info.put("password", authInfo.getPassword()); - info.put("connectTimeout", String.valueOf(connTimeout)); - info.put("socketTimeout", String.valueOf(soTimeout)); - String url = "jdbc:mysql://" + authInfo.getAddress().getHostName() + ":" - + String.valueOf(authInfo.getAddress().getPort()) + "?allowMultiQueries=true"; - try { - Driver driver = new com.mysql.jdbc.Driver(); - conn = driver.connect(url, info); - // conn = DriverManager.getConnection(url, info); - } catch (SQLException e) { - throw new CanalParseException(e); - } - } public void reconnect() throws IOException { @@ -114,23 +88,9 @@ public class MysqlConnection implements ErosaConnection { return exector.query(cmd); } - public T query(String sql, ProcessJdbcResult processor) { - Statement stmt = null; - try { - stmt = conn.createStatement(); - ResultSet rs = stmt.executeQuery(sql); - return processor.process(rs); - } catch (SQLException e) { - throw new CanalParseException(e); - } finally { - if (stmt != null) { - try { - stmt.close(); - } catch (SQLException e) { - // ignore - } - } - } + public List queryMulti(String cmd) throws IOException { + MysqlQueryExecutor exector = new MysqlQueryExecutor(connector); + return exector.queryMulti(cmd); } public void update(String cmd) throws IOException { @@ -212,7 +172,7 @@ public class MysqlConnection implements ErosaConnection { connection.setCharset(getCharset()); connection.setSlaveId(getSlaveId()); connection.setConnector(connector.fork()); - //set authInfo + // set authInfo connection.setAuthInfo(authInfo); return connection; } @@ -449,9 +409,6 @@ public class MysqlConnection implements ErosaConnection { return binlogImage; } - public Connection getConn() { - return conn; - } public InetSocketAddress getAddress() { return authInfo.getAddress(); } @@ -464,10 +421,6 @@ public class MysqlConnection implements ErosaConnection { this.soTimeout = soTimeout; } - public void setBufferSize(int bufferSize) { - this.bufferSize = bufferSize; - } - public AuthenticationInfo getAuthInfo() { return authInfo; } @@ -475,9 +428,4 @@ public class MysqlConnection implements ErosaConnection { public void setAuthInfo(AuthenticationInfo authInfo) { this.authInfo = authInfo; } - - public static interface ProcessJdbcResult { - - public T process(ResultSet rs) throws SQLException; - } } 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 e6dedbdb..4a188b01 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 @@ -68,7 +68,6 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE // update by yishun.chen,特殊异常处理参数 private int dumpErrorCount = 0; // binlogDump失败异常计数 - private int dumpTimeoutCount = 0;// socketTimeout异常 private int dumpErrorCountThreshold = 2; // binlogDump失败异常计数阀值 protected ErosaConnection buildErosaConnection() { @@ -118,12 +117,12 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE } } - if(tableMetaManager != null){ + if (tableMetaManager != null) { tableMetaManager.setConnection(metaConnection); tableMetaManager.setFilter(eventFilter); } - tableMetaCache = new TableMetaCache(metaConnection,tableMetaManager); + tableMetaCache = new TableMetaCache(metaConnection, tableMetaManager); ((LogEventConvert) binlogParser).setTableMetaCache(tableMetaCache); } } @@ -443,7 +442,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE public boolean sink(LogEvent event) { try { - CanalEntry.Entry entry = parseAndProfilingIfNecessary(event,true); + CanalEntry.Entry entry = parseAndProfilingIfNecessary(event, true); if (entry == null) { return true; } @@ -476,7 +475,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE public boolean sink(LogEvent event) { try { - CanalEntry.Entry entry = parseAndProfilingIfNecessary(event,true); + CanalEntry.Entry entry = parseAndProfilingIfNecessary(event, true); if (entry == null) { return true; } @@ -555,7 +554,8 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE } } catch (Exception e) { logger.warn(String.format("the binlogfile:%s doesn't exist, to continue to search the next binlogfile , caused by", - startSearchBinlogFile), e); + startSearchBinlogFile), + e); int binlogSeqNum = Integer.parseInt(startSearchBinlogFile.substring(startSearchBinlogFile.indexOf(".") + 1)); if (binlogSeqNum <= 1) { logger.warn("Didn't find the corresponding binlog files"); @@ -687,7 +687,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE public boolean sink(LogEvent event) { EntryPosition entryPosition = null; try { - CanalEntry.Entry entry = parseAndProfilingIfNecessary(event,true); + CanalEntry.Entry entry = parseAndProfilingIfNecessary(event, true); if (entry == null) { return true; } @@ -707,7 +707,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE } if (StringUtils.equals(endPosition.getJournalName(), logfilename) - && endPosition.getPosition() <= (logfileoffset + event.getEventLen())) { + && endPosition.getPosition() <= logfileoffset) { return false; } @@ -715,9 +715,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE // position = current + // data.length，代表该事务的下一条offest，避免多余的事务重复 if (CanalEntry.EntryType.TRANSACTIONEND.equals(entry.getEntryType())) { - entryPosition = new EntryPosition(logfilename, - logfileoffset + event.getEventLen(), - logposTimestamp); + entryPosition = new EntryPosition(logfilename, logfileoffset, logposTimestamp); logger.debug("set {} to be pending start position before finding another proper one...", entryPosition); logPosition.setPostion(entryPosition); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DBMSAction.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DBMSAction.java deleted file mode 100644 index 4efbdc64..00000000 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DBMSAction.java +++ /dev/null @@ -1,77 +0,0 @@ -package com.alibaba.otter.canal.parse.inbound.mysql.dbsync; - -/** - * Defines database change action types: INSERT, UPDATE, DELETE, OTHER. - * - * @author Changyuan.lh - * @version 1.0 - */ -public enum DBMSAction { - - INSERT('I'), UPDATE('U'), DELETE('D'), REPLACE('R'), OTHER('O'), CREATE('C'), ALTER('A'), ERASE('E'), QUERY('Q'), - ROWQUERY('W'), TRUNCATE('T'), CINDEX('X'), DINDEX('Y'), RENAME('Z'); - - protected final byte bValue; - - DBMSAction(char ch){ - this.bValue = (byte) ch; - } - - /** - * Return action type from byte value. - */ - public static DBMSAction fromValue(int iValue) { - switch ((char) iValue) { - case 'I': - case 'M': // MERGE (Oracle only) - return INSERT; - case 'U': - return UPDATE; - case 'D': // DELETE - return DELETE; - case 'R': // REPLACE - return REPLACE; - case 'C': - return CREATE; - case 'A': - return ALTER; - case 'E': - return ERASE; - case 'Q': - return QUERY; - case 'W': - return ROWQUERY; - case 'T': - return TRUNCATE; - case 'X': - return CINDEX; - case 'Y': - return DINDEX; - case 'Z': - return RENAME; - } - return OTHER; - } - - /** - * Return action type from query. - */ - public static DBMSAction fromQuery(String query) { - int length = query.length(); - for (int index = 0; index < length; index++) { - char ch = query.charAt(index); - if (!Character.isWhitespace(ch)) { - return DBMSAction.fromValue(ch); - } - } - - return OTHER; - } - - /** - * Return byte value of action type. - */ - public byte value() { - return bValue; - } -} 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 222d3010..4f0ada45 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 @@ -9,7 +9,6 @@ import java.sql.Types; import java.util.BitSet; import java.util.List; -import com.taobao.tddl.dbsync.binlog.BinlogPosition; import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.exception.ExceptionUtils; import org.slf4j.Logger; @@ -22,7 +21,9 @@ import com.alibaba.otter.canal.parse.exception.TableIdNotFoundException; import com.alibaba.otter.canal.parse.inbound.BinlogParser; import com.alibaba.otter.canal.parse.inbound.TableMeta; import com.alibaba.otter.canal.parse.inbound.TableMeta.FieldMeta; -import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.SimpleDdlParser.DdlResult; +import com.alibaba.otter.canal.parse.inbound.mysql.ddl.DdlResult; +import com.alibaba.otter.canal.parse.inbound.mysql.ddl.DruidDdlParser; +import com.alibaba.otter.canal.parse.inbound.mysql.ddl.SimpleDdlParser; import com.alibaba.otter.canal.protocol.CanalEntry.Column; import com.alibaba.otter.canal.protocol.CanalEntry.Entry; import com.alibaba.otter.canal.protocol.CanalEntry.EntryType; @@ -35,6 +36,7 @@ import com.alibaba.otter.canal.protocol.CanalEntry.TransactionBegin; import com.alibaba.otter.canal.protocol.CanalEntry.TransactionEnd; import com.alibaba.otter.canal.protocol.CanalEntry.Type; import com.google.protobuf.ByteString; +import com.taobao.tddl.dbsync.binlog.BinlogPosition; import com.taobao.tddl.dbsync.binlog.LogEvent; import com.taobao.tddl.dbsync.binlog.event.DeleteRowsLogEvent; import com.taobao.tddl.dbsync.binlog.event.IntvarLogEvent; @@ -87,6 +89,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar private boolean filterTableError = false; // 新增rows过滤，用于仅订阅除rows以外的数据 private boolean filterRows = false; + private boolean useDruidDdlFilter = true; @Override public Entry parse(LogEvent logEvent, boolean isSeek) throws CanalParseException { @@ -100,7 +103,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar binlogFileName = ((RotateLogEvent) logEvent).getFilename(); break; case LogEvent.QUERY_EVENT: - return parseQueryEvent((QueryLogEvent) logEvent,isSeek); + return parseQueryEvent((QueryLogEvent) logEvent, isSeek); case LogEvent.XID_EVENT: return parseXidEvent((XidLogEvent) logEvent); case LogEvent.TABLE_MAP_EVENT: @@ -150,110 +153,139 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar Header header = createHeader(binlogFileName, event.getHeader(), "", "", null); return createEntry(header, EntryType.TRANSACTIONEND, transactionEnd.toByteString()); } else { - // DDL语句处理 - DdlResult result = SimpleDdlParser.parse(queryString, event.getDbName()); - - String schemaName = event.getDbName(); - if (StringUtils.isNotEmpty(result.getSchemaName())) { - schemaName = result.getSchemaName(); - } - - String tableName = result.getTableName(); + boolean notFilter = false; EventType type = EventType.QUERY; - // fixed issue https://github.com/alibaba/canal/issues/58 - if (result.getType() == EventType.ALTER || result.getType() == EventType.ERASE - || result.getType() == EventType.CREATE || result.getType() == EventType.TRUNCATE - || result.getType() == EventType.RENAME || result.getType() == EventType.CINDEX - || result.getType() == EventType.DINDEX) { // 针对DDL类型 - - if (filterQueryDdl) { - return null; + String tableName = null; + String schemaName = null; + if (useDruidDdlFilter) { + List results = DruidDdlParser.parse(queryString, event.getDbName()); + for (DdlResult result : results) { + if (!processFilter(queryString, result)) { + // 只要有一个数据不进行过滤 + notFilter = true; + } + } + if (results.size() > 0) { + // 如果针对多行的DDL,只能取第一条 + type = results.get(0).getType(); + schemaName = results.get(0).getSchemaName(); + tableName = results.get(0).getTableName(); + } + } else { + DdlResult result = SimpleDdlParser.parse(queryString, event.getDbName()); + if (!processFilter(queryString, result)) { + notFilter = true; } type = result.getType(); - if (StringUtils.isEmpty(tableName) - || (result.getType() == EventType.RENAME && StringUtils.isEmpty(result.getOriTableName()))) { - // 如果解析不出tableName,记录一下日志，方便bugfix，目前直接抛出异常，中断解析 - throw new CanalParseException("SimpleDdlParser process query failed. pls submit issue with this queryString: " - + queryString + " , and DdlResult: " + result.toString()); - // return null; - } else { - // check name filter - String name = schemaName + "." + tableName; - if (nameFilter != null && !nameFilter.filter(name)) { - if (result.getType() == EventType.RENAME) { - // rename校验只要源和目标满足一个就进行操作 - if (nameFilter != null - && !nameFilter.filter(result.getOriSchemaName() + "." + result.getOriTableName())) { - return null; - } - } else { - // 其他情况返回null - return null; - } - } + schemaName = result.getSchemaName(); + tableName = result.getTableName(); + } - if (nameBlackFilter != null && nameBlackFilter.filter(name)) { - if (result.getType() == EventType.RENAME) { - // rename校验只要源和目标满足一个就进行操作 - if (nameBlackFilter != null - && nameBlackFilter.filter(result.getOriSchemaName() + "." + result.getOriTableName())) { - return null; - } - } else { - // 其他情况返回null - return null; - } - } - } - } else if (result.getType() == EventType.INSERT || result.getType() == EventType.UPDATE - || result.getType() == EventType.DELETE) { - // 对外返回，保证兼容，还是返回QUERY类型，这里暂不解析tableName，所以无法支持过滤 - if (filterQueryDml) { - return null; - } - } else if (filterQueryDcl) { + if (!notFilter) { + // 如果是过滤的数据就不处理了 return null; } - // 更新下table meta cache - //if (tableMetaCache != null - // && (result.getType() == EventType.ALTER || result.getType() == EventType.ERASE || result.getType() == EventType.RENAME)) { - // for (DdlResult renameResult = result; renameResult != null; renameResult = renameResult.getRenameTableResult()) { - // String schemaName0 = event.getDbName(); // 防止rename语句后产生schema变更带来影响 - // if (StringUtils.isNotEmpty(renameResult.getSchemaName())) { - // schemaName0 = renameResult.getSchemaName(); - // } - // - // tableName = renameResult.getTableName(); - // if (StringUtils.isNotEmpty(tableName)) { - // // 如果解析到了正确的表信息，则根据全名进行清除 - // tableMetaCache.clearTableMeta(schemaName0, tableName); - // } else { - // // 如果无法解析正确的表信息，则根据schema进行清除 - // tableMetaCache.clearTableMetaWithSchemaName(schemaName0); - // } - // } - //} - - //使用新的表结构元数据管理方式 - BinlogPosition position = createPosition(event.getHeader()); - tableMetaCache.apply(position, event.getDbName(), queryString); + if (!isSeek) { + // 使用新的表结构元数据管理方式 + BinlogPosition position = createPosition(event.getHeader()); + tableMetaCache.apply(position, event.getDbName(), queryString); + } Header header = createHeader(binlogFileName, event.getHeader(), schemaName, tableName, type); RowChange.Builder rowChangeBuider = RowChange.newBuilder(); - if (result.getType() != EventType.QUERY) { + if (type != EventType.QUERY) { rowChangeBuider.setIsDdl(true); } rowChangeBuider.setSql(queryString); if (StringUtils.isNotEmpty(event.getDbName())) {// 可能为空 rowChangeBuider.setDdlSchemaName(event.getDbName()); } - rowChangeBuider.setEventType(result.getType()); + rowChangeBuider.setEventType(type); return createEntry(header, EntryType.ROWDATA, rowChangeBuider.build().toByteString()); } } + private boolean processFilter(String queryString, DdlResult result) { + String schemaName = result.getSchemaName(); + String tableName = result.getTableName(); + // fixed issue https://github.com/alibaba/canal/issues/58 + // 更新下table meta cache + if (tableMetaCache != null + && (result.getType() == EventType.ALTER || result.getType() == EventType.ERASE || result.getType() == EventType.RENAME)) { + // 对外返回，保证兼容，还是返回QUERY类型，这里暂不解析tableName，所以无法支持过滤 + for (DdlResult renameResult = result; renameResult != null; renameResult = renameResult.getRenameTableResult()) { + String schemaName0 = renameResult.getSchemaName(); + String tableName0 = renameResult.getTableName(); + if (StringUtils.isNotEmpty(tableName0)) { + // 如果解析到了正确的表信息，则根据全名进行清除 + tableMetaCache.clearTableMeta(schemaName0, tableName0); + } else { + // 如果无法解析正确的表信息，则根据schema进行清除 + tableMetaCache.clearTableMetaWithSchemaName(schemaName0); + } + } + } + + // fixed issue https://github.com/alibaba/canal/issues/58 + if (result.getType() == EventType.ALTER || result.getType() == EventType.ERASE + || result.getType() == EventType.CREATE || result.getType() == EventType.TRUNCATE + || result.getType() == EventType.RENAME || result.getType() == EventType.CINDEX + || result.getType() == EventType.DINDEX) { // 针对DDL类型 + + if (filterQueryDdl) { + return true; + } + + if (StringUtils.isEmpty(tableName) + || (result.getType() == EventType.RENAME && StringUtils.isEmpty(result.getOriTableName()))) { + // 如果解析不出tableName,记录一下日志，方便bugfix，目前直接抛出异常，中断解析 + throw new CanalParseException("SimpleDdlParser process query failed. pls submit issue with this queryString: " + + queryString + " , and DdlResult: " + result.toString()); + // return null; + } else { + // check name filter + String name = schemaName + "." + tableName; + if (nameFilter != null && !nameFilter.filter(name)) { + if (result.getType() == EventType.RENAME) { + // rename校验只要源和目标满足一个就进行操作 + if (nameFilter != null + && !nameFilter.filter(result.getOriSchemaName() + "." + result.getOriTableName())) { + return true; + } + } else { + // 其他情况返回null + return true; + } + } + + if (nameBlackFilter != null && nameBlackFilter.filter(name)) { + if (result.getType() == EventType.RENAME) { + // rename校验只要源和目标满足一个就进行操作 + if (nameBlackFilter != null + && nameBlackFilter.filter(result.getOriSchemaName() + "." + result.getOriTableName())) { + return true; + } + } else { + // 其他情况返回null + return true; + } + } + } + } else if (result.getType() == EventType.INSERT || result.getType() == EventType.UPDATE + || result.getType() == EventType.DELETE) { + // 对外返回，保证兼容，还是返回QUERY类型，这里暂不解析tableName，所以无法支持过滤 + if (filterQueryDml) { + return true; + } + } else if (filterQueryDcl) { + return true; + } + + return false; + } + private Entry parseRowsQueryEvent(RowsQueryLogEvent event) { if (filterQueryDml) { return null; @@ -366,7 +398,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar boolean tableError = false; TableMeta tableMeta = null; if (tableMetaCache != null) {// 入错存在table meta cache - tableMeta = getTableMeta(table.getDbName(), table.getTableName(), true,position); + tableMeta = getTableMeta(table.getDbName(), table.getTableName(), true, position); if (tableMeta == null) { tableError = true; if (!filterTableError) { @@ -444,7 +476,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar // 3. 锁住应用请求，将临时表rename为老表的名字，完成增加字段的操作 // 尝试做一次reload，可能因为ddl没有正确解析，或者使用了类似online ddl的操作 // 因为online ddl没有对应表名的alter语法，所以不会有clear cache的操作 - tableMeta = getTableMeta(event.getTable().getDbName(), event.getTable().getTableName(), false,position);// 强制重新获取一次 + tableMeta = getTableMeta(event.getTable().getDbName(), event.getTable().getTableName(), false, position);// 强制重新获取一次 if (tableMeta == null) { tableError = true; if (!filterTableError) { @@ -686,7 +718,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar private TableMeta getTableMeta(String dbName, String tbName, boolean useCache, BinlogPosition position) { try { - return tableMetaCache.getTableMeta(dbName, tbName, useCache,position); + return tableMetaCache.getTableMeta(dbName, tbName, useCache, position); } catch (Exception e) { String message = ExceptionUtils.getRootCauseMessage(e); if (filterTableError) { 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 ff0e720f..23229859 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 @@ -1,24 +1,24 @@ package com.alibaba.otter.canal.parse.inbound.mysql.dbsync; import java.io.IOException; -import java.sql.ResultSet; -import java.sql.SQLException; import java.util.ArrayList; +import java.util.HashMap; import java.util.List; import java.util.Map; -import com.google.common.cache.CacheBuilder; -import com.google.common.cache.CacheLoader; -import com.google.common.cache.LoadingCache; -import com.taobao.tddl.dbsync.binlog.BinlogPosition; import org.apache.commons.lang.StringUtils; +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.inbound.TableMeta; import com.alibaba.otter.canal.parse.inbound.TableMeta.FieldMeta; import com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection; -import com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection.ProcessJdbcResult; import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaManager; +import com.google.common.cache.CacheBuilder; +import com.google.common.cache.CacheLoader; +import com.google.common.cache.LoadingCache; +import com.taobao.tddl.dbsync.binlog.BinlogPosition; /** * 处理table meta解析和缓存 @@ -28,57 +28,23 @@ import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaManager; */ public class TableMetaCache { - public static final String COLUMN_NAME = "COLUMN_NAME"; - public static final String COLUMN_TYPE = "COLUMN_TYPE"; - public static final String IS_NULLABLE = "IS_NULLABLE"; - public static final String COLUMN_KEY = "COLUMN_KEY"; - public static final String COLUMN_DEFAULT = "COLUMN_DEFAULT"; - public static final String EXTRA = "EXTRA"; - private MysqlConnection connection; - private boolean isOnRDS = false; + public static final String COLUMN_NAME = "COLUMN_NAME"; + public static final String COLUMN_TYPE = "COLUMN_TYPE"; + public static final String IS_NULLABLE = "IS_NULLABLE"; + public static final String COLUMN_KEY = "COLUMN_KEY"; + public static final String COLUMN_DEFAULT = "COLUMN_DEFAULT"; + public static final String EXTRA = "EXTRA"; + private MysqlConnection connection; + private boolean isOnRDS = false; - private TableMetaManager tableMetaManager; + private TableMetaManager tableMetaManager; // 第一层tableId,第二层schema.table,解决tableId重复，对应多张表 private LoadingCache tableMetaDB; - - // 第一层tableId,第二层schema.table,解决tableId重复，对应多张表 - private Map tableMetaCache; - - /** - * 从db获取表结构 - * @param fullname - * @return - */ - private TableMeta getTableMetaByDB(final String fullname) { - return connection.query("desc " + fullname, new ProcessJdbcResult() { - - @Override - public TableMeta process(ResultSet rs) throws SQLException { - List metas = new ArrayList(); - while (rs.next()) { - FieldMeta meta = new FieldMeta(); - // 做一个优化，使用String.intern()，共享String对象，减少内存使用 - meta.setColumnName(rs.getString("Field")); - meta.setColumnType(rs.getString("Type")); - meta.setNullable(StringUtils.equalsIgnoreCase(rs.getString("Null"), "YES")); - meta.setKey("PRI".equalsIgnoreCase(rs.getString("Key"))); - meta.setDefaultValue(rs.getString("Default")); - metas.add(meta); - } - - String[] names = StringUtils.split(fullname, "`.`"); - String schema = names[0]; - String table = names[1].substring(0, names[1].length()); - return new TableMeta(schema, table, metas); - } - }); - } - - public TableMetaCache(MysqlConnection con,TableMetaManager tableMetaManager) { + public TableMetaCache(MysqlConnection con, TableMetaManager tableMetaManager){ this.connection = con; this.tableMetaManager = tableMetaManager; - //如果持久存储的表结构为空，从db里面获取下 + // 如果持久存储的表结构为空，从db里面获取下 if (tableMetaManager == null) { this.tableMetaDB = CacheBuilder.newBuilder().build(new CacheLoader() { @@ -100,16 +66,62 @@ public class TableMetaCache { }); } - isOnRDS = connection.query("show global variables like 'rds\\_%'", new ProcessJdbcResult() { - - @Override - public Boolean process(ResultSet rs) throws SQLException { - if (rs.next()) { - return true; - } - return false; + try { + ResultSetPacket packet = connection.query("show global variables like 'rds\\_%'"); + if (packet.getFieldValues().size() > 0) { + isOnRDS = true; } - }); + } catch (IOException e) { + } + } + + private TableMeta getTableMetaByDB(String fullname) throws IOException { + ResultSetPacket packet = connection.query("desc " + fullname); + String[] names = StringUtils.split(fullname, "`.`"); + String schema = names[0]; + String table = names[1].substring(0, names[1].length()); + return new TableMeta(schema, table, parserTableMeta(packet)); + } + + public static List parserTableMeta(ResultSetPacket packet) { + Map nameMaps = new HashMap(6, 1f); + + int index = 0; + for (FieldPacket fieldPacket : packet.getFieldDescriptors()) { + nameMaps.put(fieldPacket.getOriginalName(), index++); + } + + int size = packet.getFieldDescriptors().size(); + int count = packet.getFieldValues().size() / packet.getFieldDescriptors().size(); + List result = new ArrayList(); + for (int i = 0; i < count; i++) { + FieldMeta meta = new FieldMeta(); + // 做一个优化，使用String.intern()，共享String对象，减少内存使用 + meta.setColumnName(packet.getFieldValues().get(nameMaps.get(COLUMN_NAME) + i * size).intern()); + meta.setColumnType(packet.getFieldValues().get(nameMaps.get(COLUMN_TYPE) + i * size)); + meta.setNullable(StringUtils.equalsIgnoreCase(packet.getFieldValues().get(nameMaps.get(IS_NULLABLE) + i + * size), + "YES")); + meta.setKey("PRI".equalsIgnoreCase(packet.getFieldValues().get(nameMaps.get(COLUMN_KEY) + i * size))); + meta.setDefaultValue(packet.getFieldValues().get(nameMaps.get(COLUMN_DEFAULT) + i * size)); + meta.setExtra(packet.getFieldValues().get(nameMaps.get(EXTRA) + i * size)); + + result.add(meta); + } + + return result; + } + + public TableMeta getTableMeta(String schema, String table) { + return getTableMeta(schema, table, true); + } + + public TableMeta getTableMeta(String schema, String table, boolean useCache) { + if (!useCache) { + tableMetaDB.invalidate(getFullName(schema, table)); + } + + return tableMetaDB.getUnchecked(getFullName(schema, table)); } public TableMeta getTableMeta(String schema, String table, BinlogPosition position) { @@ -122,20 +134,19 @@ public class TableMetaCache { tableMeta = tableMetaManager.find(schema, table); if (tableMeta == null) { // 因为条件变化，可能第一次的tableMeta没取到，需要从db获取一次，并记录到snapshot中 - String createDDL = connection.query("show create table " + getFullName(schema, table), - new ProcessJdbcResult() { - - @Override - public String process(ResultSet rs) throws SQLException { - while (rs.next()) { - return rs.getString(2); - } - return null; - } - }); - // 强制覆盖掉内存值 - tableMetaManager.apply(position, schema, createDDL); - tableMeta = tableMetaManager.find(schema, table); + String fullName = getFullName(schema, table); + try { + ResultSetPacket packet = connection.query("show create table " + fullName); + String createDDL = null; + if (packet.getFieldValues().size() > 0) { + createDDL = packet.getFieldValues().get(1); + } + // 强制覆盖掉内存值 + tableMetaManager.apply(position, schema, createDDL); + tableMeta = tableMetaManager.find(schema, table); + } catch (IOException e) { + throw new CanalParseException("fetch failed by table meta:" + fullName, e); + } } return tableMeta; } else { @@ -147,8 +158,6 @@ public class TableMetaCache { } } - - public void clearTableMeta(String schema, String table) { if (tableMetaManager != null) { // tsdb不需要做,会基于ddl sql自动清理 @@ -195,8 +204,6 @@ public class TableMetaCache { } } - - private String getFullName(String schema, String table) { StringBuilder builder = new StringBuilder(); return builder.append('`') diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/ddl/DdlResult.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/ddl/DdlResult.java index 44975ffb..9a7a5bb5 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/ddl/DdlResult.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/ddl/DdlResult.java @@ -1,6 +1,6 @@ package com.alibaba.otter.canal.parse.inbound.mysql.ddl; -import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.DBMSAction; +import com.alibaba.otter.canal.protocol.CanalEntry.EventType; /** * @author agapple 2017年8月1日 下午7:30:42 @@ -8,30 +8,30 @@ import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.DBMSAction; */ public class DdlResult { - private String schemaName; - private String tableName; - private String oriSchemaName; // rename ddl中的源表 - private String oriTableName; // rename ddl中的目标表 - private DBMSAction type; + private String schemaName; + private String tableName; + private String oriSchemaName; // rename ddl中的源表 + private String oriTableName; // rename ddl中的目标表 + private EventType type; private DdlResult renameTableResult; // 多个rename table的存储 /* * RENAME TABLE tbl_name TO new_tbl_name [, tbl_name2 TO new_tbl_name2] ... */ - public DdlResult() { + public DdlResult(){ } - public DdlResult(String schemaName) { + public DdlResult(String schemaName){ this.schemaName = schemaName; } - public DdlResult(String schemaName, String tableName) { + public DdlResult(String schemaName, String tableName){ this.schemaName = schemaName; this.tableName = tableName; } - public DdlResult(String schemaName, String tableName, String oriSchemaName, String oriTableName) { + public DdlResult(String schemaName, String tableName, String oriSchemaName, String oriTableName){ this.schemaName = schemaName; this.tableName = tableName; this.oriSchemaName = oriSchemaName; @@ -54,11 +54,11 @@ public class DdlResult { this.tableName = tableName; } - public DBMSAction getType() { + public EventType getType() { return type; } - public void setType(DBMSAction type) { + public void setType(EventType type) { this.type = type; } @@ -93,7 +93,7 @@ public class DdlResult { result.setOriTableName(oriTableName); result.setSchemaName(schemaName); result.setTableName(tableName); - //result.setType(type); + // result.setType(type); return result; } @@ -102,13 +102,12 @@ public class DdlResult { DdlResult ddlResult = this; StringBuffer sb = new StringBuffer(); do { - sb.append(String - .format("DdlResult [schemaName=%s , tableName=%s , oriSchemaName=%s , oriTableName=%s , type=%s ];", - ddlResult.schemaName, - ddlResult.tableName, - ddlResult.oriSchemaName, - ddlResult.oriTableName, - ddlResult.type)); + sb.append(String.format("DdlResult [schemaName=%s , tableName=%s , oriSchemaName=%s , oriTableName=%s , type=%s ];", + ddlResult.schemaName, + ddlResult.tableName, + ddlResult.oriSchemaName, + ddlResult.oriTableName, + ddlResult.type)); ddlResult = ddlResult.renameTableResult; } while (ddlResult != null); return sb.toString(); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/ddl/DruidDdlParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/ddl/DruidDdlParser.java index 647955c5..15577298 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/ddl/DruidDdlParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/ddl/DruidDdlParser.java @@ -33,11 +33,11 @@ import com.alibaba.druid.sql.dialect.mysql.ast.statement.MySqlRenameTableStateme import com.alibaba.druid.sql.dialect.mysql.ast.statement.MySqlRenameTableStatement.Item; import com.alibaba.druid.sql.parser.ParserException; import com.alibaba.druid.util.JdbcConstants; -import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.DBMSAction; +import com.alibaba.otter.canal.protocol.CanalEntry.EventType; /** * @author agapple 2017年7月27日 下午4:05:34 - * @since 3.2.5 + * @since 1.0.25 */ public class DruidDdlParser { @@ -48,7 +48,7 @@ public class DruidDdlParser { } catch (ParserException e) { // 可能存在一些SQL是不支持的，比如存储过程 DdlResult ddlResult = new DdlResult(); - ddlResult.setType(DBMSAction.QUERY); + ddlResult.setType(EventType.QUERY); return Arrays.asList(ddlResult); } @@ -58,7 +58,7 @@ public class DruidDdlParser { DdlResult ddlResult = new DdlResult(); SQLCreateTableStatement createTable = (SQLCreateTableStatement) statement; processName(ddlResult, schmeaName, createTable.getName(), false); - ddlResult.setType(DBMSAction.CREATE); + ddlResult.setType(EventType.CREATE); ddlResults.add(ddlResult); } else if (statement instanceof SQLAlterTableStatement) { SQLAlterTableStatement alterTable = (SQLAlterTableStatement) statement; @@ -71,30 +71,30 @@ public class DruidDdlParser { } else if (item instanceof SQLAlterTableAddIndex) { DdlResult ddlResult = new DdlResult(); processName(ddlResult, schmeaName, alterTable.getName(), false); - ddlResult.setType(DBMSAction.CINDEX); + ddlResult.setType(EventType.CINDEX); ddlResults.add(ddlResult); } else if (item instanceof SQLAlterTableDropIndex || item instanceof SQLAlterTableDropKey) { DdlResult ddlResult = new DdlResult(); processName(ddlResult, schmeaName, alterTable.getName(), false); - ddlResult.setType(DBMSAction.DINDEX); + ddlResult.setType(EventType.DINDEX); ddlResults.add(ddlResult); } else if (item instanceof SQLAlterTableAddConstraint) { DdlResult ddlResult = new DdlResult(); processName(ddlResult, schmeaName, alterTable.getName(), false); SQLConstraint constraint = ((SQLAlterTableAddConstraint) item).getConstraint(); if (constraint instanceof SQLUnique) { - ddlResult.setType(DBMSAction.CINDEX); + ddlResult.setType(EventType.CINDEX); ddlResults.add(ddlResult); } } else if (item instanceof SQLAlterTableDropConstraint) { DdlResult ddlResult = new DdlResult(); processName(ddlResult, schmeaName, alterTable.getName(), false); - ddlResult.setType(DBMSAction.DINDEX); + ddlResult.setType(EventType.DINDEX); ddlResults.add(ddlResult); } else { DdlResult ddlResult = new DdlResult(); processName(ddlResult, schmeaName, alterTable.getName(), false); - ddlResult.setType(DBMSAction.ALTER); + ddlResult.setType(EventType.ALTER); ddlResults.add(ddlResult); } } @@ -103,7 +103,7 @@ public class DruidDdlParser { for (SQLExprTableSource tableSource : dropTable.getTableSources()) { DdlResult ddlResult = new DdlResult(); processName(ddlResult, schmeaName, tableSource.getExpr(), false); - ddlResult.setType(DBMSAction.ERASE); + ddlResult.setType(EventType.ERASE); ddlResults.add(ddlResult); } } else if (statement instanceof SQLCreateIndexStatement) { @@ -111,21 +111,21 @@ public class DruidDdlParser { SQLTableSource tableSource = createIndex.getTable(); DdlResult ddlResult = new DdlResult(); processName(ddlResult, schmeaName, ((SQLExprTableSource) tableSource).getExpr(), false); - ddlResult.setType(DBMSAction.CINDEX); + ddlResult.setType(EventType.CINDEX); ddlResults.add(ddlResult); } else if (statement instanceof SQLDropIndexStatement) { SQLDropIndexStatement dropIndex = (SQLDropIndexStatement) statement; SQLExprTableSource tableSource = dropIndex.getTableName(); DdlResult ddlResult = new DdlResult(); processName(ddlResult, schmeaName, tableSource.getExpr(), false); - ddlResult.setType(DBMSAction.DINDEX); + ddlResult.setType(EventType.DINDEX); ddlResults.add(ddlResult); } else if (statement instanceof SQLTruncateStatement) { SQLTruncateStatement truncate = (SQLTruncateStatement) statement; for (SQLExprTableSource tableSource : truncate.getTableSources()) { DdlResult ddlResult = new DdlResult(); processName(ddlResult, schmeaName, tableSource.getExpr(), false); - ddlResult.setType(DBMSAction.TRUNCATE); + ddlResult.setType(EventType.TRUNCATE); ddlResults.add(ddlResult); } } else if (statement instanceof MySqlRenameTableStatement) { @@ -134,28 +134,28 @@ public class DruidDdlParser { DdlResult ddlResult = new DdlResult(); processName(ddlResult, schmeaName, item.getName(), true); processName(ddlResult, schmeaName, item.getTo(), false); - ddlResult.setType(DBMSAction.RENAME); + ddlResult.setType(EventType.RENAME); ddlResults.add(ddlResult); } } else if (statement instanceof SQLInsertStatement) { DdlResult ddlResult = new DdlResult(); SQLInsertStatement insert = (SQLInsertStatement) statement; processName(ddlResult, schmeaName, insert.getTableName(), true); - ddlResult.setType(DBMSAction.INSERT); + ddlResult.setType(EventType.INSERT); ddlResults.add(ddlResult); } else if (statement instanceof SQLUpdateStatement) { DdlResult ddlResult = new DdlResult(); SQLUpdateStatement update = (SQLUpdateStatement) statement; // 拿到的表名可能为null,比如update a,b set a.id=x processName(ddlResult, schmeaName, update.getTableName(), true); - ddlResult.setType(DBMSAction.UPDATE); + ddlResult.setType(EventType.UPDATE); ddlResults.add(ddlResult); } else if (statement instanceof SQLDeleteStatement) { DdlResult ddlResult = new DdlResult(); SQLDeleteStatement delete = (SQLDeleteStatement) statement; // 拿到的表名可能为null,比如delete a,b from a where a.id = b.id processName(ddlResult, schmeaName, delete.getTableName(), true); - ddlResult.setType(DBMSAction.DELETE); + ddlResult.setType(EventType.DELETE); ddlResults.add(ddlResult); } } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/SimpleDdlParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/ddl/SimpleDdlParser.java similarity index 73% rename from parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/SimpleDdlParser.java rename to parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/ddl/SimpleDdlParser.java index c6342bff..53f7de82 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/SimpleDdlParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/ddl/SimpleDdlParser.java @@ -1,4 +1,4 @@ -package com.alibaba.otter.canal.parse.inbound.mysql.dbsync; +package com.alibaba.otter.canal.parse.inbound.mysql.ddl; import org.apache.commons.lang.StringUtils; import org.apache.oro.text.regex.Perl5Matcher; @@ -218,102 +218,4 @@ public class SimpleDdlParser { sql = sb.toString(); } } - - public static class DdlResult { - - private String schemaName; - private String tableName; - private String oriSchemaName; // rename ddl中的源表 - private String oriTableName; // rename ddl中的目标表 - private EventType type; - private DdlResult renameTableResult; // 多个rename table的存储 - - /* - * RENAME TABLE tbl_name TO new_tbl_name [, tbl_name2 TO new_tbl_name2] - * ... - */ - - public DdlResult(){ - } - - public DdlResult(String schemaName){ - this.schemaName = schemaName; - } - - public DdlResult(String schemaName, String tableName){ - this.schemaName = schemaName; - this.tableName = tableName; - } - - public DdlResult(String schemaName, String tableName, String oriSchemaName, String oriTableName){ - this.schemaName = schemaName; - this.tableName = tableName; - this.oriSchemaName = oriSchemaName; - this.oriTableName = oriTableName; - } - - public String getSchemaName() { - return schemaName; - } - - public void setSchemaName(String schemaName) { - this.schemaName = schemaName; - } - - public String getTableName() { - return tableName; - } - - public void setTableName(String tableName) { - this.tableName = tableName; - } - - public EventType getType() { - return type; - } - - public void setType(EventType type) { - this.type = type; - } - - public String getOriSchemaName() { - return oriSchemaName; - } - - public void setOriSchemaName(String oriSchemaName) { - this.oriSchemaName = oriSchemaName; - } - - public String getOriTableName() { - return oriTableName; - } - - public void setOriTableName(String oriTableName) { - this.oriTableName = oriTableName; - } - - public DdlResult getRenameTableResult() { - return renameTableResult; - } - - public void setRenameTableResult(DdlResult renameTableResult) { - this.renameTableResult = renameTableResult; - } - - @Override - public String toString() { - DdlResult ddlResult = this; - StringBuffer sb = new StringBuffer(); - do { - sb.append(String.format("DdlResult [schemaName=%s , tableName=%s , oriSchemaName=%s , oriTableName=%s , type=%s ];", - ddlResult.schemaName, - ddlResult.tableName, - ddlResult.oriSchemaName, - ddlResult.oriTableName, - ddlResult.type)); - ddlResult = ddlResult.renameTableResult; - } while (ddlResult != null); - return sb.toString(); - } - } } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DataSourceFactoryTSDB.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DataSourceFactoryTSDB.java index 118da365..25f60ee3 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DataSourceFactoryTSDB.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DataSourceFactoryTSDB.java @@ -1,43 +1,41 @@ package com.alibaba.otter.canal.parse.inbound.mysql.tsdb; -import java.sql.SQLException; - import javax.sql.DataSource; import com.alibaba.druid.pool.DruidDataSource; +import com.alibaba.druid.pool.vendor.MySqlExceptionSorter; +import com.alibaba.druid.pool.vendor.MySqlValidConnectionChecker; import com.alibaba.otter.canal.parse.exception.CanalParseException; -import org.apache.commons.lang.StringUtils; -import org.apache.commons.lang.math.NumberUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - /** - * Created by wanshao - * Date: 2017/9/22 - * Time: 下午2:46 + * Created by wanshao Date: 2017/9/22 Time: 下午2:46 **/ public class DataSourceFactoryTSDB { - private static Logger logger = LoggerFactory.getLogger(DataSourceFactoryTSDB.class); - - public static DataSource getDataSource(String address, String userName, String password, - boolean enable, - String defaultDatabaseName, String url, String driverClassName) { - - DruidDataSource druidDataSource = new DruidDataSource(); - druidDataSource.setUrl(url); - druidDataSource.setUsername(userName); - druidDataSource.setPassword(password); - + public static DataSource getDataSource(String url, String userName, String password) { try { + DruidDataSource druidDataSource = new DruidDataSource(); + druidDataSource.setUrl(url); + druidDataSource.setUsername(userName); + druidDataSource.setPassword(password); + druidDataSource.setTestWhileIdle(true); + druidDataSource.setTestOnBorrow(false); + druidDataSource.setTestOnReturn(false); + druidDataSource.setNotFullTimeoutRetryCount(2); + druidDataSource.setValidConnectionCheckerClassName(MySqlValidConnectionChecker.class.getName()); + druidDataSource.setExceptionSorterClassName(MySqlExceptionSorter.class.getName()); + druidDataSource.setValidationQuery("SELECT 1"); + druidDataSource.setInitialSize(1); + druidDataSource.setMinIdle(1); + druidDataSource.setMaxActive(30); + druidDataSource.setMaxWait(10 * 1000); + druidDataSource.setTimeBetweenEvictionRunsMillis(60 * 1000); + druidDataSource.setMinEvictableIdleTimeMillis(50 * 1000); + druidDataSource.setUseUnfairLock(true); druidDataSource.init(); - } catch (SQLException e) { - logger.error("druidDataSource.init", e); - throw new CanalParseException("初始化druid dataSource出错"); + return druidDataSource; + } catch (Throwable e) { + throw new CanalParseException("init druidDataSource failed", e); } - return druidDataSource; } - - } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java index 293661f4..f3c6aeab 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java @@ -45,8 +45,8 @@ import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDB; public class MemoryTableMeta implements TableMetaTSDB { private Map, TableMeta> tableMetas = new ConcurrentHashMap, TableMeta>(); - private SchemaRepository repository = new SchemaRepository(JdbcConstants.MYSQL); - private Logger logger = LoggerFactory.getLogger(MemoryTableMeta.class); + private SchemaRepository repository = new SchemaRepository(JdbcConstants.MYSQL); + private Logger logger = LoggerFactory.getLogger(MemoryTableMeta.class); public MemoryTableMeta(Logger logger){ this.logger = logger; @@ -208,7 +208,7 @@ public class MemoryTableMeta implements TableMetaTSDB { if (sqlName instanceof SQLPropertyExpr) { SQLIdentifierExpr owner = (SQLIdentifierExpr) ((SQLPropertyExpr) sqlName).getOwner(); return DruidDdlParser.unescapeName(owner.getName()) + "." - + DruidDdlParser.unescapeName(((SQLPropertyExpr) sqlName).getName()); + + DruidDdlParser.unescapeName(((SQLPropertyExpr) sqlName).getName()); } else if (sqlName instanceof SQLIdentifierExpr) { return DruidDdlParser.unescapeName(((SQLIdentifierExpr) sqlName).getName()); } else { 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 859152f7..9d926548 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 @@ -1,8 +1,6 @@ package com.alibaba.otter.canal.parse.inbound.mysql.tsdb; -import java.sql.ResultSet; -import java.sql.SQLException; -import java.sql.Statement; +import java.io.IOException; import java.util.ArrayList; import java.util.HashMap; import java.util.List; @@ -15,58 +13,54 @@ import java.util.regex.Pattern; import javax.annotation.Resource; -import com.alibaba.druid.sql.repository.Schema; -import com.alibaba.fastjson.JSON; -import com.alibaba.fastjson.JSONArray; -import com.alibaba.fastjson.JSONObject; -import com.alibaba.otter.canal.filter.CanalEventFilter; -import com.alibaba.otter.canal.parse.exception.CanalParseException; -import com.alibaba.otter.canal.parse.inbound.TableMeta; -import com.alibaba.otter.canal.parse.inbound.TableMeta.FieldMeta; -import com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection; -import com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection.ProcessJdbcResult; -import com.alibaba.otter.canal.parse.inbound.mysql.ddl.DdlResult; -import com.alibaba.otter.canal.parse.inbound.mysql.ddl.DruidDdlParser; -import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryDAO; -import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaSnapshotDAO; -import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.model.MetaHistoryDO; -import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.model.MetaSnapshotDO; - -import com.taobao.tddl.dbsync.binlog.BinlogPosition; import org.apache.commons.beanutils.BeanUtils; import org.apache.commons.lang.ObjectUtils; import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.alibaba.druid.sql.repository.Schema; +import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson.JSONObject; +import com.alibaba.otter.canal.filter.CanalEventFilter; +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.inbound.TableMeta; +import com.alibaba.otter.canal.parse.inbound.TableMeta.FieldMeta; +import com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection; +import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.TableMetaCache; +import com.alibaba.otter.canal.parse.inbound.mysql.ddl.DdlResult; +import com.alibaba.otter.canal.parse.inbound.mysql.ddl.DruidDdlParser; +import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaHistoryDAO; +import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao.MetaSnapshotDAO; +import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.model.MetaHistoryDO; +import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.model.MetaSnapshotDO; +import com.taobao.tddl.dbsync.binlog.BinlogPosition; + /** - * 基于console远程管理 - * - * see internal class: CanalTableMeta , ConsoleTableMetaTSDB + * 基于console远程管理 see internal class: CanalTableMeta , ConsoleTableMetaTSDB * * @author agapple 2017年7月27日 下午10:47:55 * @since 3.2.5 */ public class TableMetaManager implements TableMetaTSDB { - private static Pattern pattern = Pattern.compile("Duplicate entry '.*' for key '*'"); + private static Logger logger = LoggerFactory.getLogger(TableMetaManager.class); + private static Pattern pattern = Pattern.compile("Duplicate entry '.*' for key '*'"); private static final BinlogPosition INIT_POSITION = BinlogPosition.parseFromString("0:0#-2.-1"); - private Logger logger = LoggerFactory.getLogger(TableMetaManager.class); - private String consoleDomain = null; - private int retry = 3; - private MemoryTableMeta memoryTableMeta; - private MysqlConnection connection; // 查询meta信息的链接 - private CanalEventFilter filter; - private BinlogPosition lastPosition; - private ScheduledExecutorService scheduler; + private MemoryTableMeta memoryTableMeta; + private MysqlConnection connection; // 查询meta信息的链接 + private CanalEventFilter filter; + private BinlogPosition lastPosition; + private ScheduledExecutorService scheduler; @Resource - private MetaHistoryDAO metaHistoryDAO; + private MetaHistoryDAO metaHistoryDAO; @Resource - private MetaSnapshotDAO metaSnapshotDAO; + private MetaSnapshotDAO metaSnapshotDAO; - public void init(){ + public void init() { this.memoryTableMeta = new MemoryTableMeta(logger); this.scheduler = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() { @@ -91,7 +85,7 @@ public class TableMetaManager implements TableMetaTSDB { }, 24, 24, TimeUnit.SECONDS); } - public TableMetaManager() { + public TableMetaManager(){ } @@ -147,80 +141,46 @@ public class TableMetaManager implements TableMetaTSDB { * 初始化的时候dump一下表结构 */ private boolean dumpTableMeta(MysqlConnection connection, final CanalEventFilter filter) { - List schemas = connection.query("show databases", new ProcessJdbcResult() { - - @Override - public List process(ResultSet rs) throws SQLException { - List schemas = new ArrayList(); - while (rs.next()) { - String schema = rs.getString(1); - if (!filter.filter(schema)) { - schemas.add(schema); - } + try { + ResultSetPacket packet = connection.query("show databases"); + List schemas = new ArrayList(); + for (String schema : packet.getFieldValues()) { + if (!filter.filter(schema)) { + schemas.add(schema); } - return schemas; - } - }); - - for (String schema : schemas) { - List tables = connection.query("show tables from `" + schema + "`", new ProcessJdbcResult() { - - @Override - public List process(ResultSet rs) throws SQLException { - List tables = new ArrayList(); - while (rs.next()) { - String table = rs.getString(1); - if (!filter.filter(table)) { - tables.add(table); - } - } - return tables; - } - }); - - StringBuilder sql = new StringBuilder(); - for (String table : tables) { - sql.append("show create table `" + schema + "`.`" + table + "`;"); } - // 使用多语句方式读取 - Statement stmt = null; - try { - stmt = connection.getConn().createStatement(); - ResultSet rs = stmt.executeQuery(sql.toString()); - boolean existMoreResult = false; - do { - if (existMoreResult) { - rs = stmt.getResultSet(); + for (String schema : schemas) { + packet = connection.query("show tables from `" + schema + "`"); + List tables = new ArrayList(); + for (String table : packet.getFieldValues()) { + if (!filter.filter(table)) { + tables.add(table); } + } - while (rs.next()) { - String oneTableCreateSql = rs.getString(2); + StringBuilder sql = new StringBuilder(); + for (String table : tables) { + sql.append("show create table `" + schema + "`.`" + table + "`;"); + } + + List packets = connection.queryMulti(sql.toString()); + for (ResultSetPacket onePacket : packets) { + if (onePacket.getFieldValues().size() > 1) { + String oneTableCreateSql = onePacket.getFieldValues().get(1); memoryTableMeta.apply(INIT_POSITION, schema, oneTableCreateSql); } - - existMoreResult = stmt.getMoreResults(); - } while (existMoreResult); - } catch (SQLException e) { - throw new CanalParseException(e); - } finally { - if (stmt != null) { - try { - stmt.close(); - } catch (SQLException e) { - // ignore - } } } - } - return true; + return true; + } catch (IOException e) { + throw new CanalParseException(e); + } } private boolean applyHistoryToDB(BinlogPosition position, String schema, String ddl) { - Map content = new HashMap(); - content.put("binlogFile", position.getFileName()); content.put("binlogOffest", String.valueOf(position.getPosition())); content.put("binlogMasterId", String.valueOf(position.getMasterId())); @@ -240,26 +200,23 @@ public class TableMetaManager implements TableMetaTSDB { // content.put("extra", ""); } - for (int i = 0; i < retry; i++) { - MetaHistoryDO metaDO = new MetaHistoryDO(); - try { - BeanUtils.populate(metaDO, content); - // 会建立唯一约束,解决: - // 1. 重复的binlog file+offest - // 2. 重复的masterId+timestamp - metaHistoryDAO.insert(metaDO); - } catch (Throwable e) { - if (isUkDuplicateException(e)) { - // 忽略掉重复的位点 - logger.warn("dup apply for sql : " + ddl); - } else { - throw new RuntimeException("apply history to db failed caused by : " + e.getMessage()); - } - + MetaHistoryDO metaDO = new MetaHistoryDO(); + try { + BeanUtils.populate(metaDO, content); + // 会建立唯一约束,解决: + // 1. 重复的binlog file+offest + // 2. 重复的masterId+timestamp + metaHistoryDAO.insert(metaDO); + } catch (Throwable e) { + if (isUkDuplicateException(e)) { + // 忽略掉重复的位点 + logger.warn("dup apply for sql : " + ddl); + } else { + throw new CanalParseException("apply history to db failed caused by : " + e.getMessage()); } - return true; + } - return false; + return true; } /** @@ -290,9 +247,7 @@ public class TableMetaManager implements TableMetaTSDB { } } if (compareAll) { - Map content = new HashMap(); - content.put("binlogFile", position.getFileName()); content.put("binlogOffest", String.valueOf(position.getPosition())); content.put("binlogMasterId", String.valueOf(position.getMasterId())); @@ -302,24 +257,19 @@ public class TableMetaManager implements TableMetaTSDB { throw new RuntimeException("apply failed caused by content is empty in applySnapshotToDB"); } - for (int i = 0; i < retry; i++) { - MetaSnapshotDO snapshotDO = new MetaSnapshotDO(); - try { - BeanUtils.populate(snapshotDO, content); - metaSnapshotDAO.insert(snapshotDO); - } catch (Throwable e) { - if (isUkDuplicateException(e)) { - // 忽略掉重复的位点 - logger.warn("dup apply snapshot for data : " + snapshotDO.getData()); - } else { - throw new RuntimeException("apply failed caused by : " + e.getMessage()); - } + MetaSnapshotDO snapshotDO = new MetaSnapshotDO(); + try { + BeanUtils.populate(snapshotDO, content); + metaSnapshotDAO.insert(snapshotDO); + } catch (Throwable e) { + if (isUkDuplicateException(e)) { + // 忽略掉重复的位点 + logger.warn("dup apply snapshot for data : " + snapshotDO.getData()); + } else { + throw new CanalParseException("apply failed caused by : " + e.getMessage()); } - return true; - } - return false; - + return true; } else { logger.error("compare failed , check log"); } @@ -327,26 +277,15 @@ public class TableMetaManager implements TableMetaTSDB { } private boolean compareTableMetaDbAndMemory(MysqlConnection connection, final String schema, final String table) { - TableMeta tableMetaFromDB = connection.query("desc " + getFullName(schema, table), - new ProcessJdbcResult() { - - @Override - public TableMeta process(ResultSet rs) throws SQLException { - List metas = new ArrayList(); - while (rs.next()) { - FieldMeta meta = new FieldMeta(); - // 做一个优化，使用String.intern()，共享String对象，减少内存使用 - meta.setColumnName(rs.getString("Field")); - meta.setColumnType(rs.getString("Type")); - meta.setNullable(StringUtils.equalsIgnoreCase(rs.getString("Null"), "YES")); - meta.setKey("PRI".equalsIgnoreCase(rs.getString("Key"))); - meta.setDefaultValue(rs.getString("Default")); - metas.add(meta); - } - - return new TableMeta(schema, table, metas); - } - }); + TableMeta tableMetaFromDB = new TableMeta(); + tableMetaFromDB.setSchema(schema); + tableMetaFromDB.setTable(table); + try { + ResultSetPacket packet = connection.query("desc " + getFullName(schema, table)); + tableMetaFromDB.setFields(TableMetaCache.parserTableMeta(packet)); + } catch (IOException e) { + throw new CanalParseException(e); + } TableMeta tableMetaFromMem = memoryTableMeta.find(schema, table); boolean result = compareTableMeta(tableMetaFromMem, tableMetaFromDB); @@ -358,119 +297,70 @@ public class TableMetaManager implements TableMetaTSDB { } private BinlogPosition buildMemFromSnapshot(BinlogPosition position) { + try { + MetaSnapshotDO snapshotDO = metaSnapshotDAO.findByTimestamp(position.getTimestamp()); + String binlogFile = snapshotDO.getBinlogFile(); + Long binlogOffest = snapshotDO.getBinlogOffest(); + String binlogMasterId = snapshotDO.getBinlogMasterId(); + Long binlogTimestamp = snapshotDO.getBinlogTimestamp(); - Map content = new HashMap(); - - content.put("binlogFile", position.getFileName()); - content.put("binlogOffest", String.valueOf(position.getPosition())); - content.put("binlogMasterId", String.valueOf(position.getMasterId())); - content.put("binlogTimestamp", String.valueOf(position.getTimestamp())); - if (content.isEmpty()) { - throw new RuntimeException("apply failed caused by content is empty in buildMemFromSnapshot"); - } - for (int i = 0; i < retry; i++) { - - try { - - String timestamp = content.get("binlogTimestamp"); - MetaSnapshotDO snapshotDO = metaSnapshotDAO.findByTimestamp(Long.valueOf(timestamp)); - JSONObject jsonData = new JSONObject(); - jsonData.put("content", JSON.toJSONString(snapshotDO)); - if (jsonData == null) { - // 可能没有任何snapshot数据 + BinlogPosition snapshotPosition = new BinlogPosition(binlogFile, + binlogOffest == null ? 0l : binlogOffest, + Long.valueOf(binlogMasterId == null ? "-2" : binlogMasterId), + binlogTimestamp == null ? 0l : binlogTimestamp); + // data存储为Map，每个分库一套建表 + String sqlData = snapshotDO.getData(); + JSONObject jsonObj = JSON.parseObject(sqlData); + for (Map.Entry entry : jsonObj.entrySet()) { + // 记录到内存 + if (!memoryTableMeta.apply(snapshotPosition, + ObjectUtils.toString(entry.getKey()), + ObjectUtils.toString(entry.getValue()))) { return null; } - - String binlogFile = jsonData.getString("binlogFile"); - String binlogOffest = jsonData.getString("binlogOffest"); - String binlogMasterId = jsonData.getString("binlogMasterId"); - String binlogTimestamp = jsonData.getString("binlogTimestamp"); - - BinlogPosition snapshotPosition = new BinlogPosition(binlogFile, - Long.valueOf(binlogOffest == null ? "0" : binlogOffest), - Long.valueOf(binlogMasterId == null ? "-2" : binlogMasterId), - Long.valueOf(binlogTimestamp == null ? "0" : binlogTimestamp)); - // data存储为Map，每个分库一套建表 - String sqlData = jsonData.getString("data"); - JSONObject jsonObj = JSON.parseObject(sqlData); - for (Map.Entry entry : jsonObj.entrySet()) { - // 记录到内存 - if (!memoryTableMeta.apply(snapshotPosition, - ObjectUtils.toString(entry.getKey()), - ObjectUtils.toString(entry.getValue()))) { - return null; - } - } - - return snapshotPosition; - - } catch (Throwable e) { - throw new RuntimeException("apply failed caused by : " + e.getMessage()); } + return snapshotPosition; + } catch (Throwable e) { + throw new CanalParseException("apply failed caused by : " + e.getMessage()); } - - return null; - } private boolean applyHistoryOnMemory(BinlogPosition position, BinlogPosition rollbackPosition) { - Map content = new HashMap(); - content.put("binlogSnapshotTimestamp", String.valueOf(position.getTimestamp())); - content.put("binlogFile", rollbackPosition.getFileName()); - content.put("binlogOffest", String.valueOf(rollbackPosition.getPosition())); - content.put("binlogMasterId", String.valueOf(rollbackPosition.getMasterId())); - content.put("binlogTimestamp", String.valueOf(rollbackPosition.getTimestamp())); - String timestamp = content.get("binlogTimestamp"); - String binlogSnapshotTimestamp = content.get("binlogSnapshotTimestamp"); - - for (int i = 0; i < retry; i++) { - try { - List metaHistoryDOList = metaHistoryDAO.findByTimestamp( - Long.valueOf(binlogSnapshotTimestamp), - Long.valueOf(timestamp)); - JSONObject json = new JSONObject(); - json.put("content", JSON.toJSONString(metaHistoryDOList)); - String data = ObjectUtils.toString(json.get("content")); - JSONArray jsonArray = JSON.parseArray(data); - for (Object jsonObj : jsonArray) { - JSONObject jsonData = (JSONObject)jsonObj; - String binlogFile = jsonData.getString("binlogFile"); - String binlogOffest = jsonData.getString("binlogOffest"); - String binlogMasterId = jsonData.getString("binlogMasterId"); - String binlogTimestamp = jsonData.getString("binlogTimestamp"); - String useSchema = jsonData.getString("useSchema"); - String sqlData = jsonData.getString("sql"); - BinlogPosition snapshotPosition = new BinlogPosition(binlogFile, - Long.valueOf(binlogOffest == null ? "0" : binlogOffest), - Long.valueOf(binlogMasterId == null ? "-2" : binlogMasterId), - Long.valueOf(binlogTimestamp == null ? "0" : binlogTimestamp)); - - // 如果是同一秒内,对比一下history的位点，如果比期望的位点要大，忽略之 - if (snapshotPosition.getTimestamp() > rollbackPosition.getTimestamp()) { - continue; - } else if (rollbackPosition.getMasterId() == snapshotPosition.getMasterId() - && snapshotPosition.compareTo(rollbackPosition) > 0) { - continue; - } - - // 记录到内存 - if (!memoryTableMeta.apply(snapshotPosition, useSchema, sqlData)) { - return false; - } + try { + List metaHistoryDOList = metaHistoryDAO.findByTimestamp(position.getTimestamp(), + rollbackPosition.getTimestamp()); + for (MetaHistoryDO metaHistoryDO : metaHistoryDOList) { + String binlogFile = metaHistoryDO.getBinlogFile(); + Long binlogOffest = metaHistoryDO.getBinlogOffest(); + String binlogMasterId = metaHistoryDO.getBinlogMasterId(); + Long binlogTimestamp = metaHistoryDO.getBinlogTimestamp(); + String useSchema = metaHistoryDO.getUseSchema(); + String sqlData = metaHistoryDO.getSql(); + BinlogPosition snapshotPosition = new BinlogPosition(binlogFile, + binlogOffest == null ? 0L : binlogOffest, + Long.valueOf(binlogMasterId == null ? "-2" : binlogMasterId), + binlogTimestamp == null ? 0L : binlogTimestamp); + // 如果是同一秒内,对比一下history的位点，如果比期望的位点要大，忽略之 + if (snapshotPosition.getTimestamp() > rollbackPosition.getTimestamp()) { + continue; + } else if (rollbackPosition.getMasterId() == snapshotPosition.getMasterId() + && snapshotPosition.compareTo(rollbackPosition) > 0) { + continue; } - return jsonArray.size() > 0; - } catch (Throwable e) { - - throw new RuntimeException("apply failed", e); + // 记录到内存 + if (!memoryTableMeta.apply(snapshotPosition, useSchema, sqlData)) { + return false; + } } + return metaHistoryDOList.size() > 0; + } catch (Throwable e) { + throw new CanalParseException("apply failed", e); } - - return false; } private String getFullName(String schema, String table) { 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 054a4b44..46ffafbf 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 @@ -3,18 +3,17 @@ package com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao; import java.util.HashMap; import java.util.List; -import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.model.MetaHistoryDO; - -import com.google.common.collect.Maps; import org.springframework.orm.ibatis.support.SqlMapClientDaoSupport; +import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.model.MetaHistoryDO; +import com.google.common.collect.Maps; + /** * canal数据的存储 * * @author wanshao 2017年7月27日 下午10:51:55 * @since 3.2.5 */ -@SuppressWarnings("deprecation") public class MetaHistoryDAO extends SqlMapClientDaoSupport { public List getAll() { @@ -22,14 +21,14 @@ public class MetaHistoryDAO extends SqlMapClientDaoSupport { } public Long insert(MetaHistoryDO metaDO) { - return (Long)getSqlMapClientTemplate().insert("table_meta_history.insert", metaDO); + return (Long) getSqlMapClientTemplate().insert("table_meta_history.insert", metaDO); } public List findByTimestamp(long snapshotTimestamp, long timestamp) { HashMap params = Maps.newHashMapWithExpectedSize(2); params.put("snapshotTimestamp", snapshotTimestamp); params.put("timestamp", timestamp); - return (List)getSqlMapClientTemplate().queryForList("table_meta_history.findByTimestamp", + return (List) getSqlMapClientTemplate().queryForList("table_meta_history.findByTimestamp", params); } 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 79f1f16f..ca32ebd3 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 @@ -2,11 +2,11 @@ package com.alibaba.otter.canal.parse.inbound.mysql.tsdb.dao; import java.util.HashMap; -import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.model.MetaSnapshotDO; - -import com.google.common.collect.Maps; import org.springframework.orm.ibatis.support.SqlMapClientDaoSupport; +import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.model.MetaSnapshotDO; +import com.google.common.collect.Maps; + /** * canal数据的存储 * @@ -15,6 +15,7 @@ import org.springframework.orm.ibatis.support.SqlMapClientDaoSupport; */ public class MetaSnapshotDAO extends SqlMapClientDaoSupport { + public Long insert(MetaSnapshotDO snapshotDO) { return (Long) getSqlMapClientTemplate().insert("table_meta_snapshot.insert", snapshotDO); } @@ -26,8 +27,7 @@ public class MetaSnapshotDAO extends SqlMapClientDaoSupport { public MetaSnapshotDO findByTimestamp(long timestamp) { HashMap params = Maps.newHashMapWithExpectedSize(2); params.put("timestamp", timestamp); - return (MetaSnapshotDO) getSqlMapClientTemplate().queryForObject("table_meta_snapshot.findByTimestamp", - params); + return (MetaSnapshotDO) getSqlMapClientTemplate().queryForObject("table_meta_snapshot.findByTimestamp", params); } public Integer deleteByTask(String taskName) { diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/model/MetaHistoryDO.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/model/MetaHistoryDO.java index 9b56606d..96108515 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/model/MetaHistoryDO.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/model/MetaHistoryDO.java @@ -11,22 +11,22 @@ public class MetaHistoryDO { /** * 主键 */ - private Long id; + private Long id; /** * 创建时间 */ - private Date gmtCreate; + private Date gmtCreate; /** * 修改时间 */ - private Date gmtModified; + private Date gmtModified; private String binlogFile; - private Long binlogOffest; + private Long binlogOffest; private String binlogMasterId; - private Long binlogTimestamp; + private Long binlogTimestamp; private String useSchema; private String schema; private String table; diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/model/MetaSnapshotDO.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/model/MetaSnapshotDO.java index 4652fe13..8ddfbc64 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/model/MetaSnapshotDO.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/model/MetaSnapshotDO.java @@ -11,22 +11,22 @@ public class MetaSnapshotDO { /** * 主键 */ - private Long id; + private Long id; /** * 创建时间 */ - private Date gmtCreate; + private Date gmtCreate; /** * 修改时间 */ - private Date gmtModified; + private Date gmtModified; private String binlogFile; - private Long binlogOffest; + private Long binlogOffest; private String binlogMasterId; - private Long binlogTimestamp; + private Long binlogTimestamp; private String data; private String extra; diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManager.java b/parse/src/main/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManager.java index ed0b9879..bcc2af7d 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManager.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManager.java @@ -45,6 +45,7 @@ public class FileMixedLogPositionManager extends AbstractLogPositionManager { private ScheduledExecutorService executorService; + @SuppressWarnings("serial") private final LogPosition nullPosition = new LogPosition() { }; diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/index/PeriodMixedLogPositionManager.java b/parse/src/main/java/com/alibaba/otter/canal/parse/index/PeriodMixedLogPositionManager.java index 4a1e60c7..6dbda0a5 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/index/PeriodMixedLogPositionManager.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/index/PeriodMixedLogPositionManager.java @@ -29,6 +29,7 @@ public class PeriodMixedLogPositionManager extends AbstractLogPositionManager { private long period; private Set persistTasks; + @SuppressWarnings("serial") private final LogPosition nullPosition = new LogPosition() { }; diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/TableMetaCacheTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/TableMetaCacheTest.java index 0c09dc0b..8a78e246 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/TableMetaCacheTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/TableMetaCacheTest.java @@ -2,34 +2,43 @@ package com.alibaba.otter.canal.parse.inbound; import java.io.IOException; import java.net.InetSocketAddress; +import java.util.List; -import com.taobao.tddl.dbsync.binlog.BinlogPosition; import org.junit.Assert; import org.junit.Test; -import com.alibaba.druid.sql.visitor.functions.Bin; -import com.alibaba.otter.canal.parse.inbound.TableMeta.FieldMeta; +import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ResultSetPacket; import com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection; -import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.TableMetaCache; public class TableMetaCacheTest { - //@Test - //public void testSimple() { - // - // MysqlConnection connection = new MysqlConnection(new InetSocketAddress("127.0.0.1", 3306), "xxxxx", "xxxxx"); - // try { - // connection.connect(); - // } catch (IOException e) { - // Assert.fail(e.getMessage()); - // } - // - // TableMetaCache cache = new TableMetaCache(connection); - // TableMeta meta = cache.getTableMeta("otter1", "otter_stability1"); - // Assert.assertNotNull(meta); - // for (FieldMeta field : meta.getFields()) { - // System.out.println("filed :" + field.getColumnName() + " , isKey : " + field.isKey() + " , isNull : " - // + field.isNullable()); - // } - //} + @Test + public void testSimple() throws IOException { + + MysqlConnection connection = new MysqlConnection(new InetSocketAddress("100.81.153.8", 3308), + "jingwei", + "jingwei"); + try { + connection.connect(); + } catch (IOException e) { + Assert.fail(e.getMessage()); + } + + List packets = connection.queryMulti("show create table test.ljh_test; show create table test.user"); + String createDDL = null; + if (packets.get(0).getFieldValues().size() > 0) { + createDDL = packets.get(0).getFieldValues().get(1); + } + + System.out.println(createDDL); + + // TableMetaCache cache = new TableMetaCache(connection); + // TableMeta meta = cache.getTableMeta("otter1", "otter_stability1"); + // Assert.assertNotNull(meta); + // for (FieldMeta field : meta.getFields()) { + // System.out.println("filed :" + field.getColumnName() + " , isKey : " + // + field.isKey() + " , isNull : " + // + field.isNullable()); + // } + } } diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/group/GroupEventPaserTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/group/GroupEventPaserTest.java index 5d3e8920..57e26784 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/group/GroupEventPaserTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/group/GroupEventPaserTest.java @@ -88,7 +88,6 @@ public class GroupEventPaserTest { private BinlogParser buildParser(AuthenticationInfo info) { return new AbstractBinlogParser() { - @Override public Entry parse(LogEvent event, boolean isSeek) throws CanalParseException { return null; diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/SimpleDdlParserTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/SimpleDdlParserTest.java index 50dce482..f2bb1ce4 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/SimpleDdlParserTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/SimpleDdlParserTest.java @@ -3,8 +3,8 @@ package com.alibaba.otter.canal.parse.inbound.mysql; import org.junit.Assert; import org.junit.Test; -import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.SimpleDdlParser; -import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.SimpleDdlParser.DdlResult; +import com.alibaba.otter.canal.parse.inbound.mysql.ddl.DdlResult; +import com.alibaba.otter.canal.parse.inbound.mysql.ddl.SimpleDdlParser; import com.alibaba.otter.canal.protocol.CanalEntry.EventType; public class SimpleDdlParserTest { diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMetaTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMetaTest.java index c928b667..8ead5e2a 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMetaTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMetaTest.java @@ -19,7 +19,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; * @author agapple 2017年8月1日 下午7:15:54 */ @RunWith(SpringJUnit4ClassRunner.class) -@ContextConfiguration(locations = {"/dal-dao.xml"}) +@ContextConfiguration(locations = { "/dal-dao.xml" }) public class MemoryTableMetaTest { @Test @@ -33,6 +33,6 @@ public class MemoryTableMetaTest { TableMeta meta = memoryTableMeta.find("test", "test"); System.out.println(meta); - Assert.assertTrue( meta.getFieldMetaByName("ID").isKey()); + Assert.assertTrue(meta.getFieldMetaByName("ID").isKey()); } } diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MetaHistoryDAOTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MetaHistoryDAOTest.java index 30ae1e66..29c02f7e 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MetaHistoryDAOTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MetaHistoryDAOTest.java @@ -13,12 +13,10 @@ import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** - * Created by wanshao - * Date: 2017/9/20 - * Time: 下午5:00 + * Created by wanshao Date: 2017/9/20 Time: 下午5:00 **/ @RunWith(SpringJUnit4ClassRunner.class) -@ContextConfiguration(locations = {"/dal-dao.xml"}) +@ContextConfiguration(locations = { "/dal-dao.xml" }) public class MetaHistoryDAOTest { @Resource @@ -26,8 +24,7 @@ public class MetaHistoryDAOTest { @Test public void testGetAll() { - List - metaHistoryDOList = metaHistoryDAO.getAll(); + List metaHistoryDOList = metaHistoryDAO.getAll(); for (MetaHistoryDO metaHistoryDO : metaHistoryDOList) { System.out.println(metaHistoryDO.getId()); } 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 1eb45d7d..13413d81 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 @@ -30,8 +30,6 @@ public class TableMetaManagerTest { @Test public void testSimple() throws FileNotFoundException, IOException { - - URL url = Thread.currentThread().getContextClassLoader().getResource("dummy.txt"); File dummyFile = new File(url.getFile()); File create = new File(dummyFile.getParent() + "/ddl", "create.sql"); diff --git a/parse/src/test/resources/dal-dao.xml b/parse/src/test/resources/dal-dao.xml deleted file mode 100644 index 7d2fd0a4..00000000 --- a/parse/src/test/resources/dal-dao.xml +++ /dev/null @@ -1,46 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/parse/src/main/resources/dal-dao.xml b/parse/src/test/resources/tsdb/dal-dao.xml similarity index 83% rename from parse/src/main/resources/dal-dao.xml rename to parse/src/test/resources/tsdb/dal-dao.xml index 942de654..1842641f 100644 --- a/parse/src/main/resources/dal-dao.xml +++ b/parse/src/test/resources/tsdb/dal-dao.xml @@ -9,13 +9,10 @@ - + - - - @@ -32,11 +29,11 @@ - + - + diff --git a/parse/src/main/resources/mybatis-config.xml b/parse/src/test/resources/tsdb/mybatis-config.xml similarity index 100% rename from parse/src/main/resources/mybatis-config.xml rename to parse/src/test/resources/tsdb/mybatis-config.xml diff --git a/deployer/src/main/resources/spring/sql-map/sqlmap_history.xml b/parse/src/test/resources/tsdb/sql-map/sqlmap_history.xml similarity index 100% rename from deployer/src/main/resources/spring/sql-map/sqlmap_history.xml rename to parse/src/test/resources/tsdb/sql-map/sqlmap_history.xml diff --git a/parse/src/main/resources/sql-map/sqlmap_snapshot.xml b/parse/src/test/resources/tsdb/sql-map/sqlmap_snapshot.xml similarity index 100% rename from parse/src/main/resources/sql-map/sqlmap_snapshot.xml rename to parse/src/test/resources/tsdb/sql-map/sqlmap_snapshot.xml diff --git a/parse/src/main/resources/sqlmap-config.xml b/parse/src/test/resources/tsdb/sqlmap-config.xml similarity index 100% rename from parse/src/main/resources/sqlmap-config.xml rename to parse/src/test/resources/tsdb/sqlmap-config.xml