chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

Update SimpleCanalConnector.java

Browse Source
This commit is contained in:
lulu2panpan
2016-04-29 09:40:27 +08:00
parent 11e1d1d600
commit 4898ed7b83
1 changed files with 10 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
client/src/main/java/com/alibaba/otter/canal/client/impl/SimpleCanalConnector.java
View File

@@ -14,6 +14,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.client.CanalConnector; 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.ClientRunningData;
import com.alibaba.otter.canal.client.impl.running.ClientRunningListener; import com.alibaba.otter.canal.client.impl.running.ClientRunningListener;
import com.alibaba.otter.canal.client.impl.running.ClientRunningMonitor; 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 readDataLock = new Object();
private Object writeDataLock = new Object(); private Object writeDataLock = new Object();
private CanalNodeAccessStrategy accessStrategy;
public SimpleCanalConnector(SocketAddress address, String username, String password, String destination){ public SimpleCanalConnector(SocketAddress address, String username, String password, String destination,
this(address, username, password, destination, 60000); CanalNodeAccessStrategy accessStrategy){
this(address, username, password, destination, 60000, accessStrategy);
} }
public SimpleCanalConnector(SocketAddress address, String username, String password, String destination, public SimpleCanalConnector(SocketAddress address, String username, String password, String destination,
int soTimeout){ int soTimeout, CanalNodeAccessStrategy accessStrategy){
this.address = address; this.address = address;
this.username = username; this.username = username;
this.password = password; this.password = password;
this.soTimeout = soTimeout; this.soTimeout = soTimeout;
this.clientIdentity = new ClientIdentity(destination, (short) 1001); this.clientIdentity = new ClientIdentity(destination, (short) 1001);
this.accessStrategy = accessStrategy;
} }
public void connect() throws CanalClientException { public void connect() throws CanalClientException {
@@ -123,6 +127,9 @@ public class SimpleCanalConnector implements CanalConnector {
private InetSocketAddress doConnect() throws CanalClientException { private InetSocketAddress doConnect() throws CanalClientException {
try { try {
if (accessStrategy != null) {
address = accessStrategy.nextNode();//实时获取一下最新的running节点，lubiao
}
channel = SocketChannel.open(); channel = SocketChannel.open();
channel.socket().setSoTimeout(soTimeout); channel.socket().setSoTimeout(soTimeout);
channel.connect(address); channel.connect(address);