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