From e35a06743c69831e71366f7b607abba9dc3cda7f Mon Sep 17 00:00:00 2001 From: rewerma Date: Tue, 12 Jun 2018 09:46:54 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0kafka=E7=9A=84=E6=97=A5?= =?UTF-8?q?=E5=BF=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- kafka/src/main/assembly/dev.xml | 3 - kafka/src/main/assembly/release.xml | 3 - .../kafka/producer/CanalKafkaProducer.java | 31 ++----- .../kafka/producer/CanalKafkaStarter.java | 21 ++--- .../canal/kafka/producer/KafkaProperties.java | 13 --- kafka/src/main/resources/kafka.yml | 4 +- kafka/src/main/resources/logback.xml | 85 ------------------- 7 files changed, 18 insertions(+), 142 deletions(-) delete mode 100644 kafka/src/main/resources/logback.xml diff --git a/kafka/src/main/assembly/dev.xml b/kafka/src/main/assembly/dev.xml index 2c3e3fff..df9eca81 100644 --- a/kafka/src/main/assembly/dev.xml +++ b/kafka/src/main/assembly/dev.xml @@ -34,9 +34,6 @@ **/* - - logback.xml - ./src/main/resources diff --git a/kafka/src/main/assembly/release.xml b/kafka/src/main/assembly/release.xml index 0b103620..d79ecf3c 100644 --- a/kafka/src/main/assembly/release.xml +++ b/kafka/src/main/assembly/release.xml @@ -34,9 +34,6 @@ **/* - - logback.xml - ./src/main/resources diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java index ae094b27..af2bcfca 100644 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaProducer.java @@ -22,9 +22,9 @@ import java.util.Properties; public class CanalKafkaProducer { private static final Logger logger = LoggerFactory.getLogger(CanalKafkaProducer.class); - private static Producer producer; + private Producer producer; - static void init(KafkaProperties kafkaProperties) { + public void init(KafkaProperties kafkaProperties) { Properties properties = new Properties(); properties.put("bootstrap.servers", kafkaProperties.getServers()); properties.put("acks", "all"); @@ -37,7 +37,7 @@ public class CanalKafkaProducer { producer = new KafkaProducer(properties); } - static void stop() { + public void stop() { try { logger.info("## stop the kafka producer"); producer.close(); @@ -48,29 +48,8 @@ public class CanalKafkaProducer { } } - static void send(Topic topic, Message message) { - try { - List entries = message.getEntries(); - boolean flag = false; - if (!entries.isEmpty()) { - for (CanalEntry.Entry entry : entries) { - CanalEntry.RowChange rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); - if (rowChage.getIsDdl()) { - flag = true; - break; - } else { - if (!rowChage.getRowDatasList().isEmpty()) { - flag = true; - break; - } - } - } - } - if (!flag) { - return; - } - } catch (Exception e) { - logger.error(e.getMessage(), e); + public void send(Topic topic, Message message) { + if (message == null || message.getEntries().isEmpty()) { return; } diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java index e89daefc..7514c5ce 100644 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/CanalKafkaStarter.java @@ -1,14 +1,7 @@ package com.alibaba.otter.canal.kafka.producer; -import java.io.FileInputStream; -import java.util.List; -import java.util.Properties; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; - import com.alibaba.otter.canal.kafka.CanalServerStarter; import com.alibaba.otter.canal.kafka.producer.KafkaProperties.Topic; -import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.ClientIdentity; import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded; @@ -17,6 +10,11 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.yaml.snakeyaml.Yaml; +import java.io.FileInputStream; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + /** * kafka 启动类 * @@ -31,6 +29,8 @@ public class CanalKafkaStarter { private static ExecutorService executorService; + private static CanalKafkaProducer canalKafkaProducer; + public static void init() { try { @@ -46,7 +46,8 @@ public class CanalKafkaStarter { } //初始化 kafka producer - CanalKafkaProducer.init(kafkaProperties); + canalKafkaProducer = new CanalKafkaProducer(); + canalKafkaProducer.init(kafkaProperties); //对应每个instance启动一个worker线程 List topics = kafkaProperties.getTopics(); @@ -70,7 +71,7 @@ public class CanalKafkaStarter { logger.info("## stop the kafka workers"); running = false; executorService.shutdown(); - CanalKafkaProducer.stop(); + canalKafkaProducer.stop(); } catch (Throwable e) { logger.warn("##something goes wrong when stopping kafka workers:", e); } finally { @@ -118,7 +119,7 @@ public class CanalKafkaStarter { //ignore } } else { - CanalKafkaProducer.send(topic, message); + canalKafkaProducer.send(topic, message); } if (batchId != -1) { diff --git a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java index 4b81238e..f70ac2c4 100644 --- a/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java +++ b/kafka/src/main/java/com/alibaba/otter/canal/kafka/producer/KafkaProperties.java @@ -1,9 +1,7 @@ package com.alibaba.otter.canal.kafka.producer; import java.util.ArrayList; -import java.util.HashMap; import java.util.List; -import java.util.Map; /** * kafka 配置项 @@ -19,7 +17,6 @@ public class KafkaProperties { private long bufferMemory = 33554432L; private List topics = new ArrayList(); - private Map topicMap = new HashMap(); public static class Topic { private String topic; @@ -97,15 +94,5 @@ public class KafkaProperties { public void setTopics(List topics) { this.topics = topics; - - if (topics != null) { - for (Topic topic : topics) { - this.topicMap.put(topic.destination, topic); - } - } - } - - public Topic getTopicByDestination(String destination) { - return this.topicMap.get(destination); } } diff --git a/kafka/src/main/resources/kafka.yml b/kafka/src/main/resources/kafka.yml index af84977d..3a846c61 100644 --- a/kafka/src/main/resources/kafka.yml +++ b/kafka/src/main/resources/kafka.yml @@ -5,8 +5,8 @@ lingerMs: 1 bufferMemory: 33554432 topics: - - topic: example2 + - topic: example partition: - destination: example2 + destination: example diff --git a/kafka/src/main/resources/logback.xml b/kafka/src/main/resources/logback.xml deleted file mode 100644 index 158c1699..00000000 --- a/kafka/src/main/resources/logback.xml +++ /dev/null @@ -1,85 +0,0 @@ - - - - - %d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{56} - %msg%n - - - - - - - destination - canal - - - - ../logs/${destination}/${destination}.log - - - ../logs/${destination}/%d{yyyy-MM-dd}/${destination}-%d{yyyy-MM-dd}-%i.log.gz - - - 512MB - - 60 - - - - %d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %logger{56} - %msg%n - - - - - - - - - destination - canal - - - - ../logs/${destination}/meta.log - - - ../logs/${destination}/%d{yyyy-MM-dd}/meta-%d{yyyy-MM-dd}-%i.log.gz - - - 32MB - - 60 - - - - %d{yyyy-MM-dd HH:mm:ss.SSS} - %msg%n - - - - - - - - - - - - - - - - - - - - - - - - - - - - \ No newline at end of file