From 33733aef252e1869e2e4f926e4c4a16e33c290f6 Mon Sep 17 00:00:00 2001 From: "jianghang.loujh" Date: Fri, 13 May 2022 11:59:57 +0800 Subject: [PATCH] fixed compatible and code format --- .../pulsarmq/config/PulsarMQConstants.java | 28 ++- .../config/PulsarMQProducerConfig.java | 6 +- .../consumer/CanalPulsarMQConsumer.java | 93 +++++----- .../producer/CanalPulsarMQProducer.java | 175 +++++++++--------- .../consumer/CanalPulsarMQConsumerTest.java | 28 ++- 5 files changed, 167 insertions(+), 163 deletions(-) diff --git a/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQConstants.java b/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQConstants.java index 55c8d80b..1c9c1a8e 100644 --- a/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQConstants.java +++ b/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQConstants.java @@ -1,7 +1,5 @@ package com.alibaba.otter.canal.connector.pulsarmq.config; -import org.joda.time.Seconds; - /** * PulsarMQ配置 * @@ -10,19 +8,20 @@ import org.joda.time.Seconds; * @since 1 by chad at 2021/9/15 新增配置文件 */ public class PulsarMQConstants { - public static final String ROOT = "pulsarmq"; + + public static final String ROOT = "pulsarmq"; /** * pulsar服务连接地址 */ - public static final String PULSARMQ_SERVER_URL = ROOT + "." + "serverUrl"; + public static final String PULSARMQ_SERVER_URL = ROOT + "." + "serverUrl"; /** * pulsar服务角色token，需要有对应token的生产者权限 */ - public static final String PULSARMQ_ROLE_TOKEN = ROOT + "." + "roleToken"; + public static final String PULSARMQ_ROLE_TOKEN = ROOT + "." + "roleToken"; /** * topic前缀 */ - public static final String PULSARMQ_TOPIC_TENANT_PREFIX = ROOT + "." + "topicTenantPrefix"; + public static final String PULSARMQ_TOPIC_TENANT_PREFIX = ROOT + "." + "topicTenantPrefix"; /**** 消费者 *****/ /** @@ -32,35 +31,34 @@ public class PulsarMQConstants { /** * 批量处理超时时间 */ - public static final String PULSARMQ_BATCH_PROCESS_TIMEOUT = ROOT + "." + "batchProcessTimeout"; + public static final String PULSARMQ_BATCH_PROCESS_TIMEOUT = ROOT + "." + "batchProcessTimeout"; /** * 消费都订阅名称，将以该名称为消费者身份标识，同一个subscriptName，认为是同一个消费实例 */ - public static final String PULSARMQ_SUBSCRIPT_NAME = ROOT + "." + "subscriptName"; + public static final String PULSARMQ_SUBSCRIPT_NAME = ROOT + "." + "subscriptName"; /** * 重试间隔秒数 */ - public static final String PULSARMQ_REDELIVERY_DELAY_SECONDS = ROOT + "." + "redeliveryDelaySeconds"; + public static final String PULSARMQ_REDELIVERY_DELAY_SECONDS = ROOT + "." + "redeliveryDelaySeconds"; /** * ACK超时秒数 */ - public static final String PULSARMQ_ACK_TIMEOUT_SECONDS = ROOT + "." + "ackTimeoutSeconds"; + public static final String PULSARMQ_ACK_TIMEOUT_SECONDS = ROOT + "." + "ackTimeoutSeconds"; /** * 是否开启消费重试 */ - public static final String PULSARMQ_IS_RETRY = ROOT + "." + "isRetry"; + public static final String PULSARMQ_IS_RETRY = ROOT + "." + "isRetry"; /** * 自动生成的 retry dlq队列名称后缀是否大写 */ - public static final String PULSARMQ_IS_RETRY_DLQ_UPPERCASE = ROOT + "." + "isRetryDLQUpperCase"; + public static final String PULSARMQ_IS_RETRY_DLQ_UPPERCASE = ROOT + "." + "isRetryDLQUpperCase"; /** * 最大重试次数 */ - public static final String PULSARMQ_MAX_REDELIVERY_COUNT = ROOT + "." + "maxRedeliveryCount"; + public static final String PULSARMQ_MAX_REDELIVERY_COUNT = ROOT + "." + "maxRedeliveryCount"; /** * Pulsar admin服务器地址 */ - public static final String PULSARMQ_ADMIN_SERVER_URL = ROOT + "." + "adminServerUrl"; - + public static final String PULSARMQ_ADMIN_SERVER_URL = ROOT + "." + "adminServerUrl"; } diff --git a/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQProducerConfig.java b/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQProducerConfig.java index a984e79e..1ace4aa0 100644 --- a/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQProducerConfig.java +++ b/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQProducerConfig.java @@ -4,16 +4,18 @@ import com.alibaba.otter.canal.connector.core.config.MQProperties; /** * Pulsar生产者配置 + * * @author chad * @date 2021/9/15 11:23 * @since 1 by chad at 2021/9/15 新增 */ public class PulsarMQProducerConfig extends MQProperties { + /** * pulsar服务连接地址 *

- * broker: pulsar://localhost:6650
- * httpUrl: http://localhost:8080 + * broker: pulsar://localhost:6650
+ * httpUrl: http://localhost:8080 *

*/ private String serverUrl; diff --git a/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumer.java b/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumer.java index 5f3e4ee6..0a535468 100644 --- a/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumer.java +++ b/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumer.java @@ -1,20 +1,24 @@ package com.alibaba.otter.canal.connector.pulsarmq.consumer; +import java.util.List; +import java.util.Properties; +import java.util.concurrent.TimeUnit; + +import org.apache.commons.lang.StringUtils; +import org.apache.pulsar.client.api.*; + import com.alibaba.fastjson2.JSON; import com.alibaba.otter.canal.common.utils.MQUtil; import com.alibaba.otter.canal.connector.core.config.CanalConstants; import com.alibaba.otter.canal.connector.core.consumer.CommonMessage; import com.alibaba.otter.canal.connector.core.spi.CanalMsgConsumer; import com.alibaba.otter.canal.connector.core.spi.SPI; +import com.alibaba.otter.canal.connector.core.util.CanalMessageSerializerUtil; +import com.alibaba.otter.canal.connector.core.util.MessageUtil; import com.alibaba.otter.canal.connector.pulsarmq.config.PulsarMQConstants; +import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.exception.CanalClientException; import com.google.common.collect.Lists; -import org.apache.commons.lang.StringUtils; -import org.apache.pulsar.client.api.*; - -import java.util.List; -import java.util.Properties; -import java.util.concurrent.TimeUnit; /** * Pulsar consumer SPI 实现 @@ -24,19 +28,20 @@ import java.util.concurrent.TimeUnit; */ @SPI("pulsarmq") public class CanalPulsarMQConsumer implements CanalMsgConsumer { + /** * 连接pulsar客户端 */ - private PulsarClient pulsarClient; - private Consumer pulsarMQConsumer; + private PulsarClient pulsarClient; + private Consumer pulsarMQConsumer; /** * 是否为扁平消息 */ - private boolean flatMessage = false; + private boolean flatMessage = false; /** * 主题名称 */ - private String topic; + private String topic; /** * 单线程控制 */ @@ -44,58 +49,57 @@ public class CanalPulsarMQConsumer implements CanalMsgConsumer { /** * 环境连接URL */ - private String serviceUrl; + private String serviceUrl; /** * 角色认证token */ - private String roleToken; + private String roleToken; /** * 订阅客户端名称 */ - private String subscriptName; + private String subscriptName; /** * 每次批量获取数据的最大条目数，默认30 */ - private int batchSize = 30; + private int batchSize = 30; /** - * 与{@code batchSize}一起决定批量获取的数据大小 - * 当： + * 与{@code batchSize}一起决定批量获取的数据大小 当： *

* 1. {@code batchSize} 条消息未消费时
* 2. 距上一次批量消费时间达到{@code batchTimeoutSeconds}秒时 *

* 任一条件满足，即执行批量消费 */ - private int getBatchTimeoutSeconds = 30; + private int getBatchTimeoutSeconds = 30; /** * 批量处理消息时，一次批量处理的超时时间 *

* 该时间应该根据{@code batchSize}和{@code batchTimeoutSeconds}合理设置 *

*/ - private long batchProcessTimeout = 60 * 1000; + private long batchProcessTimeout = 60 * 1000; /** * 消费失败后的重试秒数，默认60秒 */ - private int redeliveryDelaySeconds = 60; + private int redeliveryDelaySeconds = 60; /** * 当客户端接收到消息，30秒还没有返回ack给服务端时，ack超时，会重新消费该消息 */ - private int ackTimeoutSeconds = 30; + private int ackTimeoutSeconds = 30; /** * 是否开启消息失败重试功能，默认开启 */ - private boolean isRetry = true; + private boolean isRetry = true; /** *

* true重试(-RETRY)和死信队列(-DLQ)后缀为大写，有些地方创建的为小写，需确保正确 *

*/ - private boolean isRetryDLQUpperCase = false; + private boolean isRetryDLQUpperCase = false; /** * 最大重试次数 */ - private int maxRedeliveryCount = 128; + private int maxRedeliveryCount = 128; @Override public void init(Properties properties, String topic, String groupId) { @@ -106,9 +110,12 @@ public class CanalPulsarMQConsumer implements CanalMsgConsumer { } this.serviceUrl = properties.getProperty(PulsarMQConstants.PULSARMQ_SERVER_URL); this.roleToken = properties.getProperty(PulsarMQConstants.PULSARMQ_ROLE_TOKEN); - //this.subscriptName = properties.getProperty(PulsarMQConstants.PULSARMQ_SUBSCRIPT_NAME); + this.subscriptName = properties.getProperty(PulsarMQConstants.PULSARMQ_SUBSCRIPT_NAME); // 采用groupId作为subscriptName，避免所有的都是同一个订阅者名称 - this.subscriptName = groupId; + if (StringUtils.isEmpty(this.subscriptName)) { + this.subscriptName = groupId; + } + if (StringUtils.isEmpty(this.subscriptName)) { throw new RuntimeException("Pulsar Consumer subscriptName required"); } @@ -153,9 +160,9 @@ public class CanalPulsarMQConsumer implements CanalMsgConsumer { } // 连接创建客户端 try { - //AuthenticationDataProvider + // AuthenticationDataProvider ClientBuilder builder = PulsarClient.builder().serviceUrl(serviceUrl); - if(StringUtils.isNotEmpty(roleToken)) { + if (StringUtils.isNotEmpty(roleToken)) { builder.authentication(AuthenticationFactory.token(roleToken)); } pulsarClient = builder.build(); @@ -174,15 +181,14 @@ public class CanalPulsarMQConsumer implements CanalMsgConsumer { builder.subscriptionType(SubscriptionType.Failover); builder - // 调用consumer.negativeAcknowledge(message) （即nack）来表示消费失败的消息 - // 在指定的时间进行重新消费，默认是1分钟。 - .negativeAckRedeliveryDelay(this.redeliveryDelaySeconds, TimeUnit.SECONDS) - .subscriptionName(this.subscriptName) - ; + // 调用consumer.negativeAcknowledge(message) （即nack）来表示消费失败的消息 + // 在指定的时间进行重新消费，默认是1分钟。 + .negativeAckRedeliveryDelay(this.redeliveryDelaySeconds, TimeUnit.SECONDS) + .subscriptionName(this.subscriptName); if (this.isRetry) { DeadLetterPolicy.DeadLetterPolicyBuilder dlqBuilder = DeadLetterPolicy.builder() - // 最大重试次数 - .maxRedeliverCount(this.maxRedeliveryCount); + // 最大重试次数 + .maxRedeliverCount(this.maxRedeliveryCount); // 指定重试队列，不是多个或通配符topic才能判断重试队列 if (!MQUtil.isPatternTag(this.topic)) { String retryTopic = this.topic + (this.isRetryDLQUpperCase ? "-RETRY" : "-retry"); @@ -191,19 +197,17 @@ public class CanalPulsarMQConsumer implements CanalMsgConsumer { dlqBuilder.deadLetterTopic(dlqTopic); } - //默认关闭，如果需要重试则开启 - builder.enableRetry(true) - .deadLetterPolicy(dlqBuilder.build()); + // 默认关闭，如果需要重试则开启 + builder.enableRetry(true).deadLetterPolicy(dlqBuilder.build()); } // ack超时 builder.ackTimeout(this.ackTimeoutSeconds, TimeUnit.SECONDS); // pulsar批量获取消息设置 - builder.batchReceivePolicy(new BatchReceivePolicy.Builder() - .maxNumMessages(this.batchSize) - .timeout(this.getBatchTimeoutSeconds, TimeUnit.SECONDS) - .build()); + builder.batchReceivePolicy(new BatchReceivePolicy.Builder().maxNumMessages(this.batchSize) + .timeout(this.getBatchTimeoutSeconds, TimeUnit.SECONDS) + .build()); try { this.pulsarMQConsumer = builder.subscribe(); @@ -225,17 +229,14 @@ public class CanalPulsarMQConsumer implements CanalMsgConsumer { this.lastGetBatchMessage = messages; for (org.apache.pulsar.client.api.Message msg : messages) { byte[] data = msg.getData(); - /*if (!this.flatMessage) { + if (!this.flatMessage) { Message message = CanalMessageSerializerUtil.deserializer(data); List list = MessageUtil.convert(message); messageList.addAll(list); } else { CommonMessage commonMessage = JSON.parseObject(data, CommonMessage.class); messageList.add(commonMessage); - }*/ - // CanalMessageSerializerUtil.deserializer(data) 会转换失败 - CommonMessage commonMessage = JSON.parseObject(data, CommonMessage.class); - messageList.add(commonMessage); + } } } catch (PulsarClientException e) { throw new CanalClientException("Receive pulsar batch message error", e); diff --git a/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/producer/CanalPulsarMQProducer.java b/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/producer/CanalPulsarMQProducer.java index 631f208a..1b488935 100644 --- a/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/producer/CanalPulsarMQProducer.java +++ b/connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/producer/CanalPulsarMQProducer.java @@ -1,5 +1,17 @@ package com.alibaba.otter.canal.connector.pulsarmq.producer; +import java.util.*; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +import org.apache.commons.lang.StringUtils; +import org.apache.pulsar.client.admin.PulsarAdmin; +import org.apache.pulsar.client.admin.PulsarAdminException; +import org.apache.pulsar.client.api.*; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSONWriter.Feature; import com.alibaba.otter.canal.common.utils.ExecutorTemplate; @@ -16,19 +28,6 @@ import com.alibaba.otter.canal.connector.pulsarmq.config.PulsarMQConstants; import com.alibaba.otter.canal.connector.pulsarmq.config.PulsarMQProducerConfig; import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.FlatMessage; -import org.apache.commons.lang.StringUtils; -import org.apache.pulsar.client.admin.PulsarAdmin; -import org.apache.pulsar.client.admin.PulsarAdminException; -import org.apache.pulsar.client.api.*; -import org.apache.pulsar.shade.com.google.gson.JsonParser; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.*; -import java.util.concurrent.ArrayBlockingQueue; -import java.util.concurrent.ThreadPoolExecutor; -import java.util.concurrent.TimeUnit; -import java.util.stream.Collectors; /** * PulsarMQ Producer SPI 实现 @@ -39,23 +38,22 @@ import java.util.stream.Collectors; @SPI("pulsarmq") public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQProducer { - private static final Logger logger = LoggerFactory.getLogger(CanalPulsarMQProducer.class); - - private static final Map> PRODUCERS = new HashMap<>(); - - protected ThreadPoolExecutor sendPartitionExecutor; /** * 消息体分区属性名称 */ - public static final String MSG_PROPERTY_PARTITION_NAME = "partitionNum"; + public static final String MSG_PROPERTY_PARTITION_NAME = "partitionNum"; + private static final Logger logger = LoggerFactory + .getLogger(CanalPulsarMQProducer.class); + private static final Map> PRODUCERS = new HashMap<>(); + protected ThreadPoolExecutor sendPartitionExecutor; /** * pulsar客户端，管理连接 */ - protected PulsarClient client; + protected PulsarClient client; /** * Pulsar admin 客户端 */ - protected PulsarAdmin pulsarAdmin; + protected PulsarAdmin pulsarAdmin; @Override public void init(Properties properties) { @@ -68,9 +66,9 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ // 初始化连接客户端 try { ClientBuilder builder = PulsarClient.builder() - // 填写pulsar的连接地址 - .serviceUrl(pulsarMQProducerConfig.getServerUrl()); - if(StringUtils.isNotEmpty(pulsarMQProducerConfig.getRoleToken())) { + // 填写pulsar的连接地址 + .serviceUrl(pulsarMQProducerConfig.getServerUrl()); + if (StringUtils.isNotEmpty(pulsarMQProducerConfig.getRoleToken())) { // 角色权限认证的token builder.authentication(AuthenticationFactory.token(pulsarMQProducerConfig.getRoleToken())); } @@ -80,7 +78,7 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ } // 初始化Pulsar admin - if(StringUtils.isNotEmpty(pulsarMQProducerConfig.getAdminServerUrl())) { + if (StringUtils.isNotEmpty(pulsarMQProducerConfig.getAdminServerUrl())) { try { pulsarAdmin = PulsarAdmin.builder().serviceHttpUrl(pulsarMQProducerConfig.getAdminServerUrl()).build(); } catch (PulsarClientException e) { @@ -91,12 +89,12 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ // 加载所有生产者 --> topic可能有正则或表名，无法确认所有topic，在使用时再加载 int parallelPartitionSendThreadSize = mqProperties.getParallelSendThreadSize(); sendPartitionExecutor = new ThreadPoolExecutor(parallelPartitionSendThreadSize, - parallelPartitionSendThreadSize, - 0, - TimeUnit.SECONDS, - new ArrayBlockingQueue<>(parallelPartitionSendThreadSize * 2), - new NamedThreadFactory("MQ-Parallel-Sender-Partition"), - new ThreadPoolExecutor.CallerRunsPolicy()); + parallelPartitionSendThreadSize, + 0, + TimeUnit.SECONDS, + new ArrayBlockingQueue<>(parallelPartitionSendThreadSize * 2), + new NamedThreadFactory("MQ-Parallel-Sender-Partition"), + new ThreadPoolExecutor.CallerRunsPolicy()); } /** @@ -119,12 +117,13 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ if (!StringUtils.isEmpty(roleToken)) { tmpProperties.setRoleToken(roleToken); } - String topicTenantPrefix = PropertiesUtils.getProperty(properties, PulsarMQConstants.PULSARMQ_TOPIC_TENANT_PREFIX); + String topicTenantPrefix = PropertiesUtils.getProperty(properties, + PulsarMQConstants.PULSARMQ_TOPIC_TENANT_PREFIX); if (!StringUtils.isEmpty(topicTenantPrefix)) { tmpProperties.setTopicTenantPrefix(topicTenantPrefix); } String adminServerUrl = PropertiesUtils.getProperty(properties, PulsarMQConstants.PULSARMQ_ADMIN_SERVER_URL); - if(!StringUtils.isEmpty(adminServerUrl)) { + if (!StringUtils.isEmpty(adminServerUrl)) { tmpProperties.setAdminServerUrl(adminServerUrl); } if (logger.isDebugEnabled()) { @@ -140,8 +139,8 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ *

* * @param destination 消息目标信息 - * @param message 消息 - * @param callback 消息发送结果回调 + * @param message 消息 + * @param callback 消息发送结果回调 * @return void * @date 2021/9/2 22:01 * @author chad @@ -154,9 +153,8 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ try { if (!StringUtils.isEmpty(destination.getDynamicTopic())) { // 动态topic - Map messageMap = MQMessageUtils.messageTopics(message, - destination.getTopic(), - destination.getDynamicTopic()); + Map messageMap = MQMessageUtils + .messageTopics(message, destination.getTopic(), destination.getDynamicTopic()); for (Map.Entry entry : messageMap.entrySet()) { String topicName = entry.getKey().replace('.', '_'); @@ -195,16 +193,17 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ * @author chad * @since 1.0.0 by chad at 2021/9/2: 新增 */ - public void send(final MQDestination destination, String topicName, com.alibaba.otter.canal.protocol.Message message) { + public void send(final MQDestination destination, String topicName, + com.alibaba.otter.canal.protocol.Message message) { // 获取当前topic的分区数 Integer partitionNum = MQMessageUtils.parseDynamicTopicPartition(topicName, - destination.getDynamicTopicPartitionNum()); + destination.getDynamicTopicPartitionNum()); if (partitionNum == null) { partitionNum = destination.getPartitionsNum(); } // 创建多分区topic - if(pulsarAdmin!=null && partitionNum!=null && partitionNum>0 && PRODUCERS.get(topicName)==null) { + if (pulsarAdmin != null && partitionNum != null && partitionNum > 0 && PRODUCERS.get(topicName) == null) { createMultipleTopic(topicName, partitionNum); } @@ -221,10 +220,10 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ } // 串行分区 com.alibaba.otter.canal.protocol.Message[] messages = MQMessageUtils.messagePartition(datas, - message.getId(), - partitionNum, - destination.getPartitionHash(), - mqProperties.isDatabaseHash()); + message.getId(), + partitionNum, + destination.getPartitionHash(), + mqProperties.isDatabaseHash()); // 发送 int len = messages.length; for (int i = 0; i < len; i++) { @@ -254,9 +253,9 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ for (FlatMessage flatMessage : flatMessages) { FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage, - partitionNum, - destination.getPartitionHash(), - mqProperties.isDatabaseHash()); + partitionNum, + destination.getPartitionHash(), + mqProperties.isDatabaseHash()); int length = partitionFlatMessage.length; for (int i = 0; i < length; i++) { // 增加null判断,issue #3267 @@ -290,9 +289,9 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ /** * 发送原始消息，需要做分区处理 * - * @param topic topic + * @param topic topic * @param partitionNum 目标分区 - * @param msg 原始消息内容 + * @param msg 原始消息内容 * @return void * @date 2021/9/10 17:55 * @author chad @@ -303,8 +302,9 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ byte[] msgBytes = CanalMessageSerializerUtil.serializer(msg, mqProperties.isFilterTransactionEntry()); try { MessageId msgResultId = producer.newMessage() - .property(MSG_PROPERTY_PARTITION_NAME, String.valueOf(partitionNum)) - .value(msgBytes).send(); + .property(MSG_PROPERTY_PARTITION_NAME, String.valueOf(partitionNum)) + .value(msgBytes) + .send(); // todo 判断发送结果 if (logger.isDebugEnabled()) { logger.debug("Send Message to topic:{} Result: {}", topic, msgResultId); @@ -317,7 +317,7 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ /** * 发送扁平消息 * - * @param topic topic主题 + * @param topic topic主题 * @param flatMessages 扁平消息 * @return void * @date 2021/9/10 18:22 @@ -328,13 +328,12 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ Producer producer = getProducer(topic); for (FlatMessage f : flatMessages) { try { - MessageId msgResultId = producer - .newMessage() - .property(MSG_PROPERTY_PARTITION_NAME, String.valueOf(partition)) - .value(JSON.toJSONBytes(f, Feature.WriteNulls)) - .send() - // - ; + MessageId msgResultId = producer.newMessage() + .property(MSG_PROPERTY_PARTITION_NAME, String.valueOf(partition)) + .value(JSON.toJSONBytes(f, Feature.WriteNulls)) + .send() + // + ; if (logger.isDebugEnabled()) { logger.debug("Send Messages to topic:{} Result: {}", topic, msgResultId); } @@ -346,10 +345,11 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ /** * 创建多分区topic + * * @param topic * @param partitionNum */ - private void createMultipleTopic(String topic,Integer partitionNum) { + private void createMultipleTopic(String topic, Integer partitionNum) { // 拼接topic前缀 PulsarMQProducerConfig pulsarMQProperties = (PulsarMQProducerConfig) this.mqProperties; String prefix = pulsarMQProperties.getTopicTenantPrefix(); @@ -368,8 +368,10 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ // TODO 无论是否报错，都继续后续的操作，此处不进行阻塞 } } + /** * 获取topic + * * @param topic * @return */ @@ -396,10 +398,10 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ // 创建指定topic的生产者 producer = client.newProducer() - .topic(fullTopic) - // 指定路由器 - .messageRouter(new MessageRouterImpl(topic)) - .create(); + .topic(fullTopic) + // 指定路由器 + .messageRouter(new MessageRouterImpl(topic)) + .create(); // 放入缓存 PRODUCERS.put(topic, producer); } @@ -412,6 +414,26 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ return producer; } + @Override + public void stop() { + logger.info("## Stop PulsarMQ producer##"); + + for (Producer p : PRODUCERS.values()) { + try { + if (null != p && p.isConnected()) { + p.close(); + } + } catch (PulsarClientException e) { + logger.warn("close producer name: {}, topic: {}, error: {}", + p.getProducerName(), + p.getTopic(), + e.getMessage()); + } + } + + super.stop(); + } + /** * Pulsar自定义路由策略 * @@ -421,9 +443,10 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ * @since 2 by chad at 2021/9/17 修改为msg自带目标分区 */ private static class MessageRouterImpl implements MessageRouter { + private String topicLocal; - public MessageRouterImpl(String topicLocal) { + public MessageRouterImpl(String topicLocal){ this.topicLocal = topicLocal; } @@ -435,7 +458,8 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ try { partition = Integer.parseInt(partitionStr); } catch (NumberFormatException e) { - logger.warn("Parse msg {} property failed for value: {}", MSG_PROPERTY_PARTITION_NAME, partitionStr); + logger + .warn("Parse msg {} property failed for value: {}", MSG_PROPERTY_PARTITION_NAME, partitionStr); } } // topic创建时设置的分区数 @@ -447,21 +471,4 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ return partition; } } - - @Override - public void stop() { - logger.info("## Stop PulsarMQ producer##"); - - for (Producer p : PRODUCERS.values()) { - try { - if (null != p && p.isConnected()) { - p.close(); - } - } catch (PulsarClientException e) { - logger.warn("close producer name: {}, topic: {}, error: {}", p.getProducerName(), p.getTopic(), e.getMessage()); - } - } - - super.stop(); - } } diff --git a/connector/pulsarmq-connector/src/test/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumerTest.java b/connector/pulsarmq-connector/src/test/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumerTest.java index 760d5b1b..b1a8b2ce 100644 --- a/connector/pulsarmq-connector/src/test/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumerTest.java +++ b/connector/pulsarmq-connector/src/test/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumerTest.java @@ -1,23 +1,19 @@ package com.alibaba.otter.canal.connector.pulsarmq.consumer; +import java.time.LocalTime; +import java.time.format.DateTimeFormatter; +import java.util.List; +import java.util.Properties; +import java.util.concurrent.TimeUnit; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + import com.alibaba.fastjson2.JSON; import com.alibaba.otter.canal.connector.core.config.CanalConstants; import com.alibaba.otter.canal.connector.core.consumer.CommonMessage; import com.alibaba.otter.canal.connector.pulsarmq.config.PulsarMQConstants; -import org.apache.commons.lang.StringUtils; -import org.junit.After; -import org.junit.Before; -import org.junit.Test; - -import java.time.LocalDateTime; -import java.time.LocalTime; -import java.time.format.DateTimeFormatter; -import java.util.List; -import java.util.Properties; -import java.util.UUID; -import java.util.concurrent.TimeUnit; - -import static org.junit.Assert.*; /** * @author chad @@ -26,7 +22,7 @@ import static org.junit.Assert.*; */ public class CanalPulsarMQConsumerTest { - private Properties properties; + private Properties properties; private CanalPulsarMQConsumer consumer; @@ -73,4 +69,4 @@ public class CanalPulsarMQConsumerTest { consumer.ack(); } } -} \ No newline at end of file +}