From cb6c7ef3e8b468277a7b52e6caefce03e456617a Mon Sep 17 00:00:00 2001 From: rewerma Date: Wed, 13 Jun 2018 14:24:32 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=8C=E5=96=84=E6=B3=A8=E9=87=8A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../canal/kafka/client/KafkaCanalConnectors.java | 6 ++++++ .../canal/kafka/client/MessageDeserializer.java | 6 ++++++ .../kafka/client/running/AbstractKafkaTest.java | 6 ++++++ ...ntExample.java => CanalKafkaClientExample.java} | 14 ++++++++++---- ...unningTest.java => KafkaClientRunningTest.java} | 10 ++++++++-- .../canal/kafka/producer/MessageSerializer.java | 2 +- 6 files changed, 37 insertions(+), 7 deletions(-) rename kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/{KafkaCanalClientExample.java => CanalKafkaClientExample.java} (92%) rename kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/{ClientRunningTest.java => KafkaClientRunningTest.java} (85%) 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 eb5f248c..26c0c5ae 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,5 +1,11 @@ package com.alibaba.otter.canal.kafka.client; +/** + * canal kafka connectors创建工具类 + * + * @author machengyuan @ 2018-6-12 + * @version 1.0.0 + */ public class KafkaCanalConnectors { /** * 创建kafka客户端链接 diff --git a/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/MessageDeserializer.java b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/MessageDeserializer.java index c321d076..c6a7ee83 100644 --- a/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/MessageDeserializer.java +++ b/kafka-client/src/main/java/com/alibaba/otter/canal/kafka/client/MessageDeserializer.java @@ -11,6 +11,12 @@ import org.slf4j.LoggerFactory; import java.util.Map; +/** + * Kafka Message类的反序列化 + * + * @author machengyuan @ 2018-6-12 + * @version 1.0.0 + */ public class MessageDeserializer implements Deserializer { private static Logger logger = LoggerFactory.getLogger(MessageDeserializer.class); 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 c47e2d8e..f1e52be4 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 @@ -2,6 +2,12 @@ package com.alibaba.otter.canal.kafka.client.running; import org.junit.Assert; +/** + * Kafka 测试基类 + * + * @author machengyuan @ 2018-6-12 + * @version 1.0.0 + */ public abstract class AbstractKafkaTest { public static String topic = "example"; 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/CanalKafkaClientExample.java similarity index 92% rename from kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaCanalClientExample.java rename to kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/CanalKafkaClientExample.java index 4527d847..5a7fe4f0 100644 --- 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/CanalKafkaClientExample.java @@ -10,8 +10,14 @@ import org.springframework.util.Assert; import java.util.concurrent.TimeUnit; -public class KafkaCanalClientExample { - protected final static Logger logger = LoggerFactory.getLogger(KafkaCanalClientExample.class); +/** + * Kafka client example + * + * @author machengyuan @ 2018-6-12 + * @version 1.0.0 + */ +public class CanalKafkaClientExample { + protected final static Logger logger = LoggerFactory.getLogger(CanalKafkaClientExample.class); private KafkaCanalConnector connector; @@ -25,13 +31,13 @@ public class KafkaCanalClientExample { } }; - public KafkaCanalClientExample(String servers, String topic, Integer partition, String groupId) { + public CanalKafkaClientExample(String servers, String topic, Integer partition, String groupId) { connector = KafkaCanalConnectors.newKafkaConnector(servers, topic, partition, groupId); } public static void main(String[] args) { try { - final KafkaCanalClientExample kafkaCanalClientExample = new KafkaCanalClientExample(AbstractKafkaTest.servers, + final CanalKafkaClientExample kafkaCanalClientExample = new CanalKafkaClientExample(AbstractKafkaTest.servers, AbstractKafkaTest.topic, AbstractKafkaTest.partition, AbstractKafkaTest.groupId); logger.info("## start the kafka consumer: {}-{}", AbstractKafkaTest.topic, AbstractKafkaTest.groupId); kafkaCanalClientExample.start(); 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/KafkaClientRunningTest.java similarity index 85% rename from kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/ClientRunningTest.java rename to kafka-client/src/test/java/com/alibaba/otter/canal/kafka/client/running/KafkaClientRunningTest.java index 4ff3a159..3260c657 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/KafkaClientRunningTest.java @@ -12,8 +12,14 @@ 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); +/** + * Kafka consumer获取Message的测试例子 + * + * @author machengyuan @ 2018-6-12 + * @version 1.0.0 + */ +public class KafkaClientRunningTest extends AbstractKafkaTest { + private Logger logger = LoggerFactory.getLogger(KafkaClientRunningTest.class); private boolean running = true; diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/MessageSerializer.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/MessageSerializer.java index d1a9577a..c74eb9df 100644 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/MessageSerializer.java +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/MessageSerializer.java @@ -10,7 +10,7 @@ import org.springframework.util.CollectionUtils; import java.util.Map; /** - * Message类的序列化Kafka实现 + * Kafka Message类的序列化 * * @author machengyuan 2018-6-11 下午05:30:49 * @version 1.0.0