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/MysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java index ca892a9f..066a377b 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 @@ -13,6 +13,7 @@ import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.exception.ExceptionUtils; import org.springframework.util.CollectionUtils; +import com.alibaba.otter.canal.common.utils.JsonUtils; import com.alibaba.otter.canal.parse.CanalEventParser; import com.alibaba.otter.canal.parse.CanalHASwitchable; import com.alibaba.otter.canal.parse.driver.mysql.packets.server.FieldPacket; @@ -412,7 +413,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE } } // 其余情况 - logger.warn("prepare to find start position just last position"); + logger.warn("prepare to find start position just last position\n {}",JsonUtils.marshalToString(logPosition)); return logPosition.getPostion(); } else { // 针对切换的情况，考虑回退时间 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) { diff --git a/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java b/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java index d8488668..7123e389 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java @@ -131,6 +131,8 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C */ @Override public void subscribe(ClientIdentity clientIdentity) throws CanalServerException { + checkStart(clientIdentity.getDestination()); + CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination()); if (!canalInstance.getMetaManager().isStart()) { canalInstance.getMetaManager().start(); diff --git a/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java b/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java index ce0a0790..4cefb90c 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java @@ -66,8 +66,7 @@ public class SessionHandler extends SimpleChannelHandler { Short.valueOf(sub.getClientId()), sub.getFilter()); MDC.put("destination", clientIdentity.getDestination()); - embeddedServer.subscribe(clientIdentity); - + // 尝试启动，如果已经启动，忽略 if (!embeddedServer.isStart(clientIdentity.getDestination())) { ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(clientIdentity.getDestination()); @@ -75,6 +74,8 @@ public class SessionHandler extends SimpleChannelHandler { runningMonitor.start(); } } + + embeddedServer.subscribe(clientIdentity); ctx.setAttachment(clientIdentity);// 设置状态数据 NettyUtils.ack(ctx.getChannel(), null);