From 4898ed7b834746cc13322e53a7faa9a7518c8f1c Mon Sep 17 00:00:00 2001 From: lulu2panpan Date: Fri, 29 Apr 2016 09:40:27 +0800 Subject: [PATCH] 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);