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

block waiting interrupted by stop request

Browse Source
This commit is contained in:
Neal Hu
2018-05-05 23:06:16 +08:00
parent f46133d116
commit eda7e4ca2b
2 changed files with 23 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
client/src/main/java/com/alibaba/otter/canal/client/impl/ClusterCanalConnector.java
View File

@@ -23,7 +23,7 @@ public class ClusterCanalConnector implements CanalConnector {
private String username; private String username;
private String password; private String password;
private int soTimeout = 10000; private int soTimeout = 10000;
private int retryTimes = 3; private int retryTimes = 3; // 设置-1时可以subscribe阻塞等待时优雅停机
private int retryInterval = 5000; // 重试的时间间隔，默认5秒 private int retryInterval = 5000; // 重试的时间间隔，默认5秒
private CanalNodeAccessStrategy accessStrategy; private CanalNodeAccessStrategy accessStrategy;
private SimpleCanalConnector currentConnector; private SimpleCanalConnector currentConnector;
@@ -106,13 +106,19 @@ public class ClusterCanalConnector implements CanalConnector {
this.filter = filter; this.filter = filter;
return; return;
} catch (Throwable t) { } catch (Throwable t) {
logger.warn(String.format( if (retryTimes == -1 && t.getCause() instanceof InterruptedException) {
"something goes wrong when subscribing from server: %s", logger.info("block waiting interrupted by other thread.");
currentConnector != null ? currentConnector.getAddress() : "null"), return;
t); } else {
times++; logger.warn(String.format(
restart(); "something goes wrong when subscribing from server: %s",
logger.info("restart the connector for next round retry."); currentConnector != null ? currentConnector.getAddress() : "null"),
t);
times++;
restart();
logger.info("restart the connector for next round retry.");
}
} }
} }

9
example/src/main/java/com/alibaba/otter/canal/example/AbstractCanalClientTest.java
View File

@@ -4,6 +4,7 @@ import java.text.SimpleDateFormat;
import java.util.Date; import java.util.Date;
import java.util.List; import java.util.List;
import com.alibaba.otter.canal.client.impl.ClusterCanalConnector;
import org.apache.commons.lang.SystemUtils; import org.apache.commons.lang.SystemUtils;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
@@ -35,6 +36,7 @@ public class AbstractCanalClientTest {
protected static final String SEP = SystemUtils.LINE_SEPARATOR; protected static final String SEP = SystemUtils.LINE_SEPARATOR;
protected static final String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss"; protected static final String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";
protected volatile boolean running = false; protected volatile boolean running = false;
private volatile boolean waiting = true;
protected Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() { protected Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() {
public void uncaughtException(Thread t, Throwable e) { public void uncaughtException(Thread t, Throwable e) {
@@ -91,6 +93,12 @@ public class AbstractCanalClientTest {
return; return;
} }
running = false; running = false;
if (waiting) {
if (connector instanceof ClusterCanalConnector) {
((ClusterCanalConnector) connector).setRetryTimes(-1);
}
thread.interrupt();
}
if (thread != null) { if (thread != null) {
try { try {
thread.join(); thread.join();
@@ -109,6 +117,7 @@ public class AbstractCanalClientTest {
MDC.put("destination", destination); MDC.put("destination", destination);
connector.connect(); connector.connect();
connector.subscribe(); connector.subscribe();
waiting = false;
while (running) { while (running) {
Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据 Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
long batchId = message.getId(); long batchId = message.getId();