From 8ded94a09e55edfbf96478b27e15cf23fb631791 Mon Sep 17 00:00:00 2001 From: lulu2panpan Date: Fri, 29 Apr 2016 09:28:12 +0800 Subject: [PATCH 1/5] Update ClusterCanalConnector.java --- .../otter/canal/client/impl/ClusterCanalConnector.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/client/src/main/java/com/alibaba/otter/canal/client/impl/ClusterCanalConnector.java b/client/src/main/java/com/alibaba/otter/canal/client/impl/ClusterCanalConnector.java index 7309f326..10bcf871 100644 --- a/client/src/main/java/com/alibaba/otter/canal/client/impl/ClusterCanalConnector.java +++ b/client/src/main/java/com/alibaba/otter/canal/client/impl/ClusterCanalConnector.java @@ -45,7 +45,7 @@ public class ClusterCanalConnector implements CanalConnector { int times = 0; while (true) { try { - currentConnector = new SimpleCanalConnector(nextAddress, username, password, destination); + currentConnector = new SimpleCanalConnector(nextAddress, username, password, destination, ,accessStrategy); currentConnector.setSoTimeout(soTimeout); if (filter != null) { currentConnector.setFilter(filter); @@ -57,7 +57,7 @@ public class ClusterCanalConnector implements CanalConnector { currentConnector.connect(); break; } catch (Exception e) { - logger.warn("failed to connect to:{} after retry {} times", nextAddress, times); + logger.warn("failed to connect to:{} after retry {} times", currentConnector.getAddress(), times); currentConnector.disconnect(); currentConnector = null; // retry for #retryTimes for each node when trying to From e614c6141d6a16eb0114fc029e06a52358f84926 Mon Sep 17 00:00:00 2001 From: lulu2panpan Date: Fri, 29 Apr 2016 09:28:48 +0800 Subject: [PATCH 2/5] Update ClusterCanalConnector.java --- .../alibaba/otter/canal/client/impl/ClusterCanalConnector.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/client/src/main/java/com/alibaba/otter/canal/client/impl/ClusterCanalConnector.java b/client/src/main/java/com/alibaba/otter/canal/client/impl/ClusterCanalConnector.java index 10bcf871..cff34bb9 100644 --- a/client/src/main/java/com/alibaba/otter/canal/client/impl/ClusterCanalConnector.java +++ b/client/src/main/java/com/alibaba/otter/canal/client/impl/ClusterCanalConnector.java @@ -45,7 +45,7 @@ public class ClusterCanalConnector implements CanalConnector { int times = 0; while (true) { try { - currentConnector = new SimpleCanalConnector(nextAddress, username, password, destination, ,accessStrategy); + currentConnector = new SimpleCanalConnector(nextAddress, username, password, destination, accessStrategy); currentConnector.setSoTimeout(soTimeout); if (filter != null) { currentConnector.setFilter(filter); From 11e1d1d6000be292d5c38d21a11c1a71583ecbdd Mon Sep 17 00:00:00 2001 From: lulu2panpan Date: Fri, 29 Apr 2016 09:31:40 +0800 Subject: [PATCH 3/5] Update CanalConnectors.java --- .../java/com/alibaba/otter/canal/client/CanalConnectors.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/client/src/main/java/com/alibaba/otter/canal/client/CanalConnectors.java b/client/src/main/java/com/alibaba/otter/canal/client/CanalConnectors.java index 391b21dc..de0aa2e5 100644 --- a/client/src/main/java/com/alibaba/otter/canal/client/CanalConnectors.java +++ b/client/src/main/java/com/alibaba/otter/canal/client/CanalConnectors.java @@ -28,7 +28,7 @@ public class CanalConnectors { */ public static CanalConnector newSingleConnector(SocketAddress address, String destination, String username, String password) { - SimpleCanalConnector canalConnector = new SimpleCanalConnector(address, username, password, destination); + SimpleCanalConnector canalConnector = new SimpleCanalConnector(address, username, password, destination, null); canalConnector.setSoTimeout(30 * 1000); return canalConnector; } From 4898ed7b834746cc13322e53a7faa9a7518c8f1c Mon Sep 17 00:00:00 2001 From: lulu2panpan Date: Fri, 29 Apr 2016 09:40:27 +0800 Subject: [PATCH 4/5] Update SimpleCanalConnector.java --- .../canal/client/impl/SimpleCanalConnector.java | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/client/src/main/java/com/alibaba/otter/canal/client/impl/SimpleCanalConnector.java b/client/src/main/java/com/alibaba/otter/canal/client/impl/SimpleCanalConnector.java index a432e668..2dc44c7d 100644 --- a/client/src/main/java/com/alibaba/otter/canal/client/impl/SimpleCanalConnector.java +++ b/client/src/main/java/com/alibaba/otter/canal/client/impl/SimpleCanalConnector.java @@ -14,6 +14,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.alibaba.otter.canal.client.CanalConnector; +import com.alibaba.otter.canal.client.CanalNodeAccessStrategy; import com.alibaba.otter.canal.client.impl.running.ClientRunningData; import com.alibaba.otter.canal.client.impl.running.ClientRunningListener; import com.alibaba.otter.canal.client.impl.running.ClientRunningMonitor; @@ -69,18 +70,21 @@ public class SimpleCanalConnector implements CanalConnector { // 读写数据分别使用不同的锁进行控制，减小锁粒度,读也需要排他锁，并发度容易造成数据包混乱，反序列化失败 private Object readDataLock = new Object(); private Object writeDataLock = new Object(); + private CanalNodeAccessStrategy accessStrategy; - public SimpleCanalConnector(SocketAddress address, String username, String password, String destination){ - this(address, username, password, destination, 60000); + public SimpleCanalConnector(SocketAddress address, String username, String password, String destination, + CanalNodeAccessStrategy accessStrategy){ + this(address, username, password, destination, 60000, accessStrategy); } public SimpleCanalConnector(SocketAddress address, String username, String password, String destination, - int soTimeout){ + int soTimeout, CanalNodeAccessStrategy accessStrategy){ this.address = address; this.username = username; this.password = password; this.soTimeout = soTimeout; this.clientIdentity = new ClientIdentity(destination, (short) 1001); + this.accessStrategy = accessStrategy; } public void connect() throws CanalClientException { @@ -123,6 +127,9 @@ public class SimpleCanalConnector implements CanalConnector { private InetSocketAddress doConnect() throws CanalClientException { try { + if (accessStrategy != null) { + address = accessStrategy.nextNode();//实时获取一下最新的running节点，lubiao + } channel = SocketChannel.open(); channel.socket().setSoTimeout(soTimeout); channel.connect(address); From d7d969915e8e6fd0c96dbad3050234dcfc86f7a8 Mon Sep 17 00:00:00 2001 From: lulu2panpan Date: Fri, 29 Apr 2016 09:50:37 +0800 Subject: [PATCH 5/5] Update ClientRunningMonitor.java --- .../impl/running/ClientRunningMonitor.java | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/client/src/main/java/com/alibaba/otter/canal/client/impl/running/ClientRunningMonitor.java b/client/src/main/java/com/alibaba/otter/canal/client/impl/running/ClientRunningMonitor.java index 762b9949..7bcebb53 100644 --- a/client/src/main/java/com/alibaba/otter/canal/client/impl/running/ClientRunningMonitor.java +++ b/client/src/main/java/com/alibaba/otter/canal/client/impl/running/ClientRunningMonitor.java @@ -1,6 +1,7 @@ package com.alibaba.otter.canal.client.impl.running; import java.net.InetSocketAddress; +import java.text.MessageFormat; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; @@ -96,7 +97,11 @@ public class ClientRunningMonitor extends AbstractCanalLifeCycle { releaseRunning(); // 尝试一下release } - public void initRunning() { + // 改动记录： + // 1,在方法上加synchronized关键字,保证同步顺序执行; + // 2,判断Zk上已经存在的activeData是否是本机，是的话把mutex重置为true，否则会导致死锁 + // 3,增加异常处理，保证出现异常时，running节点能被删除,否则会导致死锁 + public synchronized void initRunning() { if (!isStart()) { return; } @@ -116,12 +121,19 @@ public class ClientRunningMonitor extends AbstractCanalLifeCycle { initRunning(); } else { activeData = JsonUtils.unmarshalFromByte(bytes, ClientRunningData.class); + if(activeData.getAddress().contains(":") && isMine(activeData.getAddress())){ + mutex.set(true); + } } } catch (ZkNoNodeException e) { zkClient.createPersistent(ZookeeperPathUtils.getClientIdNodePath(this.destination, clientData.getClientId()), true); // 尝试创建父节点 initRunning(); - } + } catch (Throwable t) { + logger.error(MessageFormat.format("There is an error when execute initRunning method, with destination [{0}].",destination),t); + zkClient.delete(path); + throw new RuntimeException("something goes wrong in initRunning method. ",t); + } } /**