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 pull #5175 , reformat

Browse Source
This commit is contained in:
jianghang.loujh
2024-07-16 14:57:53 +08:00
parent 2b8a309ec7
commit 3439c48948
2 changed files with 107 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/AdapterProcessor.java
View File

@@ -1,5 +1,16 @@
package com.alibaba.otter.canal.adapter.launcher.loader; package com.alibaba.otter.canal.adapter.launcher.loader;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.adapter.launcher.common.SyncSwitch; import com.alibaba.otter.canal.adapter.launcher.common.SyncSwitch;
import com.alibaba.otter.canal.adapter.launcher.config.SpringContext; import com.alibaba.otter.canal.adapter.launcher.config.SpringContext;
import com.alibaba.otter.canal.client.adapter.OuterAdapter; import com.alibaba.otter.canal.client.adapter.OuterAdapter;
@@ -12,16 +23,6 @@ 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.ExtensionLoader; import com.alibaba.otter.canal.connector.core.spi.ExtensionLoader;
import com.alibaba.otter.canal.connector.core.spi.ProxyCanalMsgConsumer; import com.alibaba.otter.canal.connector.core.spi.ProxyCanalMsgConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
/** /**
* 适配处理器 * 适配处理器
@@ -31,27 +32,27 @@ import java.util.concurrent.TimeoutException;
*/ */
public class AdapterProcessor { public class AdapterProcessor {
private static final Logger logger = LoggerFactory.getLogger(AdapterProcessor.class); private static final Logger logger = LoggerFactory.getLogger(AdapterProcessor.class);
private static final String CONNECTOR_SPI_DIR = "/plugin"; private static final String CONNECTOR_SPI_DIR = "/plugin";
private static final String CONNECTOR_STANDBY_SPI_DIR = "/canal-adapter/plugin"; private static final String CONNECTOR_STANDBY_SPI_DIR = "/canal-adapter/plugin";
private CanalMsgConsumer canalMsgConsumer; private CanalMsgConsumer canalMsgConsumer;
private String canalDestination; // canal实例 private String canalDestination; // canal实例
private String groupId = null; // groupId private String groupId = null; // groupId
private List<List<OuterAdapter>> canalOuterAdapters; // 外部适配器 private List<List<OuterAdapter>> canalOuterAdapters; // 外部适配器
private CanalClientConfig canalClientConfig; // 配置 private CanalClientConfig canalClientConfig; // 配置
private ExecutorService groupInnerExecutorService; // 组内工作线程池 private ExecutorService groupInnerExecutorService; // 组内工作线程池
private volatile boolean running = false; // 是否运行中 private volatile boolean running = false; // 是否运行中
private Thread thread = null; private Thread thread = null;
private Thread.UncaughtExceptionHandler handler = (t, e) -> logger private Thread.UncaughtExceptionHandler handler = (t, e) -> logger
.error("parse events has an error", e); .error("parse events has an error", e);
private SyncSwitch syncSwitch; private SyncSwitch syncSwitch;
public AdapterProcessor(CanalClientConfig canalClientConfig, String destination, String groupId, public AdapterProcessor(CanalClientConfig canalClientConfig, String destination, String groupId,
List<List<OuterAdapter>> canalOuterAdapters) { List<List<OuterAdapter>> canalOuterAdapters){
this.canalClientConfig = canalClientConfig; this.canalClientConfig = canalClientConfig;
this.canalDestination = destination; this.canalDestination = destination;
this.groupId = groupId; this.groupId = groupId;
@@ -62,10 +63,12 @@ public class AdapterProcessor {
// load connector consumer // load connector consumer
ExtensionLoader<CanalMsgConsumer> loader = new ExtensionLoader<>(CanalMsgConsumer.class); ExtensionLoader<CanalMsgConsumer> loader = new ExtensionLoader<>(CanalMsgConsumer.class);
// see https://github.com/alibaba/canal/pull/5175
String key = destination + "_" + groupId; String key = destination + "_" + groupId;
canalMsgConsumer = new ProxyCanalMsgConsumer(loader canalMsgConsumer = new ProxyCanalMsgConsumer(loader.getExtension(canalClientConfig.getMode().toLowerCase(),
.getExtension(canalClientConfig.getMode().toLowerCase(), key, CONNECTOR_SPI_DIR, key,
CONNECTOR_STANDBY_SPI_DIR)); CONNECTOR_SPI_DIR,
CONNECTOR_STANDBY_SPI_DIR));
Properties properties = canalClientConfig.getConsumerProperties(); Properties properties = canalClientConfig.getConsumerProperties();
properties.put(CanalConstants.CANAL_MQ_FLAT_MESSAGE, canalClientConfig.getFlatMessage()); properties.put(CanalConstants.CANAL_MQ_FLAT_MESSAGE, canalClientConfig.getFlatMessage());
@@ -170,7 +173,7 @@ public class AdapterProcessor {
} }
int retry = canalClientConfig.getRetries() == null int retry = canalClientConfig.getRetries() == null
|| canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries(); || canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries();
if (retry == -1) { if (retry == -1) {
// 重试次数-1代表异常时一直阻塞重试 // 重试次数-1代表异常时一直阻塞重试
retry = Integer.MAX_VALUE; retry = Integer.MAX_VALUE;

146
connector/rocketmq-connector/src/main/java/com/alibaba/otter/canal/connector/rocketmq/producer/CanalRocketMQProducer.java
View File

@@ -1,5 +1,29 @@
package com.alibaba.otter.canal.connector.rocketmq.producer; package com.alibaba.otter.canal.connector.rocketmq.producer;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import org.apache.commons.lang.StringUtils;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl;
import org.apache.rocketmq.client.impl.producer.TopicPublishInfo;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.RPCHook;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson2.JSON; import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONWriter; import com.alibaba.fastjson2.JSONWriter;
import com.alibaba.otter.canal.common.CanalException; import com.alibaba.otter.canal.common.CanalException;
@@ -16,29 +40,6 @@ import com.alibaba.otter.canal.connector.core.util.CanalMessageSerializerUtil;
import com.alibaba.otter.canal.connector.rocketmq.config.RocketMQConstants; import com.alibaba.otter.canal.connector.rocketmq.config.RocketMQConstants;
import com.alibaba.otter.canal.connector.rocketmq.config.RocketMQProducerConfig; import com.alibaba.otter.canal.connector.rocketmq.config.RocketMQProducerConfig;
import com.alibaba.otter.canal.protocol.FlatMessage; import com.alibaba.otter.canal.protocol.FlatMessage;
import org.apache.commons.lang.StringUtils;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl;
import org.apache.rocketmq.client.impl.producer.TopicPublishInfo;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.RPCHook;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
/** /**
* RocketMQ Producer SPI 实现 * RocketMQ Producer SPI 实现
@@ -46,16 +47,18 @@ import java.util.stream.Collectors;
* @author rewerma 2020-01-27 * @author rewerma 2020-01-27
* @version 1.0.0 * @version 1.0.0
*/ */
@SPI("rocketmq") public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQProducer { @SPI("rocketmq")
public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQProducer {
private static final Logger logger = LoggerFactory.getLogger(CanalRocketMQProducer.class); private static final Logger logger = LoggerFactory.getLogger(CanalRocketMQProducer.class);
private DefaultMQProducer defaultMQProducer; private DefaultMQProducer defaultMQProducer;
private static final String CLOUD_ACCESS_CHANNEL = "cloud"; private static final String CLOUD_ACCESS_CHANNEL = "cloud";
private static final String NAMESPACE_SEPARATOR = "%"; private static final String NAMESPACE_SEPARATOR = "%";
protected ThreadPoolExecutor sendPartitionExecutor; protected ThreadPoolExecutor sendPartitionExecutor;
@Override public void init(Properties properties) { @Override
public void init(Properties properties) {
RocketMQProducerConfig rocketMQProperties = new RocketMQProducerConfig(); RocketMQProducerConfig rocketMQProperties = new RocketMQProducerConfig();
this.mqProperties = rocketMQProperties; this.mqProperties = rocketMQProperties;
super.init(properties); super.init(properties);
@@ -70,9 +73,9 @@ import java.util.stream.Collectors;
} }
defaultMQProducer = new DefaultMQProducer(rocketMQProperties.getProducerGroup(), defaultMQProducer = new DefaultMQProducer(rocketMQProperties.getProducerGroup(),
rpcHook, rpcHook,
rocketMQProperties.isEnableMessageTrace(), rocketMQProperties.isEnableMessageTrace(),
rocketMQProperties.getCustomizedTraceTopic()); rocketMQProperties.getCustomizedTraceTopic());
if (CLOUD_ACCESS_CHANNEL.equals(rocketMQProperties.getAccessChannel())) { if (CLOUD_ACCESS_CHANNEL.equals(rocketMQProperties.getAccessChannel())) {
defaultMQProducer.setAccessChannel(AccessChannel.CLOUD); defaultMQProducer.setAccessChannel(AccessChannel.CLOUD);
} }
@@ -91,12 +94,12 @@ import java.util.stream.Collectors;
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());
} }
private void loadRocketMQProperties(Properties properties) { private void loadRocketMQProperties(Properties properties) {
@@ -112,12 +115,12 @@ import java.util.stream.Collectors;
rocketMQProperties.setProducerGroup(producerGroup); rocketMQProperties.setProducerGroup(producerGroup);
} }
String enableMessageTrace = PropertiesUtils.getProperty(properties, String enableMessageTrace = PropertiesUtils.getProperty(properties,
RocketMQConstants.ROCKETMQ_ENABLE_MESSAGE_TRACE); RocketMQConstants.ROCKETMQ_ENABLE_MESSAGE_TRACE);
if (!StringUtils.isEmpty(enableMessageTrace)) { if (!StringUtils.isEmpty(enableMessageTrace)) {
rocketMQProperties.setEnableMessageTrace(Boolean.parseBoolean(enableMessageTrace)); rocketMQProperties.setEnableMessageTrace(Boolean.parseBoolean(enableMessageTrace));
} }
String customizedTraceTopic = PropertiesUtils.getProperty(properties, String customizedTraceTopic = PropertiesUtils.getProperty(properties,
RocketMQConstants.ROCKETMQ_CUSTOMIZED_TRACE_TOPIC); RocketMQConstants.ROCKETMQ_CUSTOMIZED_TRACE_TOPIC);
if (!StringUtils.isEmpty(customizedTraceTopic)) { if (!StringUtils.isEmpty(customizedTraceTopic)) {
rocketMQProperties.setCustomizedTraceTopic(customizedTraceTopic); rocketMQProperties.setCustomizedTraceTopic(customizedTraceTopic);
} }
@@ -134,7 +137,7 @@ import java.util.stream.Collectors;
rocketMQProperties.setRetryTimesWhenSendFailed(Integer.parseInt(retry)); rocketMQProperties.setRetryTimesWhenSendFailed(Integer.parseInt(retry));
} }
String vipChannelEnabled = PropertiesUtils.getProperty(properties, String vipChannelEnabled = PropertiesUtils.getProperty(properties,
RocketMQConstants.ROCKETMQ_VIP_CHANNEL_ENABLED); RocketMQConstants.ROCKETMQ_VIP_CHANNEL_ENABLED);
if (!StringUtils.isEmpty(vipChannelEnabled)) { if (!StringUtils.isEmpty(vipChannelEnabled)) {
rocketMQProperties.setVipChannelEnabled(Boolean.parseBoolean(vipChannelEnabled)); rocketMQProperties.setVipChannelEnabled(Boolean.parseBoolean(vipChannelEnabled));
} }
@@ -144,15 +147,14 @@ import java.util.stream.Collectors;
} }
} }
@Override public void send(MQDestination destination, com.alibaba.otter.canal.protocol.Message message, @Override
Callback callback) { public void send(MQDestination destination, com.alibaba.otter.canal.protocol.Message message, Callback callback) {
ExecutorTemplate template = new ExecutorTemplate(sendExecutor); ExecutorTemplate template = new ExecutorTemplate(sendExecutor);
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('.', '_');
@@ -184,7 +186,7 @@ import java.util.stream.Collectors;
com.alibaba.otter.canal.protocol.Message message) { com.alibaba.otter.canal.protocol.Message message) {
// 获取当前topic的分区数 // 获取当前topic的分区数
Integer partitionNum = MQMessageUtils.parseDynamicTopicPartition(topicName, Integer partitionNum = MQMessageUtils.parseDynamicTopicPartition(topicName,
destination.getDynamicTopicPartitionNum()); destination.getDynamicTopicPartitionNum());
// 获取topic的队列数为分区数 // 获取topic的队列数为分区数
if (partitionNum == null) { if (partitionNum == null) {
@@ -200,10 +202,10 @@ import java.util.stream.Collectors;
MQMessageUtils.EntryRowData[] datas = MQMessageUtils.buildMessageData(message, buildExecutor); MQMessageUtils.EntryRowData[] datas = MQMessageUtils.buildMessageData(message, buildExecutor);
// 串行分区 // 串行分区
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 length = messages.length; int length = messages.length;
ExecutorTemplate template = new ExecutorTemplate(sendPartitionExecutor); ExecutorTemplate template = new ExecutorTemplate(sendPartitionExecutor);
@@ -213,9 +215,9 @@ import java.util.stream.Collectors;
final int index = i; final int index = i;
template.submit(() -> { template.submit(() -> {
Message data = new Message(topicName, Message data = new Message(topicName,
((RocketMQProducerConfig) this.mqProperties).getTag(), ((RocketMQProducerConfig) this.mqProperties).getTag(),
CanalMessageSerializerUtil.serializer(dataPartition, CanalMessageSerializerUtil.serializer(dataPartition,
mqProperties.isFilterTransactionEntry())); mqProperties.isFilterTransactionEntry()));
sendMessage(data, index); sendMessage(data, index);
}); });
} }
@@ -225,8 +227,8 @@ import java.util.stream.Collectors;
} else { } else {
final int partition = destination.getPartition() != null ? destination.getPartition() : 0; final int partition = destination.getPartition() != null ? destination.getPartition() : 0;
Message data = new Message(topicName, Message data = new Message(topicName,
((RocketMQProducerConfig) this.mqProperties).getTag(), ((RocketMQProducerConfig) this.mqProperties).getTag(),
CanalMessageSerializerUtil.serializer(message, mqProperties.isFilterTransactionEntry())); CanalMessageSerializerUtil.serializer(message, mqProperties.isFilterTransactionEntry()));
sendMessage(data, partition); sendMessage(data, partition);
} }
} else { } else {
@@ -243,9 +245,9 @@ import java.util.stream.Collectors;
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
@@ -262,10 +264,12 @@ import java.util.stream.Collectors;
final int index = i; final int index = i;
template.submit(() -> { template.submit(() -> {
List<Message> messages = flatMessagePart.stream() List<Message> messages = flatMessagePart.stream()
.map(flatMessage -> new Message(topicName, .map(flatMessage -> new Message(topicName,
((RocketMQProducerConfig) this.mqProperties).getTag(), ((RocketMQProducerConfig) this.mqProperties).getTag(),
JSON.toJSONBytes(flatMessage, JSONWriter.Feature.WriteNulls, JSONWriter.Feature.LargeObject))) JSON.toJSONBytes(flatMessage,
.collect(Collectors.toList()); JSONWriter.Feature.WriteNulls,
JSONWriter.Feature.LargeObject)))
.collect(Collectors.toList());
// 批量发送 // 批量发送
sendMessage(messages, index); sendMessage(messages, index);
}); });
@@ -277,10 +281,10 @@ import java.util.stream.Collectors;
} else { } else {
final int partition = destination.getPartition() != null ? destination.getPartition() : 0; final int partition = destination.getPartition() != null ? destination.getPartition() : 0;
List<Message> messages = flatMessages.stream() List<Message> messages = flatMessages.stream()
.map(flatMessage -> new Message(topicName, .map(flatMessage -> new Message(topicName,
((RocketMQProducerConfig) this.mqProperties).getTag(), ((RocketMQProducerConfig) this.mqProperties).getTag(),
JSON.toJSONBytes(flatMessage, JSONWriter.Feature.WriteNulls,JSONWriter.Feature.LargeObject))) JSON.toJSONBytes(flatMessage, JSONWriter.Feature.WriteNulls, JSONWriter.Feature.LargeObject)))
.collect(Collectors.toList()); .collect(Collectors.toList());
// 批量发送 // 批量发送
sendMessage(messages, partition); sendMessage(messages, partition);
} }
@@ -305,7 +309,8 @@ import java.util.stream.Collectors;
} }
} }
@SuppressWarnings("deprecation") private void sendMessage(List<Message> messages, int partition) { @SuppressWarnings("deprecation")
private void sendMessage(List<Message> messages, int partition) {
if (messages.isEmpty()) { if (messages.isEmpty()) {
return; return;
} }
@@ -351,7 +356,8 @@ import java.util.stream.Collectors;
} }
} }
@Override public void stop() { @Override
public void stop() {
logger.info("## Stop RocketMQ producer##"); logger.info("## Stop RocketMQ producer##");
this.defaultMQProducer.shutdown(); this.defaultMQProducer.shutdown();
if (sendPartitionExecutor != null) { if (sendPartitionExecutor != null) {