From d991e6be6086019285d65e9062d97c081319ca54 Mon Sep 17 00:00:00 2001 From: rewerma Date: Wed, 13 Jun 2018 10:34:28 +0800 Subject: [PATCH] =?UTF-8?q?kafka=20=E5=AE=A2=E6=88=B7=E7=AB=AF=E6=B6=88?= =?UTF-8?q?=E8=B4=B9=E4=BB=A3=E7=A0=81=E6=95=B4=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- kafka-client/pom.xml | 3 +- .../kafka/client/KafkaCanalConnector.java | 122 ++++++++---------- .../kafka/client/KafkaCanalConnectors.java | 6 +- .../client/running/ClientRunningTest.java | 9 +- kafka/src/main/resources/kafka.yml | 2 +- 5 files changed, 65 insertions(+), 77 deletions(-) diff --git a/kafka-client/pom.xml b/kafka-client/pom.xml index 0efd13d4..264cfa0c 100644 --- a/kafka-client/pom.xml +++ b/kafka-client/pom.xml @@ -7,6 +7,7 @@ canal com.alibaba.otter 1.0.26-SNAPSHOT + ../pom.xml com.alibaba.otter canal.kafka.client @@ -17,7 +18,7 @@ com.alibaba.otter - canal.client + canal.protocol ${project.version} 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 fc749fec..e3a68934 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,6 +1,5 @@ package com.alibaba.otter.canal.kafka.client; -import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.exception.CanalClientException; import org.apache.kafka.clients.consumer.ConsumerRecords; @@ -20,7 +19,7 @@ import java.util.concurrent.TimeUnit; * @author machengyuan @ 2018-6-12 * @version 1.0.0 */ -public class KafkaCanalConnector implements CanalConnector { +public class KafkaCanalConnector { private KafkaConsumer kafkaConsumer; @@ -39,23 +38,31 @@ public class KafkaCanalConnector implements CanalConnector { properties.put("bootstrap.servers", servers); properties.put("group.id", groupId); properties.put("enable.auto.commit", false); - properties.put("auto.commit.interval.ms", 1000); + properties.put("auto.commit.interval.ms", "1000"); properties.put("auto.offset.reset", "latest"); //earliest //如果没有offset则从最后的offset开始读 - properties.put("request.timeout.ms", 600000); - properties.put("offsets.commit.timeout.ms", 300000); - properties.put("session.timeout.ms", 30000); - properties.put("max.poll.records", 1); //一次只取一条message + properties.put("request.timeout.ms", "360000"); //必须大于session.timeout.ms的设置 + properties.put("session.timeout.ms", "300000"); //默认为5分钟 + properties.put("max.poll.records", "1"); //所以一次只取一条数据 properties.put("key.deserializer", StringDeserializer.class.getName()); properties.put("value.deserializer", MessageDeserializer.class.getName()); } - @Override - public void connect() throws CanalClientException { - kafkaConsumer = new KafkaConsumer(properties); + /** + * 重新设置sessionTime + * + * @param timeout + * @param unit + */ + public void setSessionTimeout(Long timeout, TimeUnit unit) { + long t = unit.toMillis(timeout); + properties.put("request.timeout.ms", String.valueOf(t + 60000)); + properties.put("session.timeout.ms", String.valueOf(t)); } - @Override - public void disconnect() throws CanalClientException { + /** + * 释放链接 + */ + public void disconnect() { if (kafkaConsumer != null) { try { kafkaConsumer.close(); @@ -65,17 +72,17 @@ public class KafkaCanalConnector implements CanalConnector { } } - @Override - public boolean checkValid() throws CanalClientException { - return true; - } - - @Override - public void subscribe(String filter) throws CanalClientException { + /** + * 订阅topic，不能重复订阅 + */ + public void subscribe() { try { - if (kafkaConsumer == null) { - throw new CanalClientException("connect the kafka first before subscribe"); + if (kafkaConsumer != null) { + throw new CanalClientException("resubscribe error"); } + + kafkaConsumer = new KafkaConsumer(properties); + if (partition == null) { kafkaConsumer.subscribe(Collections.singletonList(topic)); } else { @@ -85,86 +92,69 @@ public class KafkaCanalConnector implements CanalConnector { } } catch (WakeupException e) { closeByWakeupException(e); - } catch (Exception e) { - throw new CanalClientException(e); } } - @Override - public void subscribe() throws CanalClientException { - subscribe(null); - } - - @Override - public void unsubscribe() throws CanalClientException { + /** + * 取消订阅 + */ + public void unsubscribe() { try { kafkaConsumer.unsubscribe(); } catch (WakeupException e) { closeByWakeupException(e); - } catch (Exception e) { - throw new CanalClientException(e); } } - @Override - public Message get(int batchSize) throws CanalClientException { - return get(batchSize, 100L, TimeUnit.MILLISECONDS); + /** + * 获取数据，自动进行确认 + * + * @return + */ + public Message get() { + return get(100L, TimeUnit.MILLISECONDS); } - @Override - public Message get(int batchSize, Long timeout, TimeUnit unit) throws CanalClientException { - Message message = getWithoutAck(batchSize, timeout, unit); - this.ack(1); + public Message get(Long timeout, TimeUnit unit) { + Message message = getWithoutAck(timeout, unit); + this.ack(); return message; } - @Override - public Message getWithoutAck(int batchSize) throws CanalClientException { - return getWithoutAck(batchSize, 100L, TimeUnit.MILLISECONDS); + public Message getWithoutAck() { + return getWithoutAck(100L, TimeUnit.MILLISECONDS); } - @Override - public Message getWithoutAck(int batchSize, Long timeout, TimeUnit unit) throws CanalClientException { + /** + * 获取数据，不进行确认，等待处理完成手工确认 + * + * @return + */ + public Message getWithoutAck(Long timeout, TimeUnit unit) { try { - ConsumerRecords records = kafkaConsumer.poll(unit.toMillis(timeout)); //基于配置，一次最多只能poll到一条Msg + ConsumerRecords records = + kafkaConsumer.poll(unit.toMillis(timeout)); //基于配置，最多只能poll到一条数据 if (!records.isEmpty()) { return records.iterator().next().value(); } } catch (WakeupException e) { closeByWakeupException(e); - } catch (Exception e) { - throw new CanalClientException(e); } return null; } - @Override - public void ack(long batchId) throws CanalClientException { + /** + * 提交offset，如果超过 session.timeout.ms 设置的时间没有ack则会抛出异常，ack失败 + */ + public void ack() { try { kafkaConsumer.commitSync(); } catch (WakeupException e) { closeByWakeupException(e); - } catch (Exception e) { - throw new CanalClientException(e); } } - @Override - public void rollback(long batchId) throws CanalClientException { - - } - - @Override - public void rollback() throws CanalClientException { - - } - - @Override - public void stopRunning() throws CanalClientException { - - } - private void closeByWakeupException(WakeupException e) { kafkaConsumer.close(); throw e; diff --git a/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnectors.java b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnectors.java index f0d4dea6..eb5f248c 100644 --- a/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnectors.java +++ b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/KafkaCanalConnectors.java @@ -1,7 +1,5 @@ package com.alibaba.otter.canal.kafka.client; -import com.alibaba.otter.canal.client.CanalConnector; - public class KafkaCanalConnectors { /** * 创建kafka客户端链接 @@ -12,7 +10,7 @@ public class KafkaCanalConnectors { * @param groupId * @return */ - public static CanalConnector newKafkaConnector(String servers, String topic, Integer partition, String groupId) { + public static KafkaCanalConnector newKafkaConnector(String servers, String topic, Integer partition, String groupId) { return new KafkaCanalConnector(servers, topic, partition, groupId); } @@ -24,7 +22,7 @@ public class KafkaCanalConnectors { * @param groupId * @return */ - public static CanalConnector newKafkaConnector(String servers, String topic, String groupId) { + public static KafkaCanalConnector newKafkaConnector(String servers, String topic, String groupId) { return new KafkaCanalConnector(servers, topic, null, groupId); } } 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 c968f0f3..8b1ffc1d 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 @@ -1,6 +1,6 @@ package com.alibaba.otter.canal.kafka.client.running; -import com.alibaba.otter.canal.client.CanalConnector; +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.junit.Test; @@ -16,20 +16,19 @@ public class ClientRunningTest extends AbstractKafkaTest { public void testKafkaConsumer() { final ExecutorService executor = Executors.newFixedThreadPool(1); - final CanalConnector kafkaCanalConnector = KafkaCanalConnectors.newKafkaConnector(servers, topic, partition, groupId); - kafkaCanalConnector.connect(); + final KafkaCanalConnector kafkaCanalConnector = KafkaCanalConnectors.newKafkaConnector(servers, topic, partition, groupId); kafkaCanalConnector.subscribe(); executor.submit(new Runnable() { @Override public void run() { while (running) { - Message message = kafkaCanalConnector.getWithoutAck(1); + Message message = kafkaCanalConnector.getWithoutAck(); if (message != null) { System.out.println(message); sleep(40000); } - kafkaCanalConnector.ack(1); + kafkaCanalConnector.ack(); } } }); diff --git a/kafka/src/main/resources/kafka.yml b/kafka/src/main/resources/kafka.yml index a869321c..3861fe58 100644 --- a/kafka/src/main/resources/kafka.yml +++ b/kafka/src/main/resources/kafka.yml @@ -5,7 +5,7 @@ lingerMs: 1 bufferMemory: 33554432 topics: - - topic: expTest + - topic: example partition: canalDestination: example # - topic: example2