From ebf91397627e22cfbf07263663e40bdb67dbdc4a Mon Sep 17 00:00:00 2001 From: mcy Date: Mon, 3 Sep 2018 17:40:30 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0kafka=20producer=20sender?= =?UTF-8?q?=E7=9A=84callback=EF=BC=8Ccanal=E5=92=8Ckafka=E5=90=8C=E6=AD=A5?= =?UTF-8?q?=E6=8F=90=E4=BA=A4=E5=9B=9E=E6=BB=9A=20=E5=8E=BB=E9=99=A4?= =?UTF-8?q?=E4=B8=80=E4=B8=AAcanal=E5=AE=9E=E4=BE=8B=E5=AF=B9=E5=BA=94?= =?UTF-8?q?=E5=A4=9A=E4=B8=AAkafka=20topic=E7=9A=84=E6=9C=BA=E5=88=B6?= =?UTF-8?q?=EF=BC=8C=E4=B8=80=E4=B8=AA=E5=AE=9E=E4=BE=8B=E5=8F=AA=E5=AF=B9?= =?UTF-8?q?=E5=BA=94=E4=B8=80=E4=B8=AAtopic=20kafka=E6=A8=A1=E5=BC=8F?= =?UTF-8?q?=E4=B8=8Bserver=E7=AB=AF=E7=9A=84=E6=89=B9=E6=AC=A1=E5=A4=A7?= =?UTF-8?q?=E5=B0=8F=E4=B8=8D=E8=83=BD=E8=B6=85=E8=BF=871M=EF=BC=8C?= =?UTF-8?q?=E6=9C=80=E5=A4=A7=E4=B8=BA900K?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../loader/CanalAdapterKafkaWorker.java | 6 +-- .../adapter/loader/CanalAdapterLoader.java | 15 +++---- .../adapter/loader/CanalAdapterWorker.java | 13 +++--- deployer/src/main/resources/kafka.yml | 6 +-- .../otter/canal/kafka/CanalKafkaProducer.java | 38 +++++++++++++----- .../otter/canal/kafka/CanalKafkaStarter.java | 40 ++++++++++--------- 6 files changed, 66 insertions(+), 52 deletions(-) diff --git a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java index aecaceb2..5cf86611 100644 --- a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java +++ b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java @@ -97,9 +97,9 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker { try { logger.info("=============> Start to connect topic: {} <=============", this.topic); connector.connect(); - logger.info("=============> Start to subscribe topic: {}<=============", this.topic); + logger.info("=============> Start to subscribe topic: {} <=============", this.topic); connector.subscribe(); - logger.info("=============> Subscribe topic: {} succeed<=============", this.topic); + logger.info("=============> Subscribe topic: {} succeed <=============", this.topic); while (running) { try { // switcher.get(); //等待开关开启 @@ -146,7 +146,7 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker { long currentTS = System.currentTimeMillis(); while (executing.get()) { // 大于1分钟未消费完ack一次keep alive - if (System.currentTimeMillis() - currentTS > 60000) { + if (System.currentTimeMillis() - currentTS > 60000) { connector.ack(); currentTS = System.currentTimeMillis(); } diff --git a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java index 98973f11..14efe493 100644 --- a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java +++ b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java @@ -1,12 +1,5 @@ package com.alibaba.otter.canal.client.adapter.loader; -import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; -import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; -import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration; -import com.alibaba.otter.canal.client.adapter.support.ExtensionLoader; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - import java.net.InetSocketAddress; import java.net.SocketAddress; import java.util.ArrayList; @@ -16,6 +9,14 @@ import java.util.Map; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; +import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; +import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration; +import com.alibaba.otter.canal.client.adapter.support.ExtensionLoader; + /** * 外部适配器的加载器 * diff --git a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java index bca40041..e341c6a3 100644 --- a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java +++ b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java @@ -1,16 +1,15 @@ package com.alibaba.otter.canal.client.adapter.loader; -import com.alibaba.otter.canal.client.CanalConnector; -import com.alibaba.otter.canal.client.CanalConnectors; -import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; -import com.alibaba.otter.canal.client.adapter.loader.AbstractCanalAdapterWorker; -import com.alibaba.otter.canal.client.impl.ClusterCanalConnector; -import com.alibaba.otter.canal.protocol.Message; - import java.net.SocketAddress; import java.util.List; import java.util.concurrent.Executors; +import com.alibaba.otter.canal.client.CanalConnector; +import com.alibaba.otter.canal.client.CanalConnectors; +import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; +import com.alibaba.otter.canal.client.impl.ClusterCanalConnector; +import com.alibaba.otter.canal.protocol.Message; + /** * 原生canal-server对应的client适配器工作线程 * diff --git a/deployer/src/main/resources/kafka.yml b/deployer/src/main/resources/kafka.yml index 8abf4934..8f897e6d 100644 --- a/deployer/src/main/resources/kafka.yml +++ b/deployer/src/main/resources/kafka.yml @@ -3,15 +3,11 @@ retries: 0 batchSize: 16384 lingerMs: 1 bufferMemory: 33554432 -filterTransactionEntry: true +canalBatchSize: 50 # Canal的batch size, 默认50K, 由于kafka最大消息体限制请勿超过1M(900K以下) canalDestinations: - canalDestination: example topic: example partition: - # 一个destination可以对应多个topic - #topics: - # - topic: example - # partition: diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java index a6d084e4..ae63f415 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java @@ -1,11 +1,11 @@ package com.alibaba.otter.canal.kafka; -import java.io.IOException; import java.util.Properties; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.KafkaException; import org.apache.kafka.common.serialization.StringSerializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -48,16 +48,32 @@ public class CanalKafkaProducer { } } - public void send(KafkaProperties.Topic topic, Message message) { - ProducerRecord record; - if (topic.getPartition() != null) { - record = new ProducerRecord(topic.getTopic(), topic.getPartition(), null, message); - } else { - record = new ProducerRecord(topic.getTopic(), message); - } - producer.send(record); - if (logger.isDebugEnabled()) { - logger.debug("send message to kafka topic: {} \n {}", topic.getTopic(), message.toString()); + public void send(KafkaProperties.Topic topic, Message message, Callback callback) { + producer.initTransactions(); + try { + producer.beginTransaction(); + ProducerRecord record; + if (topic.getPartition() != null) { + record = new ProducerRecord(topic.getTopic(), topic.getPartition(), null, message); + } else { + record = new ProducerRecord(topic.getTopic(), message); + } + producer.send(record); + producer.commitTransaction(); + callback.commit(); + if (logger.isDebugEnabled()) { + logger.debug("send message to kafka topic: {} \n {}", topic.getTopic(), message.toString()); + } + } catch (KafkaException e) { + producer.abortTransaction(); + callback.rollback(); } } + + public interface Callback { + + void commit(); + + void rollback(); + } } diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java index feef8038..4ad17e25 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaStarter.java @@ -33,9 +33,9 @@ public class CanalKafkaStarter implements CanalServerStarter { private ExecutorService executorService; - private CanalKafkaProducer canalKafkaProducer; + private CanalKafkaProducer canalKafkaProducer; - private KafkaProperties kafkaProperties; + private KafkaProperties kafkaProperties; public void init() { try { @@ -55,7 +55,7 @@ public class CanalKafkaStarter implements CanalServerStarter { canalKafkaProducer.init(kafkaProperties); // set filterTransactionEntry // if (kafkaProperties.isFilterTransactionEntry()) { - // System.setProperty("canal.instance.filter.transaction.entry", "true"); + // System.setProperty("canal.instance.filter.transaction.entry", "true"); // } // 对应每个instance启动一个worker线程 List destinations = kafkaProperties.getCanalDestinations(); @@ -101,8 +101,8 @@ public class CanalKafkaStarter implements CanalServerStarter { while (!running) ; logger.info("## start the canal consumer: {}.", destination.getCanalDestination()); - CanalServerWithEmbedded server = CanalServerWithEmbedded.instance(); - ClientIdentity clientIdentity = new ClientIdentity(destination.getCanalDestination(), (short) 1001, ""); + final CanalServerWithEmbedded server = CanalServerWithEmbedded.instance(); + final ClientIdentity clientIdentity = new ClientIdentity(destination.getCanalDestination(), (short) 1001, ""); while (running) { try { if (!server.getCanalInstances().containsKey(clientIdentity.getDestination())) { @@ -118,26 +118,28 @@ public class CanalKafkaStarter implements CanalServerStarter { while (running) { Message message = server.getWithoutAck(clientIdentity, kafkaProperties.getCanalBatchSize()); // 获取指定数量的数据 - long batchId = message.getId(); + final long batchId = message.getId(); try { int size = message.isRaw() ? message.getRawEntries().size() : message.getEntries().size(); if (batchId != -1 && size != 0) { - if (!StringUtils.isEmpty(destination.getTopic())) { - Topic topic = new Topic(); - topic.setTopic(destination.getTopic()); - topic.setPartition(destination.getPartition()); - destination.getTopics().add(topic); - } - for (Topic topic : destination.getTopics()) { - canalKafkaProducer.send(topic, message); // 发送message到所有topic - } + Topic topic = new Topic(); + topic.setTopic(destination.getTopic()); + topic.setPartition(destination.getPartition()); + canalKafkaProducer.send(topic, message, new CanalKafkaProducer.Callback() { + + @Override + public void commit() { + server.ack(clientIdentity, batchId); // 提交确认 + } + + @Override + public void rollback() { + server.rollback(clientIdentity, batchId); + } + }); // 发送message到topic } - if (batchId != -1) { - server.ack(clientIdentity, batchId); // 提交确认 - } } catch (Exception e) { - server.rollback(clientIdentity); logger.error(e.getMessage(), e); } }