From a7720b51e4586a626e548556229764b7c38c7bb9 Mon Sep 17 00:00:00 2001 From: renyansongno1 <45755446+renyansongno1@users.noreply.github.com> Date: Mon, 19 Apr 2021 10:14:27 +0800 Subject: [PATCH] =?UTF-8?q?=E6=94=AF=E6=8C=81rocketmq=E7=9A=84=E6=B6=88?= =?UTF-8?q?=E6=81=AF=E6=8C=89=E7=85=A7tag=E5=8E=BB=E5=8F=91=E9=80=81=20(#3?= =?UTF-8?q?438)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 支持rocketmq的消息按照tag去发送 并对应修改了配置文件 * 将tag的设计转移到Rocketmq的单独配置项 --- .../connector/rocketmq/config/RocketMQConstants.java | 1 + .../rocketmq/config/RocketMQProducerConfig.java | 9 +++++++++ .../rocketmq/producer/CanalRocketMQProducer.java | 12 ++++++++---- deployer/src/main/resources/canal.properties | 1 + 4 files changed, 19 insertions(+), 4 deletions(-) diff --git a/connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/config/RocketMQConstants.java b/connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/config/RocketMQConstants.java index 5f728627..022ac54a 100644 --- a/connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/config/RocketMQConstants.java +++ b/connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/config/RocketMQConstants.java @@ -17,6 +17,7 @@ public class RocketMQConstants { public static final String ROCKETMQ_NAMESRV_ADDR = ROOT + "." + "namesrv.addr"; public static final String ROCKETMQ_RETRY_TIMES_WHEN_SEND_FAILED = ROOT + "." + "retry.times.when.send.failed"; public static final String ROCKETMQ_VIP_CHANNEL_ENABLED = ROOT + "." + "vip.channel.enabled"; + public static final String ROCKETMQ_TAG = ROOT + "." + "tag"; public static final String ROCKETMQ_ACCESS_CHANNEL = ROOT + "." + "access.channel"; public static final String ROCKETMQ_BATCH_SIZE = ROOT + "." + "batch.size"; diff --git a/connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/config/RocketMQProducerConfig.java b/connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/config/RocketMQProducerConfig.java index 5aebb9fb..d7d3ab51 100644 --- a/connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/config/RocketMQProducerConfig.java +++ b/connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/config/RocketMQProducerConfig.java @@ -17,6 +17,7 @@ public class RocketMQProducerConfig extends MQProperties { private String namesrvAddr; private int retryTimesWhenSendFailed = 0; private boolean vipChannelEnabled = false; + private String tag; public String getProducerGroup() { return producerGroup; @@ -73,4 +74,12 @@ public class RocketMQProducerConfig extends MQProperties { public void setVipChannelEnabled(boolean vipChannelEnabled) { this.vipChannelEnabled = vipChannelEnabled; } + + public String getTag() { + return tag; + } + + public void setTag(String tag) { + this.tag = tag; + } } diff --git a/connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/producer/CanalRocketMQProducer.java b/connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/producer/CanalRocketMQProducer.java index 12a02469..b013821b 100644 --- a/connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/producer/CanalRocketMQProducer.java +++ b/connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/producer/CanalRocketMQProducer.java @@ -140,6 +140,10 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ if (!StringUtils.isEmpty(vipChannelEnabled)) { rocketMQProperties.setVipChannelEnabled(Boolean.parseBoolean(vipChannelEnabled)); } + String tag = properties.getProperty(RocketMQConstants.ROCKETMQ_TAG); + if (!StringUtils.isEmpty(tag)) { + rocketMQProperties.setTag(tag); + } } @Override @@ -203,7 +207,7 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ if (dataPartition != null) { final int index = i; template.submit(() -> { - Message data = new Message(topicName, CanalMessageSerializerUtil.serializer(dataPartition, + Message data = new Message(topicName, ((RocketMQProducerConfig)this.mqProperties).getTag(), CanalMessageSerializerUtil.serializer(dataPartition, mqProperties.isFilterTransactionEntry())); sendMessage(data, index); }); @@ -213,7 +217,7 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ template.waitForResult(); } else { final int partition = destination.getPartition() != null ? destination.getPartition() : 0; - Message data = new Message(topicName, CanalMessageSerializerUtil.serializer(message, + Message data = new Message(topicName, ((RocketMQProducerConfig)this.mqProperties).getTag(), CanalMessageSerializerUtil.serializer(message, mqProperties.isFilterTransactionEntry())); sendMessage(data, partition); } @@ -250,7 +254,7 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ final int index = i; template.submit(() -> { List messages = flatMessagePart.stream() - .map(flatMessage -> new Message(topicName, JSON.toJSONBytes(flatMessage, + .map(flatMessage -> new Message(topicName, ((RocketMQProducerConfig)this.mqProperties).getTag(), JSON.toJSONBytes(flatMessage, SerializerFeature.WriteMapNullValue))) .collect(Collectors.toList()); // 批量发送 @@ -264,7 +268,7 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ } else { final int partition = destination.getPartition() != null ? destination.getPartition() : 0; List messages = flatMessages.stream() - .map(flatMessage -> new Message(topicName, JSON.toJSONBytes(flatMessage, + .map(flatMessage -> new Message(topicName, ((RocketMQProducerConfig)this.mqProperties).getTag(), JSON.toJSONBytes(flatMessage, SerializerFeature.WriteMapNullValue))) .collect(Collectors.toList()); // 批量发送 diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index b1d65b83..bb7b9f2b 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -157,6 +157,7 @@ rocketmq.namespace = rocketmq.namesrv.addr = 127.0.0.1:9876 rocketmq.retry.times.when.send.failed = 0 rocketmq.vip.channel.enabled = false +rocketmq.tag = testTag ################################################## ######### RabbitMQ #############