chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

fixed compatible and code format

Browse Source
This commit is contained in:
jianghang.loujh
2022-05-13 11:59:57 +08:00
parent 25be0259a4
commit 33733aef25
5 changed files with 167 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQConstants.java
View File

@@ -1,7 +1,5 @@
package com.alibaba.otter.canal.connector.pulsarmq.config; package com.alibaba.otter.canal.connector.pulsarmq.config;
import org.joda.time.Seconds;
/** /**
* PulsarMQ配置 * PulsarMQ配置
* *
@@ -10,19 +8,20 @@ import org.joda.time.Seconds;
* @since 1 by chad at 2021/9/15 新增配置文件 * @since 1 by chad at 2021/9/15 新增配置文件
*/ */
public class PulsarMQConstants { public class PulsarMQConstants {
public static final String ROOT = "pulsarmq";
public static final String ROOT = "pulsarmq";
/** /**
* pulsar服务连接地址 * pulsar服务连接地址
*/ */
public static final String PULSARMQ_SERVER_URL = ROOT + "." + "serverUrl"; public static final String PULSARMQ_SERVER_URL = ROOT + "." + "serverUrl";
/** /**
* pulsar服务角色token，需要有对应token的生产者权限 * pulsar服务角色token，需要有对应token的生产者权限
*/ */
public static final String PULSARMQ_ROLE_TOKEN = ROOT + "." + "roleToken"; public static final String PULSARMQ_ROLE_TOKEN = ROOT + "." + "roleToken";
/** /**
* topic前缀 * 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，认为是同一个消费实例 * 消费都订阅名称，将以该名称为消费者身份标识，同一个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超时秒数 * 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队列名称后缀是否大写 * 自动生成的 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服务器地址 * Pulsar admin服务器地址
*/ */
public static final String PULSARMQ_ADMIN_SERVER_URL = ROOT + "." + "adminServerUrl"; public static final String PULSARMQ_ADMIN_SERVER_URL = ROOT + "." + "adminServerUrl";
} }

6
connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/config/PulsarMQProducerConfig.java
View File

@@ -4,16 +4,18 @@ import com.alibaba.otter.canal.connector.core.config.MQProperties;
/** /**
* Pulsar生产者配置 * Pulsar生产者配置
*
* @author chad * @author chad
* @date 2021/9/15 11:23 * @date 2021/9/15 11:23
* @since 1 by chad at 2021/9/15 新增 * @since 1 by chad at 2021/9/15 新增
*/ */
public class PulsarMQProducerConfig extends MQProperties { public class PulsarMQProducerConfig extends MQProperties {
/** /**
* pulsar服务连接地址 * pulsar服务连接地址
* <p> * <p>
* broker: pulsar://localhost:6650<br/> * broker: pulsar://localhost:6650<br/>
* httpUrl: http://localhost:8080 * httpUrl: http://localhost:8080
* </p> * </p>
*/ */
private String serverUrl; private String serverUrl;

93
connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumer.java
View File

@@ -1,20 +1,24 @@
package com.alibaba.otter.canal.connector.pulsarmq.consumer; 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.fastjson2.JSON;
import com.alibaba.otter.canal.common.utils.MQUtil; import com.alibaba.otter.canal.common.utils.MQUtil;
import com.alibaba.otter.canal.connector.core.config.CanalConstants; 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.consumer.CommonMessage;
import com.alibaba.otter.canal.connector.core.spi.CanalMsgConsumer; 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.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.connector.pulsarmq.config.PulsarMQConstants;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.exception.CanalClientException; import com.alibaba.otter.canal.protocol.exception.CanalClientException;
import com.google.common.collect.Lists; 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 实现 * Pulsar consumer SPI 实现
@@ -24,19 +28,20 @@ import java.util.concurrent.TimeUnit;
*/ */
@SPI("pulsarmq") @SPI("pulsarmq")
public class CanalPulsarMQConsumer implements CanalMsgConsumer { public class CanalPulsarMQConsumer implements CanalMsgConsumer {
/** /**
* 连接pulsar客户端 * 连接pulsar客户端
*/ */
private PulsarClient pulsarClient; private PulsarClient pulsarClient;
private Consumer<byte[]> pulsarMQConsumer; private Consumer<byte[]> 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 * 环境连接URL
*/ */
private String serviceUrl; private String serviceUrl;
/** /**
* 角色认证token * 角色认证token
*/ */
private String roleToken; private String roleToken;
/** /**
* 订阅客户端名称 * 订阅客户端名称
*/ */
private String subscriptName; private String subscriptName;
/** /**
* 每次批量获取数据的最大条目数，默认30 * 每次批量获取数据的最大条目数，默认30
*/ */
private int batchSize = 30; private int batchSize = 30;
/** /**
* 与{@code batchSize}一起决定批量获取的数据大小 * 与{@code batchSize}一起决定批量获取的数据大小 当：
* 当：
* <p> * <p>
* 1. {@code batchSize} 条消息未消费时<br/> * 1. {@code batchSize} 条消息未消费时<br/>
* 2. 距上一次批量消费时间达到{@code batchTimeoutSeconds}秒时 * 2. 距上一次批量消费时间达到{@code batchTimeoutSeconds}秒时
* </p> * </p>
* 任一条件满足，即执行批量消费 * 任一条件满足，即执行批量消费
*/ */
private int getBatchTimeoutSeconds = 30; private int getBatchTimeoutSeconds = 30;
/** /**
* 批量处理消息时，一次批量处理的超时时间 * 批量处理消息时，一次批量处理的超时时间
* <p> * <p>
* 该时间应该根据{@code batchSize}和{@code batchTimeoutSeconds}合理设置 * 该时间应该根据{@code batchSize}和{@code batchTimeoutSeconds}合理设置
* </p> * </p>
*/ */
private long batchProcessTimeout = 60 * 1000; private long batchProcessTimeout = 60 * 1000;
/** /**
* 消费失败后的重试秒数，默认60秒 * 消费失败后的重试秒数，默认60秒
*/ */
private int redeliveryDelaySeconds = 60; private int redeliveryDelaySeconds = 60;
/** /**
* 当客户端接收到消息，30秒还没有返回ack给服务端时，ack超时，会重新消费该消息 * 当客户端接收到消息，30秒还没有返回ack给服务端时，ack超时，会重新消费该消息
*/ */
private int ackTimeoutSeconds = 30; private int ackTimeoutSeconds = 30;
/** /**
* 是否开启消息失败重试功能，默认开启 * 是否开启消息失败重试功能，默认开启
*/ */
private boolean isRetry = true; private boolean isRetry = true;
/** /**
* <p> * <p>
* true重试(-RETRY)和死信队列(-DLQ)后缀为大写，有些地方创建的为小写，需确保正确 * true重试(-RETRY)和死信队列(-DLQ)后缀为大写，有些地方创建的为小写，需确保正确
* </p> * </p>
*/ */
private boolean isRetryDLQUpperCase = false; private boolean isRetryDLQUpperCase = false;
/** /**
* 最大重试次数 * 最大重试次数
*/ */
private int maxRedeliveryCount = 128; private int maxRedeliveryCount = 128;
@Override @Override
public void init(Properties properties, String topic, String groupId) { 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.serviceUrl = properties.getProperty(PulsarMQConstants.PULSARMQ_SERVER_URL);
this.roleToken = properties.getProperty(PulsarMQConstants.PULSARMQ_ROLE_TOKEN); 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，避免所有的都是同一个订阅者名称 // 采用groupId作为subscriptName，避免所有的都是同一个订阅者名称
this.subscriptName = groupId; if (StringUtils.isEmpty(this.subscriptName)) {
this.subscriptName = groupId;
}
if (StringUtils.isEmpty(this.subscriptName)) { if (StringUtils.isEmpty(this.subscriptName)) {
throw new RuntimeException("Pulsar Consumer subscriptName required"); throw new RuntimeException("Pulsar Consumer subscriptName required");
} }
@@ -153,9 +160,9 @@ public class CanalPulsarMQConsumer implements CanalMsgConsumer {
} }
// 连接创建客户端 // 连接创建客户端
try { try {
//AuthenticationDataProvider // AuthenticationDataProvider
ClientBuilder builder = PulsarClient.builder().serviceUrl(serviceUrl); ClientBuilder builder = PulsarClient.builder().serviceUrl(serviceUrl);
if(StringUtils.isNotEmpty(roleToken)) { if (StringUtils.isNotEmpty(roleToken)) {
builder.authentication(AuthenticationFactory.token(roleToken)); builder.authentication(AuthenticationFactory.token(roleToken));
} }
pulsarClient = builder.build(); pulsarClient = builder.build();
@@ -174,15 +181,14 @@ public class CanalPulsarMQConsumer implements CanalMsgConsumer {
builder.subscriptionType(SubscriptionType.Failover); builder.subscriptionType(SubscriptionType.Failover);
builder builder
// 调用consumer.negativeAcknowledge(message) （即nack）来表示消费失败的消息 // 调用consumer.negativeAcknowledge(message) （即nack）来表示消费失败的消息
// 在指定的时间进行重新消费，默认是1分钟。 // 在指定的时间进行重新消费，默认是1分钟。
.negativeAckRedeliveryDelay(this.redeliveryDelaySeconds, TimeUnit.SECONDS) .negativeAckRedeliveryDelay(this.redeliveryDelaySeconds, TimeUnit.SECONDS)
.subscriptionName(this.subscriptName) .subscriptionName(this.subscriptName);
;
if (this.isRetry) { if (this.isRetry) {
DeadLetterPolicy.DeadLetterPolicyBuilder dlqBuilder = DeadLetterPolicy.builder() DeadLetterPolicy.DeadLetterPolicyBuilder dlqBuilder = DeadLetterPolicy.builder()
// 最大重试次数 // 最大重试次数
.maxRedeliverCount(this.maxRedeliveryCount); .maxRedeliverCount(this.maxRedeliveryCount);
// 指定重试队列，不是多个或通配符topic才能判断重试队列 // 指定重试队列，不是多个或通配符topic才能判断重试队列
if (!MQUtil.isPatternTag(this.topic)) { if (!MQUtil.isPatternTag(this.topic)) {
String retryTopic = this.topic + (this.isRetryDLQUpperCase ? "-RETRY" : "-retry"); String retryTopic = this.topic + (this.isRetryDLQUpperCase ? "-RETRY" : "-retry");
@@ -191,19 +197,17 @@ public class CanalPulsarMQConsumer implements CanalMsgConsumer {
dlqBuilder.deadLetterTopic(dlqTopic); dlqBuilder.deadLetterTopic(dlqTopic);
} }
//默认关闭，如果需要重试则开启 // 默认关闭，如果需要重试则开启
builder.enableRetry(true) builder.enableRetry(true).deadLetterPolicy(dlqBuilder.build());
.deadLetterPolicy(dlqBuilder.build());
} }
// ack超时 // ack超时
builder.ackTimeout(this.ackTimeoutSeconds, TimeUnit.SECONDS); builder.ackTimeout(this.ackTimeoutSeconds, TimeUnit.SECONDS);
// pulsar批量获取消息设置 // pulsar批量获取消息设置
builder.batchReceivePolicy(new BatchReceivePolicy.Builder() builder.batchReceivePolicy(new BatchReceivePolicy.Builder().maxNumMessages(this.batchSize)
.maxNumMessages(this.batchSize) .timeout(this.getBatchTimeoutSeconds, TimeUnit.SECONDS)
.timeout(this.getBatchTimeoutSeconds, TimeUnit.SECONDS) .build());
.build());
try { try {
this.pulsarMQConsumer = builder.subscribe(); this.pulsarMQConsumer = builder.subscribe();
@@ -225,17 +229,14 @@ public class CanalPulsarMQConsumer implements CanalMsgConsumer {
this.lastGetBatchMessage = messages; this.lastGetBatchMessage = messages;
for (org.apache.pulsar.client.api.Message<byte[]> msg : messages) { for (org.apache.pulsar.client.api.Message<byte[]> msg : messages) {
byte[] data = msg.getData(); byte[] data = msg.getData();
/*if (!this.flatMessage) { if (!this.flatMessage) {
Message message = CanalMessageSerializerUtil.deserializer(data); Message message = CanalMessageSerializerUtil.deserializer(data);
List<CommonMessage> list = MessageUtil.convert(message); List<CommonMessage> list = MessageUtil.convert(message);
messageList.addAll(list); messageList.addAll(list);
} else { } else {
CommonMessage commonMessage = JSON.parseObject(data, CommonMessage.class); CommonMessage commonMessage = JSON.parseObject(data, CommonMessage.class);
messageList.add(commonMessage); messageList.add(commonMessage);
}*/ }
// CanalMessageSerializerUtil.deserializer(data) 会转换失败
CommonMessage commonMessage = JSON.parseObject(data, CommonMessage.class);
messageList.add(commonMessage);
} }
} catch (PulsarClientException e) { } catch (PulsarClientException e) {
throw new CanalClientException("Receive pulsar batch message error", e); throw new CanalClientException("Receive pulsar batch message error", e);

175
connector/pulsarmq-connector/src/main/java/com/alibaba/otter/canal/connector/pulsarmq/producer/CanalPulsarMQProducer.java
View File

@@ -1,5 +1,17 @@
package com.alibaba.otter.canal.connector.pulsarmq.producer; 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.JSON;
import com.alibaba.fastjson2.JSONWriter.Feature; import com.alibaba.fastjson2.JSONWriter.Feature;
import com.alibaba.otter.canal.common.utils.ExecutorTemplate; 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.connector.pulsarmq.config.PulsarMQProducerConfig;
import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.FlatMessage; 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 实现 * PulsarMQ Producer SPI 实现
@@ -39,23 +38,22 @@ import java.util.stream.Collectors;
@SPI("pulsarmq") @SPI("pulsarmq")
public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQProducer { public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQProducer {
private static final Logger logger = LoggerFactory.getLogger(CanalPulsarMQProducer.class);
private static final Map<String, Producer<byte[]>> 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<String, Producer<byte[]>> PRODUCERS = new HashMap<>();
protected ThreadPoolExecutor sendPartitionExecutor;
/** /**
* pulsar客户端，管理连接 * pulsar客户端，管理连接
*/ */
protected PulsarClient client; protected PulsarClient client;
/** /**
* Pulsar admin 客户端 * Pulsar admin 客户端
*/ */
protected PulsarAdmin pulsarAdmin; protected PulsarAdmin pulsarAdmin;
@Override @Override
public void init(Properties properties) { public void init(Properties properties) {
@@ -68,9 +66,9 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
// 初始化连接客户端 // 初始化连接客户端
try { try {
ClientBuilder builder = PulsarClient.builder() ClientBuilder builder = PulsarClient.builder()
// 填写pulsar的连接地址 // 填写pulsar的连接地址
.serviceUrl(pulsarMQProducerConfig.getServerUrl()); .serviceUrl(pulsarMQProducerConfig.getServerUrl());
if(StringUtils.isNotEmpty(pulsarMQProducerConfig.getRoleToken())) { if (StringUtils.isNotEmpty(pulsarMQProducerConfig.getRoleToken())) {
// 角色权限认证的token // 角色权限认证的token
builder.authentication(AuthenticationFactory.token(pulsarMQProducerConfig.getRoleToken())); builder.authentication(AuthenticationFactory.token(pulsarMQProducerConfig.getRoleToken()));
} }
@@ -80,7 +78,7 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
} }
// 初始化Pulsar admin // 初始化Pulsar admin
if(StringUtils.isNotEmpty(pulsarMQProducerConfig.getAdminServerUrl())) { if (StringUtils.isNotEmpty(pulsarMQProducerConfig.getAdminServerUrl())) {
try { try {
pulsarAdmin = PulsarAdmin.builder().serviceHttpUrl(pulsarMQProducerConfig.getAdminServerUrl()).build(); pulsarAdmin = PulsarAdmin.builder().serviceHttpUrl(pulsarMQProducerConfig.getAdminServerUrl()).build();
} catch (PulsarClientException e) { } catch (PulsarClientException e) {
@@ -91,12 +89,12 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
// 加载所有生产者 --> topic可能有正则或表名，无法确认所有topic，在使用时再加载 // 加载所有生产者 --> topic可能有正则或表名，无法确认所有topic，在使用时再加载
int parallelPartitionSendThreadSize = mqProperties.getParallelSendThreadSize(); int parallelPartitionSendThreadSize = mqProperties.getParallelSendThreadSize();
sendPartitionExecutor = new ThreadPoolExecutor(parallelPartitionSendThreadSize, sendPartitionExecutor = new ThreadPoolExecutor(parallelPartitionSendThreadSize,
parallelPartitionSendThreadSize, parallelPartitionSendThreadSize,
0, 0,
TimeUnit.SECONDS, TimeUnit.SECONDS,
new ArrayBlockingQueue<>(parallelPartitionSendThreadSize * 2), new ArrayBlockingQueue<>(parallelPartitionSendThreadSize * 2),
new NamedThreadFactory("MQ-Parallel-Sender-Partition"), new NamedThreadFactory("MQ-Parallel-Sender-Partition"),
new ThreadPoolExecutor.CallerRunsPolicy()); new ThreadPoolExecutor.CallerRunsPolicy());
} }
/** /**
@@ -119,12 +117,13 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
if (!StringUtils.isEmpty(roleToken)) { if (!StringUtils.isEmpty(roleToken)) {
tmpProperties.setRoleToken(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)) { if (!StringUtils.isEmpty(topicTenantPrefix)) {
tmpProperties.setTopicTenantPrefix(topicTenantPrefix); tmpProperties.setTopicTenantPrefix(topicTenantPrefix);
} }
String adminServerUrl = PropertiesUtils.getProperty(properties, PulsarMQConstants.PULSARMQ_ADMIN_SERVER_URL); String adminServerUrl = PropertiesUtils.getProperty(properties, PulsarMQConstants.PULSARMQ_ADMIN_SERVER_URL);
if(!StringUtils.isEmpty(adminServerUrl)) { if (!StringUtils.isEmpty(adminServerUrl)) {
tmpProperties.setAdminServerUrl(adminServerUrl); tmpProperties.setAdminServerUrl(adminServerUrl);
} }
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {
@@ -140,8 +139,8 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
* </p> * </p>
* *
* @param destination 消息目标信息 * @param destination 消息目标信息
* @param message 消息 * @param message 消息
* @param callback 消息发送结果回调 * @param callback 消息发送结果回调
* @return void * @return void
* @date 2021/9/2 22:01 * @date 2021/9/2 22:01
* @author chad * @author chad
@@ -154,9 +153,8 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
try { try {
if (!StringUtils.isEmpty(destination.getDynamicTopic())) { if (!StringUtils.isEmpty(destination.getDynamicTopic())) {
// 动态topic // 动态topic
Map<String, com.alibaba.otter.canal.protocol.Message> messageMap = MQMessageUtils.messageTopics(message, Map<String, com.alibaba.otter.canal.protocol.Message> messageMap = MQMessageUtils
destination.getTopic(), .messageTopics(message, destination.getTopic(), destination.getDynamicTopic());
destination.getDynamicTopic());
for (Map.Entry<String, com.alibaba.otter.canal.protocol.Message> entry : messageMap.entrySet()) { for (Map.Entry<String, com.alibaba.otter.canal.protocol.Message> entry : messageMap.entrySet()) {
String topicName = entry.getKey().replace('.', '_'); String topicName = entry.getKey().replace('.', '_');
@@ -195,16 +193,17 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
* @author chad * @author chad
* @since 1.0.0 by chad at 2021/9/2: 新增 * @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的分区数 // 获取当前topic的分区数
Integer partitionNum = MQMessageUtils.parseDynamicTopicPartition(topicName, Integer partitionNum = MQMessageUtils.parseDynamicTopicPartition(topicName,
destination.getDynamicTopicPartitionNum()); destination.getDynamicTopicPartitionNum());
if (partitionNum == null) { if (partitionNum == null) {
partitionNum = destination.getPartitionsNum(); partitionNum = destination.getPartitionsNum();
} }
// 创建多分区topic // 创建多分区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); createMultipleTopic(topicName, partitionNum);
} }
@@ -221,10 +220,10 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
} }
// 串行分区 // 串行分区
com.alibaba.otter.canal.protocol.Message[] messages = MQMessageUtils.messagePartition(datas, com.alibaba.otter.canal.protocol.Message[] messages = MQMessageUtils.messagePartition(datas,
message.getId(), message.getId(),
partitionNum, partitionNum,
destination.getPartitionHash(), destination.getPartitionHash(),
mqProperties.isDatabaseHash()); mqProperties.isDatabaseHash());
// 发送 // 发送
int len = messages.length; int len = messages.length;
for (int i = 0; i < len; i++) { for (int i = 0; i < len; i++) {
@@ -254,9 +253,9 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
for (FlatMessage flatMessage : flatMessages) { for (FlatMessage flatMessage : flatMessages) {
FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage, FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage,
partitionNum, partitionNum,
destination.getPartitionHash(), destination.getPartitionHash(),
mqProperties.isDatabaseHash()); mqProperties.isDatabaseHash());
int length = partitionFlatMessage.length; int length = partitionFlatMessage.length;
for (int i = 0; i < length; i++) { for (int i = 0; i < length; i++) {
// 增加null判断,issue #3267 // 增加null判断,issue #3267
@@ -290,9 +289,9 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
/** /**
* 发送原始消息，需要做分区处理 * 发送原始消息，需要做分区处理
* *
* @param topic topic * @param topic topic
* @param partitionNum 目标分区 * @param partitionNum 目标分区
* @param msg 原始消息内容 * @param msg 原始消息内容
* @return void * @return void
* @date 2021/9/10 17:55 * @date 2021/9/10 17:55
* @author chad * @author chad
@@ -303,8 +302,9 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
byte[] msgBytes = CanalMessageSerializerUtil.serializer(msg, mqProperties.isFilterTransactionEntry()); byte[] msgBytes = CanalMessageSerializerUtil.serializer(msg, mqProperties.isFilterTransactionEntry());
try { try {
MessageId msgResultId = producer.newMessage() MessageId msgResultId = producer.newMessage()
.property(MSG_PROPERTY_PARTITION_NAME, String.valueOf(partitionNum)) .property(MSG_PROPERTY_PARTITION_NAME, String.valueOf(partitionNum))
.value(msgBytes).send(); .value(msgBytes)
.send();
// todo 判断发送结果 // todo 判断发送结果
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {
logger.debug("Send Message to topic:{} Result: {}", topic, msgResultId); 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 扁平消息 * @param flatMessages 扁平消息
* @return void * @return void
* @date 2021/9/10 18:22 * @date 2021/9/10 18:22
@@ -328,13 +328,12 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
Producer<byte[]> producer = getProducer(topic); Producer<byte[]> producer = getProducer(topic);
for (FlatMessage f : flatMessages) { for (FlatMessage f : flatMessages) {
try { try {
MessageId msgResultId = producer MessageId msgResultId = producer.newMessage()
.newMessage() .property(MSG_PROPERTY_PARTITION_NAME, String.valueOf(partition))
.property(MSG_PROPERTY_PARTITION_NAME, String.valueOf(partition)) .value(JSON.toJSONBytes(f, Feature.WriteNulls))
.value(JSON.toJSONBytes(f, Feature.WriteNulls)) .send()
.send() //
// ;
;
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {
logger.debug("Send Messages to topic:{} Result: {}", topic, msgResultId); logger.debug("Send Messages to topic:{} Result: {}", topic, msgResultId);
} }
@@ -346,10 +345,11 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
/** /**
* 创建多分区topic * 创建多分区topic
*
* @param topic * @param topic
* @param partitionNum * @param partitionNum
*/ */
private void createMultipleTopic(String topic,Integer partitionNum) { private void createMultipleTopic(String topic, Integer partitionNum) {
// 拼接topic前缀 // 拼接topic前缀
PulsarMQProducerConfig pulsarMQProperties = (PulsarMQProducerConfig) this.mqProperties; PulsarMQProducerConfig pulsarMQProperties = (PulsarMQProducerConfig) this.mqProperties;
String prefix = pulsarMQProperties.getTopicTenantPrefix(); String prefix = pulsarMQProperties.getTopicTenantPrefix();
@@ -368,8 +368,10 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
// TODO 无论是否报错，都继续后续的操作，此处不进行阻塞 // TODO 无论是否报错，都继续后续的操作，此处不进行阻塞
} }
} }
/** /**
* 获取topic * 获取topic
*
* @param topic * @param topic
* @return * @return
*/ */
@@ -396,10 +398,10 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
// 创建指定topic的生产者 // 创建指定topic的生产者
producer = client.newProducer() producer = client.newProducer()
.topic(fullTopic) .topic(fullTopic)
// 指定路由器 // 指定路由器
.messageRouter(new MessageRouterImpl(topic)) .messageRouter(new MessageRouterImpl(topic))
.create(); .create();
// 放入缓存 // 放入缓存
PRODUCERS.put(topic, producer); PRODUCERS.put(topic, producer);
} }
@@ -412,6 +414,26 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
return producer; 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自定义路由策略 * Pulsar自定义路由策略
* *
@@ -421,9 +443,10 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
* @since 2 by chad at 2021/9/17 修改为msg自带目标分区 * @since 2 by chad at 2021/9/17 修改为msg自带目标分区
*/ */
private static class MessageRouterImpl implements MessageRouter { private static class MessageRouterImpl implements MessageRouter {
private String topicLocal; private String topicLocal;
public MessageRouterImpl(String topicLocal) { public MessageRouterImpl(String topicLocal){
this.topicLocal = topicLocal; this.topicLocal = topicLocal;
} }
@@ -435,7 +458,8 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
try { try {
partition = Integer.parseInt(partitionStr); partition = Integer.parseInt(partitionStr);
} catch (NumberFormatException e) { } 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创建时设置的分区数 // topic创建时设置的分区数
@@ -447,21 +471,4 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
return partition; 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();
}
} }

26
connector/pulsarmq-connector/src/test/java/com/alibaba/otter/canal/connector/pulsarmq/consumer/CanalPulsarMQConsumerTest.java
View File

@@ -1,23 +1,19 @@
package com.alibaba.otter.canal.connector.pulsarmq.consumer; 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.fastjson2.JSON;
import com.alibaba.otter.canal.connector.core.config.CanalConstants; 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.consumer.CommonMessage;
import com.alibaba.otter.canal.connector.pulsarmq.config.PulsarMQConstants; 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 * @author chad
@@ -26,7 +22,7 @@ import static org.junit.Assert.*;
*/ */
public class CanalPulsarMQConsumerTest { public class CanalPulsarMQConsumerTest {
private Properties properties; private Properties properties;
private CanalPulsarMQConsumer consumer; private CanalPulsarMQConsumer consumer;