diff --git a/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java index e3a68934..98200c06 100644 --- a/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java +++ b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnector.java @@ -1,7 +1,6 @@ package com.alibaba.otter.canal.kafka.client; import com.alibaba.otter.canal.protocol.Message; -import com.alibaba.otter.canal.protocol.exception.CanalClientException; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.TopicPartition; @@ -73,16 +72,13 @@ public class KafkaCanalConnector { } /** - * 订阅topic，不能重复订阅 + * 订阅topic */ public void subscribe() { try { - if (kafkaConsumer != null) { - throw new CanalClientException("resubscribe error"); + if (kafkaConsumer == null) { + kafkaConsumer = new KafkaConsumer(properties); } - - kafkaConsumer = new KafkaConsumer(properties); - if (partition == null) { kafkaConsumer.subscribe(Collections.singletonList(topic)); } else { diff --git a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/AbstractKafkaTest.java b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/AbstractKafkaTest.java index 2a9ce466..df03bc5c 100644 --- a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/AbstractKafkaTest.java +++ b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/AbstractKafkaTest.java @@ -6,7 +6,7 @@ public class AbstractKafkaTest { protected String topic = "example"; protected Integer partition = null; - protected String groupId = "g1"; + protected String groupId = "g2"; protected String servers = "slave1.test.apitops.com:6667,slave2.test.apitops.com:6667,slave3.test.apitops.com:6667"; public void sleep(long time) { diff --git a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningTest.java b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningTest.java index 8b1ffc1d..4ff3a159 100644 --- a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningTest.java +++ b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningTest.java @@ -3,12 +3,17 @@ package com.alibaba.otter.canal.kafka.client.running; import com.alibaba.otter.canal.kafka.client.KafkaCanalConnector; import com.alibaba.otter.canal.kafka.client.KafkaCanalConnectors; import com.alibaba.otter.canal.protocol.Message; +import org.apache.kafka.common.errors.WakeupException; import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; public class ClientRunningTest extends AbstractKafkaTest { + private Logger logger = LoggerFactory.getLogger(ClientRunningTest.class); private boolean running = true; @@ -16,26 +21,33 @@ public class ClientRunningTest extends AbstractKafkaTest { public void testKafkaConsumer() { final ExecutorService executor = Executors.newFixedThreadPool(1); - final KafkaCanalConnector kafkaCanalConnector = KafkaCanalConnectors.newKafkaConnector(servers, topic, partition, groupId); - kafkaCanalConnector.subscribe(); + final KafkaCanalConnector connector = KafkaCanalConnectors.newKafkaConnector(servers, topic, partition, groupId); + connector.subscribe(); executor.submit(new Runnable() { @Override public void run() { while (running) { - Message message = kafkaCanalConnector.getWithoutAck(); - if (message != null) { - System.out.println(message); - sleep(40000); + try { + Message message = connector.getWithoutAck(3L, TimeUnit.SECONDS); + if (message != null) { + System.out.println(message); + } + connector.ack(); + } catch (WakeupException e) { + //ignore } - kafkaCanalConnector.ack(); } } }); - sleep(120000); + sleep(60000); + connector.disconnect(); running = false; - kafkaCanalConnector.disconnect(); + executor.shutdown(); + logger.info("shutdown completed"); + + sleep(10000); } } diff --git a/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaCanalClientExample.java b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaCanalClientExample.java new file mode 100644 index 00000000..ca304a27 --- /dev/null +++ b/kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaCanalClientExample.java @@ -0,0 +1,92 @@ +package com.alibaba.otter.canal.kafka.client.running; + +import com.alibaba.otter.canal.kafka.client.KafkaCanalConnector; +import com.alibaba.otter.canal.kafka.client.KafkaCanalConnectors; +import com.alibaba.otter.canal.protocol.Message; +import org.apache.kafka.common.errors.WakeupException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.util.Assert; + +public class KafkaCanalClientExample { + protected final static Logger logger = LoggerFactory.getLogger(KafkaCanalClientExample.class); + + private KafkaCanalConnector connector; + + private volatile boolean running = false; + + private Thread thread = null; + + private Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() { + public void uncaughtException(Thread t, Throwable e) { + logger.error("parse events has an error", e); + } + }; + + public KafkaCanalClientExample(String servers, String topic, Integer partition, String groupId) { + connector = KafkaCanalConnectors.newKafkaConnector(servers, topic, partition, groupId); + } + + public void start() { + Assert.notNull(connector, "connector is null"); + thread = new Thread(new Runnable() { + + public void run() { + process(); + } + }); + + thread.setUncaughtExceptionHandler(handler); + running = true; + } + + public void stop() { + if (!running) { + return; + } + connector.disconnect(); + running = false; + if (thread != null) { + try { + thread.join(); + } catch (InterruptedException e) { + // ignore + } + } + } + + private void process() { + while (running) { + try { + connector.subscribe(); + while (running) { + try { + Message message = connector.getWithoutAck(); //获取message + long batchId = message.getId(); + int size = message.getEntries().size(); + if (batchId == -1 || size == 0) { + // try { + // Thread.sleep(1000); + // } catch (InterruptedException e) { + // } + } else { + // printSummary(message, batchId, size); + // printEntry(message.getEntries()); + logger.info(message.toString()); + } + + connector.ack(); // 提交确认 + } catch (WakeupException e) { + //ignore + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + } catch (WakeupException e) { + //ignore + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + } +} \ No newline at end of file