From 8626c6ee247cb27cea657f1d598c19e1ef0c22da Mon Sep 17 00:00:00 2001 From: lubiao Date: Tue, 11 Oct 2016 18:46:30 +0800 Subject: [PATCH] =?UTF-8?q?1=E3=80=81SessionHandler=E7=B1=BB=E4=B8=AD?= =?UTF-8?q?=E6=89=A7=E8=A1=8CSUBSCRIPTION=E6=97=B6=EF=BC=8C=E5=BA=94?= =?UTF-8?q?=E8=AF=A5=E5=85=88=E5=B0=9D=E8=AF=95=E5=90=AF=E5=8A=A8runningMo?= =?UTF-8?q?nitor=E5=86=8D=E8=B0=83=E7=94=A8embeddedServer=E7=9A=84subscrib?= =?UTF-8?q?e=E6=96=B9=E6=B3=95=202=E3=80=81embeddedServer=E7=9A=84subscrib?= =?UTF-8?q?e=E6=96=B9=E6=B3=95=E4=B8=AD=E5=BA=94=E8=AF=A5=E9=A6=96?= =?UTF-8?q?=E5=85=88=E8=BF=9B=E8=A1=8CcheckStart=E6=93=8D=E4=BD=9C?= =?UTF-8?q?=EF=BC=8C=E5=90=A6=E5=88=99=E5=8F=AF=E8=83=BD=E5=BC=95=E8=B5=B7?= =?UTF-8?q?metaManager=E7=9A=84=E8=84=8F=E5=90=AF=E5=8A=A8=203=E3=80=81Ses?= =?UTF-8?q?sionHandler=E4=B8=AD=E5=A2=9E=E5=8A=A0=E6=97=A5=E5=BF=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../otter/canal/parse/inbound/mysql/MysqlEventParser.java | 3 ++- .../otter/canal/server/embedded/CanalServerWithEmbedded.java | 2 ++ .../otter/canal/server/netty/handler/SessionHandler.java | 5 +++-- 3 files changed, 7 insertions(+), 3 deletions(-) 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 548df70c..e4f437f8 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; @@ -410,7 +411,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/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);