From e8e00324044f54a385345f99229c3c8cd38b5bd5 Mon Sep 17 00:00:00 2001 From: mcy Date: Fri, 14 Sep 2018 11:22:47 +0800 Subject: [PATCH 1/3] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E6=89=81=E5=B9=B3Message?= =?UTF-8?q?=E7=94=A8=E4=BA=8E=E5=8D=A1=E5=A4=AB=E5=8D=A1=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E6=8A=95=E9=80=92?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../client/adapter/CanalOuterAdapter.java | 8 +- .../adapter/support/CanalClientConfig.java | 10 + .../client/adapter/support/JdbcTypeUtil.java | 4 +- .../client/adapter/support/MessageUtil.java | 58 +++- .../client/adapter/hbase/HbaseAdapter.java | 23 +- .../adapter/logger/LoggerAdapterExample.java | 15 +- .../loader/AbstractCanalAdapterWorker.java | 60 +++- .../loader/CanalAdapterKafkaWorker.java | 99 +++--- .../adapter/loader/CanalAdapterLoader.java | 7 +- .../src/main/resources/canal-client.yml | 31 +- .../client/kafka/KafkaCanalConnector.java | 162 +++++----- .../client/kafka/KafkaCanalConnectors.java | 27 +- .../kafka/CanalKafkaClientExample.java | 63 ++-- .../running/kafka/KafkaClientRunningTest.java | 9 +- deployer/src/main/resources/kafka.yml | 5 +- .../otter/canal/protocol/FlatMessage.java | 286 ++++++++++++++++++ .../otter/canal/kafka/CanalKafkaProducer.java | 50 ++- .../otter/canal/kafka/KafkaProperties.java | 9 + .../embedded/CanalServerWithEmbedded.java | 7 +- 19 files changed, 686 insertions(+), 247 deletions(-) create mode 100644 protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/CanalOuterAdapter.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/CanalOuterAdapter.java index 53368472..0871eb74 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/CanalOuterAdapter.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/CanalOuterAdapter.java @@ -1,7 +1,9 @@ package com.alibaba.otter.canal.client.adapter; import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration; +import com.alibaba.otter.canal.client.adapter.support.Dml; import com.alibaba.otter.canal.client.adapter.support.SPI; +import com.alibaba.otter.canal.protocol.FlatMessage; import com.alibaba.otter.canal.protocol.Message; /** @@ -23,9 +25,11 @@ public interface CanalOuterAdapter { /** * 往适配器中写入数据 * - * @param message message数据包 + * @param dml 数据包 */ - void writeOut(Message message); + void writeOut(Dml dml); + + // void writeOut(FlatMessage flatMessage); /** * 外部适配器销毁接口 diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java index f6fa7707..25b339fe 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java @@ -20,6 +20,8 @@ public class CanalClientConfig { private String bootstrapServers; + private Boolean flatMessage = true; + private List kafkaTopics; private List canalInstances; @@ -56,6 +58,14 @@ public class CanalClientConfig { this.bootstrapServers = bootstrapServers; } + public Boolean getFlatMessage() { + return flatMessage; + } + + public void setFlatMessage(Boolean flatMessage) { + this.flatMessage = flatMessage; + } + public List getKafkaTopics() { return kafkaTopics; } diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/JdbcTypeUtil.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/JdbcTypeUtil.java index fd65589f..777770ef 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/JdbcTypeUtil.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/JdbcTypeUtil.java @@ -21,7 +21,7 @@ public class JdbcTypeUtil { private static Logger logger = LoggerFactory.getLogger(JdbcTypeUtil.class); - public static Object typeConvert(String tableName, String columnName, String value, int sqlType, String mysqlType) { + public static Object typeConvert(String columnName, String value, int sqlType, String mysqlType) { if (value == null || value.equals("")) { return null; } @@ -96,7 +96,7 @@ public class JdbcTypeUtil { } return res; } catch (Exception e) { - logger.error("table: {} column: {}, failed convert type {} to {}", tableName, columnName, value, sqlType); + logger.error("table: {} column: {}, failed convert type {} to {}", columnName, value, sqlType); return value; } } diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java index 7f3722f1..4765ec30 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java @@ -3,6 +3,7 @@ package com.alibaba.otter.canal.client.adapter.support; import java.util.*; import com.alibaba.otter.canal.protocol.CanalEntry; +import com.alibaba.otter.canal.protocol.FlatMessage; import com.alibaba.otter.canal.protocol.Message; /** @@ -63,8 +64,7 @@ public class MessageUtil { for (CanalEntry.Column column : columns) { row.put(column.getName(), - JdbcTypeUtil.typeConvert(dml.getTable(), - column.getName(), + JdbcTypeUtil.typeConvert(column.getName(), column.getValue(), column.getSqlType(), column.getMysqlType())); @@ -82,8 +82,7 @@ public class MessageUtil { for (CanalEntry.Column column : rowData.getBeforeColumnsList()) { if (updateSet.contains(column.getName())) { rowOld.put(column.getName(), - JdbcTypeUtil.typeConvert(dml.getTable(), - column.getName(), + JdbcTypeUtil.typeConvert(column.getName(), column.getValue(), column.getSqlType(), column.getMysqlType())); @@ -106,6 +105,57 @@ public class MessageUtil { } } + public static Dml flatMessage2Dml(FlatMessage flatMessage) { + if (flatMessage == null) { + return null; + } + Dml dml = new Dml(); + dml.setDatabase(flatMessage.getDatabase()); + dml.setTable(flatMessage.getTable()); + dml.setType(flatMessage.getType()); + dml.setTs(flatMessage.getTs()); + dml.setSql(flatMessage.getSql()); + if (flatMessage.getSqlType() == null || flatMessage.getMysqlType() == null) { + throw new RuntimeException("SqlType or mysqlType is null"); + } + List> data = flatMessage.getData(); + if (data != null) { + dml.setData(changeRows(data, flatMessage.getSqlType(), flatMessage.getMysqlType())); + } + List> old = flatMessage.getOld(); + if (old != null) { + dml.setOld(changeRows(old, flatMessage.getSqlType(), flatMessage.getMysqlType())); + } + return dml; + } + + private static List> changeRows(List> rows, Map sqlTypes, + Map mysqlTypes) { + List> result = new ArrayList<>(); + for (Map row : rows) { + Map resultRow = new LinkedHashMap<>(); + for (Map.Entry entry : row.entrySet()) { + String columnName = entry.getKey(); + String columnValue = entry.getValue(); + + Integer sqlType = sqlTypes.get(columnName); + if (sqlType == null) { + continue; + } + + String mysqlType = mysqlTypes.get(columnName); + if (mysqlType == null) { + continue; + } + + Object finalValue = JdbcTypeUtil.typeConvert(columnName, columnValue, sqlType, mysqlType); + resultRow.put(columnName, finalValue); + } + result.add(resultRow); + } + return result; + } + public interface Consumer { void accept(T t); diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java index 03103b8b..bbd8fb1c 100644 --- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java +++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java @@ -4,6 +4,7 @@ import java.io.IOException; import java.util.HashMap; import java.util.Map; +import com.alibaba.otter.canal.protocol.FlatMessage; import org.apache.commons.lang.StringUtils; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseConfiguration; @@ -86,20 +87,14 @@ public class HbaseAdapter implements CanalOuterAdapter { } @Override - public void writeOut(Message message) { - MessageUtil.parse4Dml(message, new MessageUtil.Consumer() { - - @Override - public void accept(Dml dml) { - if (dml == null) { - return; - } - String database = dml.getDatabase(); - String table = dml.getTable(); - MappingConfig config = mappingConfigCache.get(database + "-" + table); - hbaseSyncService.sync(config, dml); - } - }); + public void writeOut(Dml dml) { + if (dml == null) { + return; + } + String database = dml.getDatabase(); + String table = dml.getTable(); + MappingConfig config = mappingConfigCache.get(database + "-" + table); + hbaseSyncService.sync(config, dml); } @Override diff --git a/client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java b/client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java index c6c1cb44..c72a0ee0 100644 --- a/client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java +++ b/client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java @@ -6,9 +6,7 @@ import org.slf4j.LoggerFactory; import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration; import com.alibaba.otter.canal.client.adapter.support.Dml; -import com.alibaba.otter.canal.client.adapter.support.MessageUtil; import com.alibaba.otter.canal.client.adapter.support.SPI; -import com.alibaba.otter.canal.protocol.Message; /** * 外部适配器示例 @@ -22,20 +20,13 @@ public class LoggerAdapterExample implements CanalOuterAdapter { private Logger logger = LoggerFactory.getLogger(this.getClass()); @Override - public void writeOut(Message message) { - // 直接输出日志信息 - MessageUtil.parse4Dml(message, new MessageUtil.Consumer() { + public void init(CanalOuterAdapterConfiguration configuration) { - @Override - public void accept(Dml dml) { - logger.info(dml.toString()); - } - }); } @Override - public void init(CanalOuterAdapterConfiguration configuration) { - + public void writeOut(Dml dml) { + logger.info(dml.toString()); } @Override diff --git a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/AbstractCanalAdapterWorker.java b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/AbstractCanalAdapterWorker.java index 9dd328e3..46e5922f 100644 --- a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/AbstractCanalAdapterWorker.java +++ b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/AbstractCanalAdapterWorker.java @@ -7,10 +7,13 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; -import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; +import com.alibaba.otter.canal.client.adapter.support.Dml; +import com.alibaba.otter.canal.client.adapter.support.MessageUtil; +import com.alibaba.otter.canal.protocol.FlatMessage; import com.alibaba.otter.canal.protocol.Message; /** @@ -47,9 +50,16 @@ public abstract class AbstractCanalAdapterWorker { public Boolean call() { try { // 组内适配器穿行运行，尽量不要配置组内适配器 - for (CanalOuterAdapter c : adapters) { + for (final CanalOuterAdapter c : adapters) { long begin = System.currentTimeMillis(); - c.writeOut(message); + MessageUtil.parse4Dml(message, new MessageUtil.Consumer() { + + @Override + public void accept(Dml dml) { + c.writeOut(dml); + } + }); + if (logger.isDebugEnabled()) { logger.debug("{} elapsed time: {}", c.getClass().getName(), @@ -71,7 +81,49 @@ public abstract class AbstractCanalAdapterWorker { logger.error("Outer adapter write failed"); } } catch (InterruptedException | ExecutionException e) { - logger.error(e.getMessage(), e); + // ignore + } + } + } + } + + protected void writeOut(final FlatMessage flatMessage) { + List> futures = new ArrayList<>(); + // 组间适配器并行运行 + for (List outerAdapters : canalOuterAdapters) { + final List adapters = outerAdapters; + futures.add(groupInnerExecutorService.submit(new Callable() { + + @Override + public Boolean call() { + try { + // 组内适配器穿行运行，尽量不要配置组内适配器 + for (CanalOuterAdapter c : adapters) { + long begin = System.currentTimeMillis(); + Dml dml = MessageUtil.flatMessage2Dml(flatMessage); + c.writeOut(dml); + if (logger.isDebugEnabled()) { + logger.debug("{} elapsed time: {}", + c.getClass().getName(), + (System.currentTimeMillis() - begin)); + } + } + return true; + } catch (Exception e) { + return false; + } + } + })); + + // 等待所有适配器写入完成 + // 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成 + for (Future f : futures) { + try { + if (!f.get()) { + logger.error("Outer adapter write failed"); + } + } catch (InterruptedException | ExecutionException e) { + // ignore } } } 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 4a7bd7a3..d7324cda 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 @@ -12,6 +12,7 @@ import org.apache.kafka.common.errors.WakeupException; import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; import com.alibaba.otter.canal.client.kafka.KafkaCanalConnector; import com.alibaba.otter.canal.client.kafka.KafkaCanalConnectors; +import com.alibaba.otter.canal.protocol.FlatMessage; import com.alibaba.otter.canal.protocol.Message; /** @@ -26,14 +27,17 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker { private String topic; - public CanalAdapterKafkaWorker(String zkServers, String bootstrapServers, String topic, String groupId, - List> canalOuterAdapters){ + private boolean flatMessage; + + public CanalAdapterKafkaWorker(String bootstrapServers, String topic, String groupId, + List> canalOuterAdapters, boolean flatMessage){ this.canalOuterAdapters = canalOuterAdapters; this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); this.topic = topic; this.canalDestination = topic; - connector = KafkaCanalConnectors.newKafkaConnector(zkServers, bootstrapServers, topic, null, groupId); - connector.setSessionTimeout(5L, TimeUnit.MINUTES); + this.flatMessage = flatMessage; + connector = KafkaCanalConnectors.newKafkaConnector(bootstrapServers, topic, null, groupId, flatMessage); + // connector.setSessionTimeout(1L, TimeUnit.MINUTES); // super.initSwitcher(topic); } @@ -104,56 +108,47 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker { try { // switcher.get(); //等待开关开启 - final Message message = connector.getWithoutAck(); + List messages; + if (!flatMessage) { + messages = connector.getWithoutAck(); + } else { + messages = connector.getFlatMessageWithoutAck(100L, TimeUnit.MILLISECONDS); + } + if (messages != null) { + for (final Object message : messages) { + executing.set(true); + if (message != null) { + executor.submit(new Runnable() { - executing.set(true); - if (message != null) { - executor.submit(new Runnable() { + @Override + public void run() { + try { + if (message instanceof FlatMessage) { + writeOut((FlatMessage) message); + } else { + writeOut((Message) message); + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } finally { + executing.compareAndSet(true, false); + } + } + }); - @Override - public void run() { - try { - if (logger.isDebugEnabled()) { - logger.debug("topic: {} batchId: {} batchSize: {} ", - topic, - message.getId(), - message.getEntries().size()); + // 间隔一段时间ack一次, 防止因超时未响应切换到另外台客户端 + long currentTS = System.currentTimeMillis(); + while (executing.get()) { + // 大于10秒未消费完ack一次keep alive + if (System.currentTimeMillis() - currentTS > 10000) { + connector.ack(); + currentTS = System.currentTimeMillis(); } - long begin = System.currentTimeMillis(); - writeOut(message); - long now = System.currentTimeMillis(); - if ((System.currentTimeMillis() - begin) > 5 * 60 * 1000) { - logger.error("topic: {} batchId {} elapsed time: {} ms", - topic, - message.getId(), - now - begin); - } - if (logger.isDebugEnabled()) { - logger.debug("topic: {} batchId {} elapsed time: {} ms", - topic, - message.getId(), - now - begin); - } - } catch (Exception e) { - logger.error(e.getMessage(), e); - } finally { - executing.compareAndSet(true, false); } } - }); - - // 间隔一段时间ack一次, 防止因超时未响应切换到另外台客户端 - long currentTS = System.currentTimeMillis(); - while (executing.get()) { - // 大于1分钟未消费完ack一次keep alive - if (System.currentTimeMillis() - currentTS > 60000) { - connector.ack(); - currentTS = System.currentTimeMillis(); - } } - } else { - connector.ack(); } + connector.ack(); } catch (CommitFailedException e) { logger.warn(e.getMessage()); } catch (Exception e) { @@ -161,19 +156,23 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker { TimeUnit.SECONDS.sleep(1L); } } - } catch (Exception e) { + } catch ( + + Exception e) { logger.error(e.getMessage(), e); } } executor.shutdown(); - try { + try + + { connector.unsubscribe(); } catch (WakeupException e) { // No-op. Continue process } - connector.disconnnect(); + connector.disconnect(); logger.info("=============> Disconnect topic: {} <=============", this.topic); } } 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 14efe493..15945b6c 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 @@ -59,6 +59,8 @@ public class CanalAdapterLoader { } String zkHosts = this.canalClientConfig.getZookeeperHosts(); + boolean flatMessage = this.canalClientConfig.getFlatMessage(); + // if (zkHosts == null && sa == null) { // throw new RuntimeException("Blank config property: canalServerHost or // zookeeperHosts"); @@ -104,11 +106,12 @@ public class CanalAdapterLoader { canalOuterAdapterGroups.add(canalOuterAdapters); // String zkServers = canalClientConfig.getZookeeperHosts(); - CanalAdapterKafkaWorker canalKafkaWorker = new CanalAdapterKafkaWorker(zkHosts, + CanalAdapterKafkaWorker canalKafkaWorker = new CanalAdapterKafkaWorker( canalClientConfig.getBootstrapServers(), kafkaTopic.getTopic(), group.getGroupId(), - canalOuterAdapterGroups); + canalOuterAdapterGroups, + flatMessage); canalKafkaWorkers.put(kafkaTopic.getTopic() + "-" + group.getGroupId(), canalKafkaWorker); canalKafkaWorker.start(); logger.info("Start adapter for canal-client kafka topic: {} succeed", diff --git a/client-launcher/src/main/resources/canal-client.yml b/client-launcher/src/main/resources/canal-client.yml index 12e016a7..6c8874c3 100644 --- a/client-launcher/src/main/resources/canal-client.yml +++ b/client-launcher/src/main/resources/canal-client.yml @@ -1,21 +1,22 @@ -canalServerHost: 127.0.0.1:11111 +#canalServerHost: 127.0.0.1:11111 #zookeeperHosts: slave1:2181 -#bootstrapServers: slave1:6667,slave2:6667 +bootstrapServers: slave1:6667,slave2:6667 +flatMessage: true -canalInstances: -- instance: example - adapterGroups: - - outAdapters: - - name: logger - - name: hbase - hosts: slave1:2181 - properties: {znodeParent: "/hbase-unsecure"} -#kafkaTopics: -#- topic: example -# groups: -# - groupId: example_g1 -# outAdapters: +#canalInstances: +#- instance: example +# adapterGroups: +# - outAdapters: # - name: logger # - name: hbase # hosts: slave1:2181 # properties: {znodeParent: "/hbase-unsecure"} +kafkaTopics: +- topic: example + groups: + - groupId: egroup + outAdapters: + - name: logger +# - name: hbase +# hosts: slave1:2181 +# properties: {znodeParent: "/hbase-unsecure"} diff --git a/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java b/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java index 7afb9c09..ad94ba98 100644 --- a/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java +++ b/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java @@ -1,19 +1,20 @@ package com.alibaba.otter.canal.client.kafka; +import java.util.ArrayList; import java.util.Collections; +import java.util.List; import java.util.Properties; import java.util.concurrent.TimeUnit; +import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.serialization.StringDeserializer; -import com.alibaba.otter.canal.client.kafka.running.ClientRunningData; -import com.alibaba.otter.canal.common.utils.AddressUtils; -import com.alibaba.otter.canal.common.zookeeper.ZkClientx; +import com.alibaba.fastjson.JSON; +import com.alibaba.otter.canal.protocol.FlatMessage; import com.alibaba.otter.canal.protocol.Message; -import com.alibaba.otter.canal.protocol.exception.CanalClientException; /** * canal kafka 数据操作客户端 @@ -24,18 +25,19 @@ import com.alibaba.otter.canal.protocol.exception.CanalClientException; public class KafkaCanalConnector { private KafkaConsumer kafkaConsumer; + private KafkaConsumer kafkaConsumer2; // 用于扁平message的数据消费 private String topic; private Integer partition; private Properties properties; - // private ClientRunningMonitor runningMonitor; // 运行控制 - // private BooleanMutex mutex = new BooleanMutex(false); - private ZkClientx zkClientx; private volatile boolean connected = false; private volatile boolean running = false; + private boolean flatMessage; - public KafkaCanalConnector(String zkServers, String servers, String topic, Integer partition, String groupId){ + public KafkaCanalConnector(String servers, String topic, Integer partition, String groupId, + boolean flatMessage){ this.topic = topic; this.partition = partition; + this.flatMessage = flatMessage; properties = new Properties(); properties.put("bootstrap.servers", servers); @@ -45,32 +47,13 @@ public class KafkaCanalConnector { properties.put("auto.offset.reset", "latest"); // 如果没有offset则从最后的offset开始读 properties.put("request.timeout.ms", "40000"); // 必须大于session.timeout.ms的设置 properties.put("session.timeout.ms", "30000"); // 默认为30秒 - properties.put("max.poll.records", "1"); // 一次只取一条message数据 + properties.put("max.poll.records", "100"); properties.put("key.deserializer", StringDeserializer.class.getName()); - properties.put("value.deserializer", MessageDeserializer.class.getName()); - - if (zkServers != null) { - zkClientx = new ZkClientx(zkServers); - - ClientRunningData clientData = new ClientRunningData(); - clientData.setGroupId(groupId); - clientData.setAddress(AddressUtils.getHostIp()); - - // runningMonitor = new ClientRunningMonitor(); - // runningMonitor.setTopic(topic); - // runningMonitor.setZkClient(zkClientx); - // runningMonitor.setClientData(clientData); - // runningMonitor.setListener(new ClientRunningListener() { - // public void processActiveEnter() { - // mutex.set(true); - // } - // - // public void processActiveExit() { - // mutex.set(false); - // } - // }); + if (!flatMessage) { + properties.put("value.deserializer", MessageDeserializer.class.getName()); + } else { + properties.put("value.deserializer", StringDeserializer.class.getName()); } - } /** @@ -93,57 +76,36 @@ public class KafkaCanalConnector { return; } - // if (runningMonitor != null) { - // if (!runningMonitor.isStart()) { - // runningMonitor.start(); - // } - // } - connected = true; - if (kafkaConsumer == null) { + if (kafkaConsumer == null && !flatMessage) { kafkaConsumer = new KafkaConsumer(properties); } + if (kafkaConsumer2 == null && flatMessage) { + kafkaConsumer2 = new KafkaConsumer(properties); + } } /** * 关闭链接 */ - public void disconnnect() { - kafkaConsumer.close(); + public void disconnect() { + if (kafkaConsumer != null) { + kafkaConsumer.close(); + } + if (kafkaConsumer2 != null) { + kafkaConsumer2.close(); + } connected = false; - // if (runningMonitor.isStart()) { - // runningMonitor.stop(); - // } } private void waitClientRunning() { - try { - if (zkClientx != null) { - if (!connected) {// 未调用connect - throw new CanalClientException("should connect first"); - } - - running = true; - // mutex.get();// 阻塞等待 - } else { - // 单机模式直接设置为running - running = true; - } - } catch (Exception e) { - Thread.currentThread().interrupt(); - throw new CanalClientException(e); - } + running = true; } public boolean checkValid() { - if (zkClientx != null) { - // return mutex.state(); - return true; - } else { - return true;// 默认都放过 - } + return true;// 默认都放过 } /** @@ -156,10 +118,20 @@ public class KafkaCanalConnector { } if (partition == null) { - kafkaConsumer.subscribe(Collections.singletonList(topic)); + if (kafkaConsumer != null) { + kafkaConsumer.subscribe(Collections.singletonList(topic)); + } + if (kafkaConsumer2 != null) { + kafkaConsumer2.subscribe(Collections.singletonList(topic)); + } } else { TopicPartition topicPartition = new TopicPartition(topic, partition); - kafkaConsumer.assign(Collections.singletonList(topicPartition)); + if (kafkaConsumer != null) { + kafkaConsumer.assign(Collections.singletonList(topicPartition)); + } + if (kafkaConsumer2 != null) { + kafkaConsumer2.assign(Collections.singletonList(topicPartition)); + } } } @@ -172,7 +144,12 @@ public class KafkaCanalConnector { return; } - kafkaConsumer.unsubscribe(); + if (kafkaConsumer != null) { + kafkaConsumer.unsubscribe(); + } + if (kafkaConsumer2 != null) { + kafkaConsumer2.unsubscribe(); + } } /** @@ -180,22 +157,22 @@ public class KafkaCanalConnector { * * @return */ - public Message get() { + public List get() { return get(100L, TimeUnit.MILLISECONDS); } - public Message get(Long timeout, TimeUnit unit) { + public List get(Long timeout, TimeUnit unit) { waitClientRunning(); if (!running) { return null; } - Message message = getWithoutAck(timeout, unit); + List messages = getWithoutAck(timeout, unit); this.ack(); - return message; + return messages; } - public Message getWithoutAck() { + public List getWithoutAck() { return getWithoutAck(100L, TimeUnit.MILLISECONDS); } @@ -204,7 +181,7 @@ public class KafkaCanalConnector { * * @return */ - public Message getWithoutAck(Long timeout, TimeUnit unit) { + public List getWithoutAck(Long timeout, TimeUnit unit) { waitClientRunning(); if (!running) { return null; @@ -213,7 +190,33 @@ public class KafkaCanalConnector { ConsumerRecords records = kafkaConsumer.poll(unit.toMillis(timeout)); // 基于配置，最多只能poll到一条数据 if (!records.isEmpty()) { - return records.iterator().next().value(); + // return records.iterator().next().value(); + List messages = new ArrayList<>(); + for (ConsumerRecord record : records) { + messages.add(record.value()); + } + return messages; + } + return null; + } + + public List getFlatMessageWithoutAck(Long timeout, TimeUnit unit) { + waitClientRunning(); + if (!running) { + return null; + } + + ConsumerRecords records = kafkaConsumer2.poll(unit.toMillis(timeout)); + + if (!records.isEmpty()) { + List flatMessages = new ArrayList<>(); + for (ConsumerRecord record : records) { + String flatMessageJson = record.value(); + FlatMessage flatMessage = JSON.parseObject(flatMessageJson, FlatMessage.class); + flatMessages.add(flatMessage); + } + + return flatMessages; } return null; } @@ -227,7 +230,12 @@ public class KafkaCanalConnector { return; } - kafkaConsumer.commitSync(); + if (kafkaConsumer != null) { + kafkaConsumer.commitSync(); + } + if (kafkaConsumer2 != null) { + kafkaConsumer2.commitAsync(); + } } public void stopRunning() { diff --git a/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnectors.java b/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnectors.java index 0634a747..436fde8f 100644 --- a/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnectors.java +++ b/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnectors.java @@ -17,8 +17,9 @@ public class KafkaCanalConnectors { * @param groupId * @return */ - public static KafkaCanalConnector newKafkaConnector(String servers, String topic, Integer partition, String groupId) { - return new KafkaCanalConnector(null, servers, topic, partition, groupId); + public static KafkaCanalConnector newKafkaConnector(String servers, String topic, Integer partition, + String groupId) { + return new KafkaCanalConnector(servers, topic, partition, groupId, false); } /** @@ -30,33 +31,21 @@ public class KafkaCanalConnectors { * @return */ public static KafkaCanalConnector newKafkaConnector(String servers, String topic, String groupId) { - return new KafkaCanalConnector(null, servers, topic, null, groupId); + return new KafkaCanalConnector(servers, topic, null, groupId, false); } /** * 创建kafka客户端链接 * - * @param zkServers * @param servers * @param topic * @param partition * @param groupId + * @param flatMessage * @return */ - public static KafkaCanalConnector newKafkaConnector(String zkServers, String servers, String topic, Integer partition, String groupId) { - return new KafkaCanalConnector(zkServers, servers, topic, partition, groupId); - } - - /** - * 创建kafka客户端链接 - * - * @param zkServers - * @param servers - * @param topic - * @param groupId - * @return - */ - public static KafkaCanalConnector newKafkaConnector(String zkServers, String servers, String topic, String groupId) { - return new KafkaCanalConnector(zkServers, servers, topic, null, groupId); + public static KafkaCanalConnector newKafkaConnector( String servers, String topic, + Integer partition, String groupId,boolean flatMessage) { + return new KafkaCanalConnector(servers, topic, partition, groupId, flatMessage); } } diff --git a/client/src/test/java/com/alibaba/otter/canal/client/running/kafka/CanalKafkaClientExample.java b/client/src/test/java/com/alibaba/otter/canal/client/running/kafka/CanalKafkaClientExample.java index 0a53a1df..93f443cd 100644 --- a/client/src/test/java/com/alibaba/otter/canal/client/running/kafka/CanalKafkaClientExample.java +++ b/client/src/test/java/com/alibaba/otter/canal/client/running/kafka/CanalKafkaClientExample.java @@ -1,5 +1,6 @@ package com.alibaba.otter.canal.client.running.kafka; +import java.util.List; import java.util.concurrent.TimeUnit; import org.apache.kafka.common.errors.WakeupException; @@ -19,32 +20,33 @@ import com.alibaba.otter.canal.protocol.Message; */ public class CanalKafkaClientExample { - protected final static Logger logger = LoggerFactory.getLogger(CanalKafkaClientExample.class); + protected final static Logger logger = LoggerFactory.getLogger(CanalKafkaClientExample.class); - private KafkaCanalConnector connector; + private KafkaCanalConnector connector; - private static volatile boolean running = false; + private static volatile boolean running = false; - private Thread thread = null; + 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 void uncaughtException(Thread t, Throwable e) { + logger.error("parse events has an error", e); + } + }; - public CanalKafkaClientExample(String zkServers, String servers, String topic, Integer partition, String groupId) { - connector = KafkaCanalConnectors.newKafkaConnector(zkServers, servers, topic, partition, groupId); + public CanalKafkaClientExample(String zkServers, String servers, String topic, Integer partition, String groupId){ + connector = KafkaCanalConnectors.newKafkaConnector(servers, topic, partition, groupId, false); } public static void main(String[] args) { try { - final CanalKafkaClientExample kafkaCanalClientExample = new CanalKafkaClientExample(AbstractKafkaTest.zkServers, - AbstractKafkaTest.servers, - AbstractKafkaTest.topic, - AbstractKafkaTest.partition, - AbstractKafkaTest.groupId); + final CanalKafkaClientExample kafkaCanalClientExample = new CanalKafkaClientExample( + AbstractKafkaTest.zkServers, + AbstractKafkaTest.servers, + AbstractKafkaTest.topic, + AbstractKafkaTest.partition, + AbstractKafkaTest.groupId); logger.info("## start the kafka consumer: {}-{}", AbstractKafkaTest.topic, AbstractKafkaTest.groupId); kafkaCanalClientExample.start(); logger.info("## the canal kafka consumer is running now ......"); @@ -107,21 +109,24 @@ public class CanalKafkaClientExample { connector.subscribe(); while (running) { try { - Message message = connector.getWithoutAck(1L, TimeUnit.SECONDS); // 获取message - if (message == null) { + List messages = connector.getWithoutAck(1L, TimeUnit.SECONDS); // 获取message + + if (messages == null) { continue; } - 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()); + for (Message message : messages) { + 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(); // 提交确认 @@ -139,6 +144,6 @@ public class CanalKafkaClientExample { } catch (WakeupException e) { // No-op. Continue process } - connector.disconnnect(); + connector.disconnect(); } } diff --git a/client/src/test/java/com/alibaba/otter/canal/client/running/kafka/KafkaClientRunningTest.java b/client/src/test/java/com/alibaba/otter/canal/client/running/kafka/KafkaClientRunningTest.java index ebc4d549..fcdc0bf6 100644 --- a/client/src/test/java/com/alibaba/otter/canal/client/running/kafka/KafkaClientRunningTest.java +++ b/client/src/test/java/com/alibaba/otter/canal/client/running/kafka/KafkaClientRunningTest.java @@ -1,5 +1,6 @@ package com.alibaba.otter.canal.client.running.kafka; +import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; @@ -39,9 +40,9 @@ public class KafkaClientRunningTest extends AbstractKafkaTest { connector.subscribe(); while (running) { try { - Message message = connector.getWithoutAck(3L, TimeUnit.SECONDS); - if (message != null) { - System.out.println(message); + List messages = connector.getWithoutAck(3L, TimeUnit.SECONDS); + if (messages != null) { + System.out.println(messages); } connector.ack(); } catch (WakeupException e) { @@ -49,7 +50,7 @@ public class KafkaClientRunningTest extends AbstractKafkaTest { } } connector.unsubscribe(); - connector.disconnnect(); + connector.disconnect(); } }); diff --git a/deployer/src/main/resources/kafka.yml b/deployer/src/main/resources/kafka.yml index db4a9abe..735f723d 100644 --- a/deployer/src/main/resources/kafka.yml +++ b/deployer/src/main/resources/kafka.yml @@ -8,8 +8,11 @@ bufferMemory: 33554432 canalBatchSize: 50 # Canal get数据的超时时间, 单位: 毫秒, 空为不限超时 canalGetTimeout: 100 +flatMessage: true canalDestinations: - canalDestination: example topic: example - partition: \ No newline at end of file + partition: + + diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java new file mode 100644 index 00000000..e6aab531 --- /dev/null +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/FlatMessage.java @@ -0,0 +1,286 @@ +package com.alibaba.otter.canal.protocol; + +import java.io.Serializable; +import java.util.*; + +import com.google.common.collect.Table; +import com.google.protobuf.ByteString; + +/** + * @author machengyuan 2018-9-13 下午10:31:14 + * @version 1.0.0 + */ +public class FlatMessage implements Serializable { + + private static final long serialVersionUID = -3386650678735860050L; + + private long id; + private String database; + private String table; + private Boolean isDdl; + private String type; + private Long ts; + private String sql; + private Map sqlType; + private Map mysqlType; + private List> data; + private List> old; + + public FlatMessage() { + } + + public FlatMessage(long id){ + this.id = id; + } + + public long getId() { + return id; + } + + public void setId(long id) { + this.id = id; + } + + public String getDatabase() { + return database; + } + + public void setDatabase(String database) { + this.database = database; + } + + public String getTable() { + return table; + } + + public void setTable(String table) { + this.table = table; + } + + public Boolean getIsDdl() { + return isDdl; + } + + public void setIsDdl(Boolean isDdl) { + this.isDdl = isDdl; + } + + public String getType() { + return type; + } + + public void setType(String type) { + this.type = type; + } + + public Long getTs() { + return ts; + } + + public void setTs(Long ts) { + this.ts = ts; + } + + public String getSql() { + return sql; + } + + public void setSql(String sql) { + this.sql = sql; + } + + public Map getSqlType() { + return sqlType; + } + + public void setSqlType(Map sqlType) { + this.sqlType = sqlType; + } + + public Map getMysqlType() { + return mysqlType; + } + + public void setMysqlType(Map mysqlType) { + this.mysqlType = mysqlType; + } + + public List> getData() { + return data; + } + + public void setData(List> data) { + this.data = data; + } + + public List> getOld() { + return old; + } + + public void setOld(List> old) { + this.old = old; + } + + public static List messageConverter(Message message) { + try { + if (message == null) { + return null; + } + + List flatMessages = new ArrayList<>(); + + List rawEntries = message.getRawEntries(); + + for (ByteString byteString : rawEntries) { + CanalEntry.Entry entry = CanalEntry.Entry.parseFrom(byteString); + if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN + || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) { + continue; + } + + CanalEntry.RowChange rowChange; + try { + rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); + } catch (Exception e) { + throw new RuntimeException( + "ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), + e); + } + + CanalEntry.EventType eventType = rowChange.getEventType(); + + FlatMessage flatMessage = new FlatMessage(message.getId()); + flatMessages.add(flatMessage); + flatMessage.setDatabase(entry.getHeader().getSchemaName()); + flatMessage.setTable(entry.getHeader().getTableName()); + flatMessage.setIsDdl(rowChange.getIsDdl()); + flatMessage.setType(eventType.toString()); + flatMessage.setTs(System.currentTimeMillis()); + flatMessage.setSql(rowChange.getSql()); + + if (!rowChange.getIsDdl()) { + Map sqlType = new LinkedHashMap<>(); + Map mysqlType = new LinkedHashMap<>(); + List> data = new ArrayList<>(); + List> old = new ArrayList<>(); + + Set updateSet = new HashSet<>(); + for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { + if (eventType != CanalEntry.EventType.INSERT && eventType != CanalEntry.EventType.UPDATE + && eventType != CanalEntry.EventType.DELETE) { + continue; + } + + Map row = new LinkedHashMap<>(); + List columns; + + if (eventType == CanalEntry.EventType.DELETE) { + columns = rowData.getBeforeColumnsList(); + } else { + columns = rowData.getAfterColumnsList(); + } + + for (CanalEntry.Column column : columns) { + sqlType.put(column.getName(), column.getSqlType()); + mysqlType.put(column.getName(), column.getMysqlType()); + row.put(column.getName(), column.getValue()); + // 获取update为true的字段 + if (column.getUpdated()) { + updateSet.add(column.getName()); + } + } + if (!row.isEmpty()) { + data.add(row); + } + + if (eventType == CanalEntry.EventType.UPDATE) { + Map rowOld = new LinkedHashMap<>(); + for (CanalEntry.Column column : rowData.getBeforeColumnsList()) { + if (updateSet.contains(column.getName())) { + rowOld.put(column.getName(), column.getValue()); + } + } + // update操作将记录修改前的值 + if (!rowOld.isEmpty()) { + old.add(rowOld); + } + } + } + if (!sqlType.isEmpty()) { + flatMessage.setSqlType(sqlType); + } + if (!mysqlType.isEmpty()) { + flatMessage.setMysqlType(mysqlType); + } + if (!data.isEmpty()) { + flatMessage.setData(data); + } + if (!old.isEmpty()) { + flatMessage.setOld(old); + } + } + } + return flatMessages; + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + public static FlatMessage[] messagePartition(FlatMessage flatMessage, Integer partitionsNum, + Table pkHashConfig) { + FlatMessage[] partitionMessages = new FlatMessage[partitionsNum]; + + String pk = pkHashConfig.get(flatMessage.getDatabase(), flatMessage.getTable()); + if (pk == null || flatMessage.getIsDdl()) { + partitionMessages[0] = flatMessage; + } else { + if (flatMessage.getData() != null) { + int idx = 0; + for (Map row : flatMessage.getData()) { + String value = row.get(pk); + if (value == null) { + value = ""; + } + int hash = value.hashCode(); + int pkHash = Math.abs(hash) % partitionsNum; + + FlatMessage flatMessageTmp = partitionMessages[pkHash]; + if (flatMessageTmp == null) { + flatMessageTmp = new FlatMessage(flatMessage.getId()); + partitionMessages[pkHash] = flatMessageTmp; + flatMessageTmp.setDatabase(flatMessage.getDatabase()); + flatMessageTmp.setTable(flatMessage.getTable()); + flatMessageTmp.setIsDdl(flatMessage.getIsDdl()); + flatMessageTmp.setType(flatMessage.getType()); + flatMessageTmp.setSql(flatMessage.getSql()); + flatMessageTmp.setSqlType(flatMessage.getSqlType()); + flatMessageTmp.setMysqlType(flatMessage.getMysqlType()); + } + List> data = flatMessageTmp.getData(); + if (data == null) { + data = new ArrayList<>(); + flatMessageTmp.setData(data); + } + data.add(row); + if (flatMessage.getOld() != null && !flatMessage.getOld().isEmpty()) { + List> old = flatMessageTmp.getOld(); + if (old == null) { + old = new ArrayList<>(); + flatMessageTmp.setOld(old); + } + old.add(flatMessage.getOld().get(idx)); + } + idx++; + } + } + } + return partitionMessages; + } + + @Override + public String toString() { + return "FlatMessage{" + "id=" + id + ", database='" + database + '\'' + ", table='" + table + '\'' + ", isDdl=" + + isDdl + ", type='" + type + '\'' + ", ts=" + ts + ", sql='" + sql + '\'' + ", sqlType=" + sqlType + + ", mysqlType=" + mysqlType + ", data=" + data + ", old=" + old + '}'; + } +} 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 0685f76e..22f15229 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,5 +1,6 @@ package com.alibaba.otter.canal.kafka; +import java.util.List; import java.util.Properties; import java.util.concurrent.Future; @@ -11,6 +12,8 @@ import org.apache.kafka.common.serialization.StringSerializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.alibaba.fastjson.JSON; +import com.alibaba.otter.canal.protocol.FlatMessage; import com.alibaba.otter.canal.protocol.Message; /** @@ -25,7 +28,12 @@ public class CanalKafkaProducer { private Producer producer; + private Producer producer2; // 用于扁平message的数据投递 + + private KafkaProperties kafkaProperties; + public void init(KafkaProperties kafkaProperties) { + this.kafkaProperties = kafkaProperties; Properties properties = new Properties(); properties.put("bootstrap.servers", kafkaProperties.getServers()); properties.put("acks", "all"); @@ -34,15 +42,26 @@ public class CanalKafkaProducer { properties.put("linger.ms", kafkaProperties.getLingerMs()); properties.put("buffer.memory", kafkaProperties.getBufferMemory()); properties.put("key.serializer", StringSerializer.class.getName()); - properties.put("value.serializer", MessageSerializer.class.getName()); - producer = new KafkaProducer(properties); + if (!kafkaProperties.getFlatMessage()) { + properties.put("value.serializer", MessageSerializer.class.getName()); + producer = new KafkaProducer(properties); + } else { + properties.put("value.serializer", StringSerializer.class.getName()); + producer2 = new KafkaProducer(properties); + } + // producer.initTransactions(); } public void stop() { try { logger.info("## stop the kafka producer"); - producer.close(); + if (producer != null) { + producer.close(); + } + if (producer2 != null) { + producer2.close(); + } } catch (Throwable e) { logger.warn("##something goes wrong when stopping kafka producer:", e); } finally { @@ -53,14 +72,27 @@ public class CanalKafkaProducer { public void send(KafkaProperties.Topic topic, Message message, Callback callback) { try { // producer.beginTransaction(); - ProducerRecord record; - if (topic.getPartition() != null) { - record = new ProducerRecord(topic.getTopic(), topic.getPartition(), null, message); + if (!kafkaProperties.getFlatMessage()) { + ProducerRecord record; + if (topic.getPartition() != null) { + record = new ProducerRecord(topic.getTopic(), topic.getPartition(), null, message); + } else { + record = new ProducerRecord(topic.getTopic(), message); + } + Future future = producer.send(record); + future.get(); } else { - record = new ProducerRecord(topic.getTopic(), message); + // 发送扁平数据json + List flatMessages = FlatMessage.messageConverter(message); + if (flatMessages != null) { + for (FlatMessage flatMessage : flatMessages) { + ProducerRecord record = new ProducerRecord(topic.getTopic(), + JSON.toJSONString(flatMessage)); + producer2.send(record); + } + } } - Future future = producer.send(record); - future.get(); + // producer.commitTransaction(); callback.commit(); if (logger.isDebugEnabled()) { diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java b/server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java index 40f11bf5..82707ba4 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/KafkaProperties.java @@ -21,6 +21,7 @@ public class KafkaProperties { private boolean filterTransactionEntry = true; private int canalBatchSize = 50; private Long canalGetTimeout; + private boolean flatMessage = true; private List canalDestinations = new ArrayList(); @@ -160,6 +161,14 @@ public class KafkaProperties { this.canalGetTimeout = canalGetTimeout; } + public boolean getFlatMessage() { + return flatMessage; + } + + public void setFlatMessage(boolean flatMessage) { + this.flatMessage = flatMessage; + } + public List getCanalDestinations() { return canalDestinations; } diff --git a/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java b/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java index 9ac5a834..a4f74c90 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java @@ -311,9 +311,10 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C } if (CollectionUtils.isEmpty(events.getEvents())) { - logger.debug("getWithoutAck successfully, clientId:{} batchSize:{} but result is null", - clientIdentity.getClientId(), - batchSize); + // logger.debug("getWithoutAck successfully, clientId:{} batchSize:{} but result + // is null", + // clientIdentity.getClientId(), + // batchSize); return new Message(-1, true, new ArrayList()); // 返回空包，避免生成batchId，浪费性能 } else { // 记录到流式信息 From db56effd911fffc60fa068993d3e9008c4707754 Mon Sep 17 00:00:00 2001 From: mcy Date: Fri, 14 Sep 2018 11:45:52 +0800 Subject: [PATCH 2/3] =?UTF-8?q?=E6=95=B4=E7=90=86=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../client/adapter/loader/CanalAdapterKafkaWorker.java | 8 ++------ 1 file changed, 2 insertions(+), 6 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 d7324cda..72320bb5 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 @@ -156,18 +156,14 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker { TimeUnit.SECONDS.sleep(1L); } } - } catch ( - - Exception e) { + } catch (Exception e) { logger.error(e.getMessage(), e); } } executor.shutdown(); - try - - { + try { connector.unsubscribe(); } catch (WakeupException e) { // No-op. Continue process From 7005c11ac2b64abad4704c1bab1212d657f5e614 Mon Sep 17 00:00:00 2001 From: mcy Date: Mon, 17 Sep 2018 11:10:50 +0800 Subject: [PATCH 3/3] =?UTF-8?q?fix=E5=BA=8F=E5=88=97=E5=8C=96=E7=9A=84?= =?UTF-8?q?=E4=B8=80=E4=BA=9B=E9=97=AE=E9=A2=98,=20kafka=E7=94=9F=E4=BA=A7?= =?UTF-8?q?=E7=AB=AF=E7=BB=9F=E4=B8=80=E4=BD=BF=E7=94=A8=E5=BC=82=E6=AD=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../canal/client/impl/SimpleCanalConnector.java | 15 +++++++-------- .../canal/client/kafka/MessageDeserializer.java | 3 ++- .../otter/canal/kafka/CanalKafkaProducer.java | 4 ++-- 3 files changed, 11 insertions(+), 11 deletions(-) diff --git a/client/src/main/java/com/alibaba/otter/canal/client/impl/SimpleCanalConnector.java b/client/src/main/java/com/alibaba/otter/canal/client/impl/SimpleCanalConnector.java index b7951912..a73baadc 100644 --- a/client/src/main/java/com/alibaba/otter/canal/client/impl/SimpleCanalConnector.java +++ b/client/src/main/java/com/alibaba/otter/canal/client/impl/SimpleCanalConnector.java @@ -180,8 +180,8 @@ public class SimpleCanalConnector implements CanalConnector { Ack ackBody = Ack.parseFrom(ack.getBody()); if (ackBody.getErrorCode() > 0) { - throw new CanalClientException("something goes wrong when doing authentication: " - + ackBody.getErrorMessage()); + throw new CanalClientException( + "something goes wrong when doing authentication: " + ackBody.getErrorMessage()); } connected = true; @@ -323,7 +323,9 @@ public class SimpleCanalConnector implements CanalConnector { Packet p = Packet.parseFrom(data); switch (p.getType()) { case MESSAGES: { - if (!p.getCompression().equals(Compression.NONE)) { + // if (!p.getCompression().equals(Compression.NONE)) { + if (!p.getCompression().equals(Compression.NONE) + && !p.getCompression().equals(Compression.COMPRESSIONCOMPATIBLEPROTO2)) { throw new CanalClientException("compression is not supported in this connector"); } @@ -360,11 +362,8 @@ public class SimpleCanalConnector implements CanalConnector { .setBatchId(batchId) .build(); try { - writeWithHeader(Packet.newBuilder() - .setType(PacketType.CLIENTACK) - .setBody(ca.toByteString()) - .build() - .toByteArray()); + writeWithHeader( + Packet.newBuilder().setType(PacketType.CLIENTACK).setBody(ca.toByteString()).build().toByteArray()); } catch (IOException e) { throw new CanalClientException(e); } diff --git a/client/src/main/java/com/alibaba/otter/canal/client/kafka/MessageDeserializer.java b/client/src/main/java/com/alibaba/otter/canal/client/kafka/MessageDeserializer.java index e716d2d6..b5640cfd 100644 --- a/client/src/main/java/com/alibaba/otter/canal/client/kafka/MessageDeserializer.java +++ b/client/src/main/java/com/alibaba/otter/canal/client/kafka/MessageDeserializer.java @@ -36,7 +36,8 @@ public class MessageDeserializer implements Deserializer { CanalPacket.Packet p = CanalPacket.Packet.parseFrom(data); switch (p.getType()) { case MESSAGES: { - if (!p.getCompression().equals(CanalPacket.Compression.NONE)) { + if (!p.getCompression().equals(CanalPacket.Compression.NONE) + && !p.getCompression().equals(CanalPacket.Compression.COMPRESSIONCOMPATIBLEPROTO2)) { throw new CanalClientException("compression is not supported in this connector"); } 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 22f15229..84656fb1 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 @@ -79,8 +79,8 @@ public class CanalKafkaProducer { } else { record = new ProducerRecord(topic.getTopic(), message); } - Future future = producer.send(record); - future.get(); + + producer.send(record); } else { // 发送扁平数据json List flatMessages = FlatMessage.messageConverter(message);