From 350f8637c3d54e4745b26d1bd07ea6c4c0af0607 Mon Sep 17 00:00:00 2001 From: lubiao Date: Tue, 11 Oct 2016 17:28:24 +0800 Subject: [PATCH] =?UTF-8?q?1=E3=80=81BinLogFileQueue=E7=9A=84listBinlogFil?= =?UTF-8?q?es=E6=96=B9=E6=B3=95=E4=B8=AD=E5=AF=B9binlog=E6=96=87=E4=BB=B6?= =?UTF-8?q?=E5=90=8D=E8=BF=9B=E8=A1=8C=E6=9B=B4=E4=B8=A5=E6=A0=BC=E7=9A=84?= =?UTF-8?q?=E9=AA=8C=E8=AF=81=EF=BC=8C=E5=BF=85=E9=A1=BB=E4=BB=A5=E6=95=B0?= =?UTF-8?q?=E5=AD=97=E7=BB=93=E5=B0=BE=202=E3=80=81LocalBinLogConnection?= =?UTF-8?q?=E7=B1=BB=E4=B8=AD=E8=A7=A3=E5=86=B3=E5=A6=82=E4=B8=8Bbug=20=09?= =?UTF-8?q?2.1=E7=AC=AC=E4=BA=8C=E4=B8=AAdump=E6=96=B9=E6=B3=95=E4=B8=AD?= =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E4=BB=A3=E7=A0=81"decoder.handle(LogEvent.FO?= =?UTF-8?q?RMAT=5FDESCRIPTION=5FEVENT);",=E8=A7=A3=E5=86=B3"((QueryLogEven?= =?UTF-8?q?t)event).getQuery()"=E8=AF=AD=E5=8F=A5=E5=87=BA=E7=8E=B0?= =?UTF-8?q?=E4=B9=B1=E7=A0=81=E7=9A=84=E9=97=AE=E9=A2=98(BEGIN=E5=90=8E?= =?UTF-8?q?=E9=9D=A2=E5=A4=9A=E4=BA=864=E4=B8=AA=E5=AD=97=E8=8A=82),?= =?UTF-8?q?=E5=87=BA=E7=8E=B0=E4=B9=B1=E7=A0=81=E4=B8=8D=E4=BC=9A=E8=BF=9B?= =?UTF-8?q?=E5=85=A5if=E8=AF=AD=E5=8F=A5=EF=BC=8C=E5=AF=BC=E8=87=B4?= =?UTF-8?q?=E6=89=BE=E5=88=B0=E7=9A=84binlog=E4=BD=8D=E7=82=B9=E4=B8=8D?= =?UTF-8?q?=E5=87=86=E7=A1=AE=20=092.2=E7=AC=AC=E4=B8=80=E4=B8=AAdump?= =?UTF-8?q?=E6=96=B9=E6=B3=95=E4=B8=AD=EF=BC=8C=E8=A2=AB=E6=B3=A8=E9=87=8A?= =?UTF-8?q?=E7=9A=84decode=E4=BB=A3=E7=A0=81=E6=89=8D=E6=98=AF=E6=AD=A3?= =?UTF-8?q?=E7=A1=AE=E7=9A=84=EF=BC=8C=E5=92=8CDirectLogFetcher=E4=B8=8D?= =?UTF-8?q?=E5=90=8C=EF=BC=8C=E6=AD=A4=E5=A4=84=E9=9C=80=E8=A6=81=E6=9E=84?= =?UTF-8?q?=E9=80=A0=E5=AD=90=E5=BE=AA=E7=8E=AF=E8=BF=9B=E8=A1=8C=E9=81=8D?= =?UTF-8?q?=E5=8E=86=20=092.3=E7=AC=AC=E4=B8=80=E4=B8=AAdump=E6=96=B9?= =?UTF-8?q?=E6=B3=95=E4=B8=AD=EF=BC=8C=E5=B0=86=E6=AF=8F=E4=B8=AAbinlog-fi?= =?UTF-8?q?le=E7=9A=84=E9=A6=96=E4=B8=AAevent=E8=AE=BE=E7=BD=AE=E4=B8=BARo?= =?UTF-8?q?tateLogEvent=EF=BC=8C=E5=90=A6=E5=88=99=E4=BC=9A=E5=AF=BC?= =?UTF-8?q?=E8=87=B4=E6=89=80=E6=9C=89event=E7=9A=84binlogFileName?= =?UTF-8?q?=E9=83=BD=E6=98=AF=E9=BB=98=E8=AE=A4=E7=9A=84mysql-bin.000001?= =?UTF-8?q?=E3=80=82=E8=BF=99=E4=B9=88=E6=94=B9=E5=AE=8C=E4=B9=8B=E5=90=8E?= =?UTF-8?q?=EF=BC=8C=E5=92=8CDirectLogFetcher=E7=9A=84=E8=A1=8C=E4=B8=BA?= =?UTF-8?q?=E7=9B=B8=E5=A5=91=E5=90=88=EF=BC=8C=E5=90=8E=E8=80=85=E5=9C=A8?= =?UTF-8?q?=E6=AF=8F=E4=B8=AAbinlogfile=E7=9A=84=E9=A6=96=E4=B8=AAevent?= =?UTF-8?q?=E5=B0=B1=E6=98=AFRotateLogEvent=203=E3=80=81LocalBinlogEventPa?= =?UTF-8?q?rser=E4=B8=AD=E5=A2=9E=E5=8A=A0=E5=85=83=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E5=A4=84=E7=90=86=E7=9A=84=E7=9B=B8=E5=85=B3=E4=BB=A3=E7=A0=81?= =?UTF-8?q?=EF=BC=8C=E4=B8=8D=E5=8A=A0=E7=9A=84=E8=AF=9D=EF=BC=8C=E6=89=80?= =?UTF-8?q?=E6=9C=89=E7=9A=84event=E9=83=BD=E4=B8=8D=E5=B8=A6=E5=88=97?= =?UTF-8?q?=E5=90=8D=EF=BC=8C=E5=9F=BA=E6=9C=AC=E4=B8=8A=E6=B2=A1=E6=B3=95?= =?UTF-8?q?=E5=84=BF=E4=BD=BF=E7=94=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../main/resources/spring/local-instance.xml | 163 ++++++++++++++++++ .../inbound/mysql/LocalBinLogConnection.java | 30 ++-- .../inbound/mysql/LocalBinlogEventParser.java | 70 +++++++- .../inbound/mysql/local/BinLogFileQueue.java | 6 +- 4 files changed, 252 insertions(+), 17 deletions(-) create mode 100644 deployer/src/main/resources/spring/local-instance.xml diff --git a/deployer/src/main/resources/spring/local-instance.xml b/deployer/src/main/resources/spring/local-instance.xml new file mode 100644 index 00000000..52cdf404 --- /dev/null +++ b/deployer/src/main/resources/spring/local-instance.xml @@ -0,0 +1,163 @@ + + + + + + + + + + classpath:canal.properties + classpath:${canal.instance.destination:}/instance.properties + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + ${canal.zkServers:127.0.0.1:2181} + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java index dc821ce4..49c8323f 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java @@ -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(); 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 4f139e3a..3a22e8d0 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 @@ -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) { diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/local/BinLogFileQueue.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/local/BinLogFileQueue.java index a5a5dadb..6e3d9142 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/local/BinLogFileQueue.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/local/BinLogFileQueue.java @@ -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) {