mirror of
https://github.com/alibaba/canal.git
synced 2026-11-04 00:30:15 +08:00
1、BinLogFileQueue的listBinlogFiles方法中对binlog文件名进行更严格的验证,必须以数字结尾
2、LocalBinLogConnection类中解决如下bug 2.1第二个dump方法中增加代码"decoder.handle(LogEvent.FORMAT_DESCRIPTION_EVENT);",解决"((QueryLogEvent)event).getQuery()"语句出现乱码的问题(BEGIN后面多了4个字节),出现乱码不会进入if语句,导致找到的binlog位点不准确 2.2第一个dump方法中,被注释的decode代码才是正确的,和DirectLogFetcher不同,此处需要构造子循环进行遍历 2.3第一个dump方法中,将每个binlog-file的首个event设置为RotateLogEvent,否则会导致所有event的binlogFileName都是默认的mysql-bin.000001。这么改完之后,和DirectLogFetcher的行为相契合,后者在每个binlogfile的首个event就是RotateLogEvent 3、LocalBinlogEventParser中增加元数据处理的相关代码,不加的话,所有的event都不带列名,基本上没法儿使用
This commit is contained in:
@@ -8,7 +8,6 @@ import org.apache.commons.lang.StringUtils;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import com.alibaba.otter.canal.parse.exception.CanalParseException;
|
||||
import com.alibaba.otter.canal.parse.inbound.ErosaConnection;
|
||||
import com.alibaba.otter.canal.parse.inbound.SinkFunction;
|
||||
import com.alibaba.otter.canal.parse.inbound.mysql.local.BinLogFileQueue;
|
||||
@@ -18,6 +17,7 @@ import com.taobao.tddl.dbsync.binlog.LogDecoder;
|
||||
import com.taobao.tddl.dbsync.binlog.LogEvent;
|
||||
import com.taobao.tddl.dbsync.binlog.LogPosition;
|
||||
import com.taobao.tddl.dbsync.binlog.event.QueryLogEvent;
|
||||
import com.taobao.tddl.dbsync.binlog.event.RotateLogEvent;
|
||||
|
||||
/**
|
||||
* local bin log connection (not real connection)
|
||||
@@ -83,9 +83,9 @@ public class LocalBinLogConnection implements ErosaConnection {
|
||||
context.setLogPosition(new LogPosition(binlogfilename, binlogPosition));
|
||||
while (running) {
|
||||
boolean needContinue = true;
|
||||
LogEvent event;
|
||||
while (fetcher.fetch()) {
|
||||
event = decoder.decode(fetcher, context);
|
||||
LogEvent event = null;
|
||||
L:while (fetcher.fetch()) {
|
||||
/*event = decoder.decode(fetcher, context);
|
||||
if (event == null) {
|
||||
throw new CanalParseException("parse failed");
|
||||
}
|
||||
@@ -93,15 +93,20 @@ public class LocalBinLogConnection implements ErosaConnection {
|
||||
if (!func.sink(event)) {
|
||||
needContinue = false;
|
||||
break;
|
||||
}
|
||||
}*/
|
||||
|
||||
// do {
|
||||
// event = decoder.decode(fetcher, context);
|
||||
// if (event != null && !func.sink(event)) {
|
||||
// needContinue = false;
|
||||
// break;
|
||||
// }
|
||||
// } while (event != null);
|
||||
do {
|
||||
if (event == null) {
|
||||
event = new RotateLogEvent(context.getLogPosition().getFileName(), context.getLogPosition().getPosition());
|
||||
} else {
|
||||
event = decoder.decode(fetcher, context);
|
||||
}
|
||||
|
||||
if (event != null && !func.sink(event)) {
|
||||
needContinue = false;
|
||||
break L;
|
||||
}
|
||||
} while (event != null);
|
||||
}
|
||||
|
||||
if (needContinue) {// 读取下一个
|
||||
@@ -145,6 +150,7 @@ public class LocalBinLogConnection implements ErosaConnection {
|
||||
|
||||
FileLogFetcher fetcher = new FileLogFetcher(bufferSize);
|
||||
LogDecoder decoder = new LogDecoder();
|
||||
decoder.handle(LogEvent.FORMAT_DESCRIPTION_EVENT);
|
||||
decoder.handle(LogEvent.QUERY_EVENT);
|
||||
decoder.handle(LogEvent.XID_EVENT);
|
||||
LogContext context = new LogContext();
|
||||
|
||||
@@ -1,10 +1,14 @@
|
||||
package com.alibaba.otter.canal.parse.inbound.mysql;
|
||||
|
||||
import java.io.IOException;
|
||||
|
||||
import org.apache.commons.lang.StringUtils;
|
||||
|
||||
import com.alibaba.otter.canal.parse.CanalEventParser;
|
||||
import com.alibaba.otter.canal.parse.exception.CanalParseException;
|
||||
import com.alibaba.otter.canal.parse.inbound.ErosaConnection;
|
||||
import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert;
|
||||
import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.TableMetaCache;
|
||||
import com.alibaba.otter.canal.parse.index.CanalLogPositionManager;
|
||||
import com.alibaba.otter.canal.parse.support.AuthenticationInfo;
|
||||
import com.alibaba.otter.canal.protocol.position.EntryPosition;
|
||||
@@ -21,7 +25,9 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements
|
||||
// 数据库信息
|
||||
private AuthenticationInfo masterInfo;
|
||||
private EntryPosition masterPosition; // binlog信息
|
||||
|
||||
private MysqlConnection metaConnection; // 查询meta信息的链接
|
||||
private TableMetaCache tableMetaCache; // 对应meta
|
||||
|
||||
private String directory;
|
||||
private boolean needWait = false;
|
||||
private int bufferSize = 16 * 1024;
|
||||
@@ -33,17 +39,60 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements
|
||||
@Override
|
||||
protected ErosaConnection buildErosaConnection() {
|
||||
return buildLocalBinLogConnection();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void preDump(ErosaConnection connection) {
|
||||
metaConnection = buildMysqlConnection();
|
||||
try {
|
||||
metaConnection.connect();
|
||||
} catch (IOException e) {
|
||||
throw new CanalParseException(e);
|
||||
}
|
||||
|
||||
tableMetaCache = new TableMetaCache(metaConnection);
|
||||
((LogEventConvert) binlogParser).setTableMetaCache(tableMetaCache);
|
||||
}
|
||||
|
||||
public void start() throws CanalParseException {
|
||||
@Override
|
||||
protected void afterDump(ErosaConnection connection) {
|
||||
if (metaConnection != null) {
|
||||
try {
|
||||
metaConnection.disconnect();
|
||||
} catch (IOException e) {
|
||||
logger.error("ERROR # disconnect meta connection for address:{}", metaConnection.getConnector()
|
||||
.getAddress(), e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public void start() throws CanalParseException {
|
||||
if (runningInfo == null) { // 第一次链接主库
|
||||
runningInfo = masterInfo;
|
||||
}
|
||||
|
||||
super.start();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void stop() {
|
||||
if (metaConnection != null) {
|
||||
try {
|
||||
metaConnection.disconnect();
|
||||
} catch (IOException e) {
|
||||
logger.error("ERROR # disconnect meta connection for address:{}", metaConnection.getConnector()
|
||||
.getAddress(), e);
|
||||
}
|
||||
}
|
||||
|
||||
private ErosaConnection buildLocalBinLogConnection() {
|
||||
if (tableMetaCache != null) {
|
||||
tableMetaCache.clearTableMeta();
|
||||
}
|
||||
|
||||
super.stop();
|
||||
}
|
||||
|
||||
private ErosaConnection buildLocalBinLogConnection() {
|
||||
LocalBinLogConnection connection = new LocalBinLogConnection();
|
||||
|
||||
connection.setBufferSize(this.bufferSize);
|
||||
@@ -52,6 +101,19 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements
|
||||
|
||||
return connection;
|
||||
}
|
||||
|
||||
private MysqlConnection buildMysqlConnection() {
|
||||
MysqlConnection connection = new MysqlConnection(runningInfo.getAddress(),
|
||||
runningInfo.getUsername(),
|
||||
runningInfo.getPassword(),
|
||||
connectionCharsetNumber,
|
||||
runningInfo.getDefaultDatabaseName());
|
||||
connection.getConnector().setReceiveBufferSize(64 * 1024);
|
||||
connection.getConnector().setSendBufferSize(64 * 1024);
|
||||
connection.getConnector().setSoTimeout(30 * 1000);
|
||||
connection.setCharset(connectionCharset);
|
||||
return connection;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected EntryPosition findStartPosition(ErosaConnection connection) {
|
||||
|
||||
@@ -9,6 +9,8 @@ import java.util.Timer;
|
||||
import java.util.TimerTask;
|
||||
import java.util.concurrent.locks.Condition;
|
||||
import java.util.concurrent.locks.ReentrantLock;
|
||||
import java.util.regex.Matcher;
|
||||
import java.util.regex.Pattern;
|
||||
|
||||
import org.apache.commons.io.FileUtils;
|
||||
import org.apache.commons.io.filefilter.IOFileFilter;
|
||||
@@ -188,7 +190,9 @@ public class BinLogFileQueue {
|
||||
files.addAll(FileUtils.listFiles(directory, new IOFileFilter() {
|
||||
|
||||
public boolean accept(File file) {
|
||||
return file.getName().startsWith(baseName);
|
||||
Pattern pattern = Pattern.compile("\\d+$");
|
||||
Matcher matcher = pattern.matcher(file.getName());
|
||||
return file.getName().startsWith(baseName) && matcher.find();
|
||||
}
|
||||
|
||||
public boolean accept(File dir, String name) {
|
||||
|
||||
Reference in New Issue
Block a user