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

fixed issue #4864 , connector.disconnect optimize

Browse Source
This commit is contained in:
jianghang.loujh
2023-09-04 19:34:06 +08:00
parent d623bea71a
commit 34e9440383
1 changed files with 11 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
connector/tcp-connector/src/main/java/com/alibaba/otter/canal/connector/tcp/consumer/CanalTCPConsumer.java
View File

@@ -1,5 +1,13 @@
package com.alibaba.otter.canal.connector.tcp.consumer; package com.alibaba.otter.canal.connector.tcp.consumer;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang.StringUtils;
import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.impl.ClusterCanalConnector; import com.alibaba.otter.canal.client.impl.ClusterCanalConnector;
import com.alibaba.otter.canal.client.impl.ClusterNodeAccessStrategy; import com.alibaba.otter.canal.client.impl.ClusterNodeAccessStrategy;
@@ -12,14 +20,6 @@ import com.alibaba.otter.canal.connector.core.util.MessageUtil;
import com.alibaba.otter.canal.connector.tcp.config.TCPConstants; import com.alibaba.otter.canal.connector.tcp.config.TCPConstants;
import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.Message;
import org.apache.commons.lang.StringUtils;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
/** /**
* TCP 消费者连接器, 一个destination对应一个SPI实例 * TCP 消费者连接器, 一个destination对应一个SPI实例
* *
@@ -99,7 +99,9 @@ public class CanalTCPConsumer implements CanalMsgConsumer {
@Override @Override
public void disconnect() { public void disconnect() {
canalConnector.unsubscribe(); // tcp模式下，因为是单tcp消费，避免adapter异常断开时直接unsubscribe
// unsubscribe发送给canal-server会导致清理cursor位点,如果此时canal-server出现重启,就会丢失binlog数据
// canalConnector.unsubscribe();
canalConnector.disconnect(); canalConnector.disconnect();
} }
} }