kafka client keep alive间隔时间

This commit is contained in:
mcy
2018-08-28 13:51:23 +08:00
parent 6b94f396ff
commit b5e05c8c77

View File

@@ -143,9 +143,13 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
});
// 间隔一段时间ack一次, 防止因超时未响应切换到另外台客户端
long currentTS = System.currentTimeMillis();
while (executing.get()) {
connector.ack();
Thread.sleep(500);
// 大于1分钟未消费完ack一次keep alive
if (System.currentTimeMillis() - currentTS > 60000) {
connector.ack();
currentTS = System.currentTimeMillis();
}
}
} else {
connector.ack();