mirror of
https://github.com/alibaba/canal.git
synced 2026-11-04 00:30:15 +08:00
Compare commits
5 Commits
canal-1.1.
...
feature/zi
| Author | SHA1 | Date | |
|---|---|---|---|
|
c6f28eb386 | ||
|
|
a28806311a | ||
|
|
05562f1dfc | ||
|
|
a98549ee0d | ||
|
|
a6d73ad16b |
@@ -36,7 +36,7 @@
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-dependencies</artifactId>
|
||||
<version>2.5.15</version>
|
||||
<version>2.5.4</version>
|
||||
<type>pom</type>
|
||||
<scope>import</scope>
|
||||
</dependency>
|
||||
@@ -48,12 +48,7 @@
|
||||
<dependency>
|
||||
<groupId>io.ebean</groupId>
|
||||
<artifactId>ebean</artifactId>
|
||||
<version>11.45.1</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.yaml</groupId>
|
||||
<artifactId>snakeyaml</artifactId>
|
||||
<version>2.0</version>
|
||||
<version>11.41.1</version>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</dependencyManagement>
|
||||
|
||||
@@ -25,9 +25,6 @@ import org.springframework.util.PropertyPlaceholderHelper;
|
||||
public class YamlUtils {
|
||||
|
||||
public static <T> T resourceYmlToObj(String resource, String prefix, Class<T> clazz) {
|
||||
if (!StringUtils.startsWithIgnoreCase(resource, "classpath:")) {
|
||||
resource = "classpath:" + resource;
|
||||
}
|
||||
ClassPathResource classPathResource = new ClassPathResource(resource);
|
||||
|
||||
String content;
|
||||
|
||||
@@ -17,7 +17,7 @@
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-dependencies</artifactId>
|
||||
<version>2.5.15</version>
|
||||
<version>2.5.4</version>
|
||||
<type>pom</type>
|
||||
<scope>import</scope>
|
||||
</dependency>
|
||||
|
||||
@@ -1,7 +1,5 @@
|
||||
#!/bin/bash
|
||||
|
||||
args=$@
|
||||
|
||||
case $(uname) in
|
||||
Linux)
|
||||
bin_abs_path=$(readlink -f $(dirname $0))
|
||||
@@ -11,5 +9,5 @@ Linux)
|
||||
;;
|
||||
esac
|
||||
|
||||
sh "$bin_abs_path"/stop.sh $args
|
||||
sh "$bin_abs_path"/startup.sh $args
|
||||
sh "$bin_abs_path"/stop.sh
|
||||
sh "$bin_abs_path"/startup.sh
|
||||
|
||||
@@ -32,27 +32,27 @@ import com.alibaba.otter.canal.connector.core.spi.ProxyCanalMsgConsumer;
|
||||
*/
|
||||
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_STANDBY_SPI_DIR = "/canal-adapter/plugin";
|
||||
private static final String CONNECTOR_SPI_DIR = "/plugin";
|
||||
private static final String CONNECTOR_STANDBY_SPI_DIR = "/canal-adapter/plugin";
|
||||
|
||||
private CanalMsgConsumer canalMsgConsumer;
|
||||
private CanalMsgConsumer canalMsgConsumer;
|
||||
|
||||
private String canalDestination; // canal实例
|
||||
private String groupId = null; // groupId
|
||||
private List<List<OuterAdapter>> canalOuterAdapters; // 外部适配器
|
||||
private CanalClientConfig canalClientConfig; // 配置
|
||||
private ExecutorService groupInnerExecutorService; // 组内工作线程池
|
||||
private volatile boolean running = false; // 是否运行中
|
||||
private Thread thread = null;
|
||||
private Thread.UncaughtExceptionHandler handler = (t, e) -> logger
|
||||
private String canalDestination; // canal实例
|
||||
private String groupId = null; // groupId
|
||||
private List<List<OuterAdapter>> canalOuterAdapters; // 外部适配器
|
||||
private CanalClientConfig canalClientConfig; // 配置
|
||||
private ExecutorService groupInnerExecutorService; // 组内工作线程池
|
||||
private volatile boolean running = false; // 是否运行中
|
||||
private Thread thread = null;
|
||||
private Thread.UncaughtExceptionHandler handler = (t, e) -> logger
|
||||
.error("parse events has an error", e);
|
||||
|
||||
private SyncSwitch syncSwitch;
|
||||
private SyncSwitch syncSwitch;
|
||||
|
||||
public AdapterProcessor(CanalClientConfig canalClientConfig, String destination, String groupId,
|
||||
List<List<OuterAdapter>> canalOuterAdapters){
|
||||
List<List<OuterAdapter>> canalOuterAdapters) {
|
||||
this.canalClientConfig = canalClientConfig;
|
||||
this.canalDestination = destination;
|
||||
this.groupId = groupId;
|
||||
@@ -63,12 +63,9 @@ public class AdapterProcessor {
|
||||
|
||||
// load connector consumer
|
||||
ExtensionLoader<CanalMsgConsumer> loader = new ExtensionLoader<>(CanalMsgConsumer.class);
|
||||
// see https://github.com/alibaba/canal/pull/5175
|
||||
String key = destination + "_" + groupId;
|
||||
canalMsgConsumer = new ProxyCanalMsgConsumer(loader.getExtension(canalClientConfig.getMode().toLowerCase(),
|
||||
key,
|
||||
CONNECTOR_SPI_DIR,
|
||||
CONNECTOR_STANDBY_SPI_DIR));
|
||||
canalMsgConsumer = new ProxyCanalMsgConsumer(loader
|
||||
.getExtension(canalClientConfig.getMode().toLowerCase(), destination, CONNECTOR_SPI_DIR,
|
||||
CONNECTOR_STANDBY_SPI_DIR));
|
||||
|
||||
Properties properties = canalClientConfig.getConsumerProperties();
|
||||
properties.put(CanalConstants.CANAL_MQ_FLAT_MESSAGE, canalClientConfig.getFlatMessage());
|
||||
@@ -173,7 +170,7 @@ public class AdapterProcessor {
|
||||
}
|
||||
|
||||
int retry = canalClientConfig.getRetries() == null
|
||||
|| canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries();
|
||||
|| canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries();
|
||||
if (retry == -1) {
|
||||
// 重试次数-1代表异常时一直阻塞重试
|
||||
retry = Integer.MAX_VALUE;
|
||||
|
||||
@@ -117,7 +117,7 @@
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot</artifactId>
|
||||
<version>2.5.15</version>
|
||||
<version>2.5.4</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>com.h2database</groupId>
|
||||
@@ -238,7 +238,7 @@
|
||||
<dependency>
|
||||
<groupId>org.yaml</groupId>
|
||||
<artifactId>snakeyaml</artifactId>
|
||||
<version>2.0</version>
|
||||
<version>1.29</version>
|
||||
</dependency>
|
||||
<!-- test -->
|
||||
<dependency>
|
||||
|
||||
@@ -90,6 +90,7 @@
|
||||
<dependency>
|
||||
<groupId>com.alibaba</groupId>
|
||||
<artifactId>druid</artifactId>
|
||||
<optional>true</optional>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework</groupId>
|
||||
|
||||
@@ -62,17 +62,8 @@ public class PulsarMQConstants {
|
||||
public static final String PULSARMQ_ADMIN_SERVER_URL = ROOT + "." + "adminServerUrl";
|
||||
|
||||
/**
|
||||
* Pulsar 监听器名字
|
||||
* Pulsar admin服务器地址
|
||||
*/
|
||||
public static final String PULSARMQ_LISTENER_NAME = ROOT + "." + "listenerName";
|
||||
|
||||
/**
|
||||
* Pulsar 开启chunking
|
||||
*/
|
||||
public static final String PULSARMQ_ENABLE_CHUNKING = ROOT + "." + "enableChunking";
|
||||
|
||||
/**
|
||||
* Pulsar 压缩算法
|
||||
*/
|
||||
public static final String PULSARMQ_COMPRESSION_TYPE = ROOT + "." + "compressionType";
|
||||
}
|
||||
|
||||
@@ -18,7 +18,7 @@ public class PulsarMQProducerConfig extends MQProperties {
|
||||
* httpUrl: http://localhost:8080
|
||||
* </p>
|
||||
*/
|
||||
private String serverUrl;
|
||||
private String serverUrl;
|
||||
/**
|
||||
* pulsar topic前缀
|
||||
* <p>
|
||||
@@ -27,30 +27,20 @@ public class PulsarMQProducerConfig extends MQProperties {
|
||||
* 在发送消息时会自动拼接上
|
||||
* </p>
|
||||
*/
|
||||
private String topicTenantPrefix;
|
||||
private String topicTenantPrefix;
|
||||
/**
|
||||
* 生产者角色权限，请确保该角色有canal使用的所有topic生产者权限（最低要求）
|
||||
*/
|
||||
private String roleToken;
|
||||
private String roleToken;
|
||||
/**
|
||||
* admin服务器地址
|
||||
*/
|
||||
private String adminServerUrl;
|
||||
private String adminServerUrl;
|
||||
|
||||
/**
|
||||
* listener name
|
||||
*/
|
||||
private String listenerName;
|
||||
|
||||
/**
|
||||
* enableChunking
|
||||
*/
|
||||
private boolean enableChunking;
|
||||
|
||||
/**
|
||||
* compressionType
|
||||
*/
|
||||
private String compressionType;
|
||||
private String listenerName;
|
||||
|
||||
public String getServerUrl() {
|
||||
return serverUrl;
|
||||
@@ -91,20 +81,4 @@ public class PulsarMQProducerConfig extends MQProperties {
|
||||
public void setListenerName(String listenerName) {
|
||||
this.listenerName = listenerName;
|
||||
}
|
||||
|
||||
public void setEnableChunking(boolean enableChunking) {
|
||||
this.enableChunking = enableChunking;
|
||||
}
|
||||
|
||||
public boolean getEnableChunking() {
|
||||
return this.enableChunking;
|
||||
}
|
||||
|
||||
public void setCompressionType(String compressionType) {
|
||||
this.compressionType = compressionType;
|
||||
}
|
||||
|
||||
public String getCompressionType() {
|
||||
return this.compressionType;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -137,16 +137,6 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
|
||||
tmpProperties.setListenerName(listenerName);
|
||||
}
|
||||
|
||||
String enableChunkingStr = PropertiesUtils.getProperty(properties, PulsarMQConstants.PULSARMQ_ENABLE_CHUNKING);
|
||||
if (!StringUtils.isEmpty(enableChunkingStr)) {
|
||||
tmpProperties.setEnableChunking(Boolean.parseBoolean(enableChunkingStr));
|
||||
}
|
||||
|
||||
String compressionType = PropertiesUtils.getProperty(properties, PulsarMQConstants.PULSARMQ_COMPRESSION_TYPE);
|
||||
if (!StringUtils.isEmpty(compressionType)) {
|
||||
tmpProperties.setCompressionType(compressionType);
|
||||
}
|
||||
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("Load pulsar properties ==> {}", JSON.toJSON(this.mqProperties));
|
||||
}
|
||||
@@ -418,34 +408,11 @@ public class CanalPulsarMQProducer extends AbstractMQProducer implements CanalMQ
|
||||
}
|
||||
|
||||
// 创建指定topic的生产者
|
||||
ProducerBuilder producerBuilder = client.newProducer();
|
||||
if (pulsarMQProperties.getEnableChunking()) {
|
||||
producerBuilder.enableChunking(true);
|
||||
producerBuilder.enableBatching(false);
|
||||
}
|
||||
|
||||
if (!StringUtils.isEmpty(pulsarMQProperties.getCompressionType())) {
|
||||
switch (pulsarMQProperties.getCompressionType().toLowerCase()) {
|
||||
case "lz4":
|
||||
producerBuilder.compressionType(CompressionType.LZ4);
|
||||
break;
|
||||
case "zlib":
|
||||
producerBuilder.compressionType(CompressionType.ZLIB);
|
||||
break;
|
||||
case "zstd":
|
||||
producerBuilder.compressionType(CompressionType.ZSTD);
|
||||
break;
|
||||
case "snappy":
|
||||
producerBuilder.compressionType(CompressionType.SNAPPY);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
producer = producerBuilder.topic(fullTopic)
|
||||
producer = client.newProducer()
|
||||
.topic(fullTopic)
|
||||
// 指定路由器
|
||||
.messageRouter(new MessageRouterImpl(topic))
|
||||
.create();
|
||||
|
||||
// 放入缓存
|
||||
PRODUCERS.put(topic, producer);
|
||||
}
|
||||
|
||||
@@ -1,29 +1,5 @@
|
||||
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.JSONWriter;
|
||||
import com.alibaba.otter.canal.common.CanalException;
|
||||
@@ -40,6 +16,29 @@ 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.RocketMQProducerConfig;
|
||||
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 实现
|
||||
@@ -47,18 +46,15 @@ import com.alibaba.otter.canal.protocol.FlatMessage;
|
||||
* @author rewerma 2020-01-27
|
||||
* @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 static final String CLOUD_ACCESS_CHANNEL = "cloud";
|
||||
private static final String NAMESPACE_SEPARATOR = "%";
|
||||
protected ThreadPoolExecutor sendPartitionExecutor;
|
||||
private DefaultMQProducer defaultMQProducer;
|
||||
private static final String CLOUD_ACCESS_CHANNEL = "cloud";
|
||||
protected ThreadPoolExecutor sendPartitionExecutor;
|
||||
|
||||
@Override
|
||||
public void init(Properties properties) {
|
||||
@Override public void init(Properties properties) {
|
||||
RocketMQProducerConfig rocketMQProperties = new RocketMQProducerConfig();
|
||||
this.mqProperties = rocketMQProperties;
|
||||
super.init(properties);
|
||||
@@ -73,9 +69,9 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
|
||||
}
|
||||
|
||||
defaultMQProducer = new DefaultMQProducer(rocketMQProperties.getProducerGroup(),
|
||||
rpcHook,
|
||||
rocketMQProperties.isEnableMessageTrace(),
|
||||
rocketMQProperties.getCustomizedTraceTopic());
|
||||
rpcHook,
|
||||
rocketMQProperties.isEnableMessageTrace(),
|
||||
rocketMQProperties.getCustomizedTraceTopic());
|
||||
if (CLOUD_ACCESS_CHANNEL.equals(rocketMQProperties.getAccessChannel())) {
|
||||
defaultMQProducer.setAccessChannel(AccessChannel.CLOUD);
|
||||
}
|
||||
@@ -94,12 +90,12 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
|
||||
|
||||
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());
|
||||
}
|
||||
|
||||
private void loadRocketMQProperties(Properties properties) {
|
||||
@@ -115,12 +111,12 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
|
||||
rocketMQProperties.setProducerGroup(producerGroup);
|
||||
}
|
||||
String enableMessageTrace = PropertiesUtils.getProperty(properties,
|
||||
RocketMQConstants.ROCKETMQ_ENABLE_MESSAGE_TRACE);
|
||||
RocketMQConstants.ROCKETMQ_ENABLE_MESSAGE_TRACE);
|
||||
if (!StringUtils.isEmpty(enableMessageTrace)) {
|
||||
rocketMQProperties.setEnableMessageTrace(Boolean.parseBoolean(enableMessageTrace));
|
||||
}
|
||||
String customizedTraceTopic = PropertiesUtils.getProperty(properties,
|
||||
RocketMQConstants.ROCKETMQ_CUSTOMIZED_TRACE_TOPIC);
|
||||
RocketMQConstants.ROCKETMQ_CUSTOMIZED_TRACE_TOPIC);
|
||||
if (!StringUtils.isEmpty(customizedTraceTopic)) {
|
||||
rocketMQProperties.setCustomizedTraceTopic(customizedTraceTopic);
|
||||
}
|
||||
@@ -137,7 +133,7 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
|
||||
rocketMQProperties.setRetryTimesWhenSendFailed(Integer.parseInt(retry));
|
||||
}
|
||||
String vipChannelEnabled = PropertiesUtils.getProperty(properties,
|
||||
RocketMQConstants.ROCKETMQ_VIP_CHANNEL_ENABLED);
|
||||
RocketMQConstants.ROCKETMQ_VIP_CHANNEL_ENABLED);
|
||||
if (!StringUtils.isEmpty(vipChannelEnabled)) {
|
||||
rocketMQProperties.setVipChannelEnabled(Boolean.parseBoolean(vipChannelEnabled));
|
||||
}
|
||||
@@ -147,14 +143,15 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void send(MQDestination destination, com.alibaba.otter.canal.protocol.Message message, Callback callback) {
|
||||
@Override public void send(MQDestination destination, com.alibaba.otter.canal.protocol.Message message,
|
||||
Callback callback) {
|
||||
ExecutorTemplate template = new ExecutorTemplate(sendExecutor);
|
||||
try {
|
||||
if (!StringUtils.isEmpty(destination.getDynamicTopic())) {
|
||||
// 动态topic
|
||||
Map<String, com.alibaba.otter.canal.protocol.Message> messageMap = MQMessageUtils
|
||||
.messageTopics(message, destination.getTopic(), destination.getDynamicTopic());
|
||||
Map<String, com.alibaba.otter.canal.protocol.Message> messageMap = MQMessageUtils.messageTopics(message,
|
||||
destination.getTopic(),
|
||||
destination.getDynamicTopic());
|
||||
|
||||
for (Map.Entry<String, com.alibaba.otter.canal.protocol.Message> entry : messageMap.entrySet()) {
|
||||
String topicName = entry.getKey().replace('.', '_');
|
||||
@@ -186,7 +183,7 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
|
||||
com.alibaba.otter.canal.protocol.Message message) {
|
||||
// 获取当前topic的分区数
|
||||
Integer partitionNum = MQMessageUtils.parseDynamicTopicPartition(topicName,
|
||||
destination.getDynamicTopicPartitionNum());
|
||||
destination.getDynamicTopicPartitionNum());
|
||||
|
||||
// 获取topic的队列数为分区数
|
||||
if (partitionNum == null) {
|
||||
@@ -202,10 +199,10 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
|
||||
MQMessageUtils.EntryRowData[] datas = MQMessageUtils.buildMessageData(message, buildExecutor);
|
||||
// 串行分区
|
||||
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 length = messages.length;
|
||||
|
||||
ExecutorTemplate template = new ExecutorTemplate(sendPartitionExecutor);
|
||||
@@ -215,9 +212,9 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
|
||||
final int index = i;
|
||||
template.submit(() -> {
|
||||
Message data = new Message(topicName,
|
||||
((RocketMQProducerConfig) this.mqProperties).getTag(),
|
||||
CanalMessageSerializerUtil.serializer(dataPartition,
|
||||
mqProperties.isFilterTransactionEntry()));
|
||||
((RocketMQProducerConfig) this.mqProperties).getTag(),
|
||||
CanalMessageSerializerUtil.serializer(dataPartition,
|
||||
mqProperties.isFilterTransactionEntry()));
|
||||
sendMessage(data, index);
|
||||
});
|
||||
}
|
||||
@@ -227,8 +224,8 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
|
||||
} else {
|
||||
final int partition = destination.getPartition() != null ? destination.getPartition() : 0;
|
||||
Message data = new Message(topicName,
|
||||
((RocketMQProducerConfig) this.mqProperties).getTag(),
|
||||
CanalMessageSerializerUtil.serializer(message, mqProperties.isFilterTransactionEntry()));
|
||||
((RocketMQProducerConfig) this.mqProperties).getTag(),
|
||||
CanalMessageSerializerUtil.serializer(message, mqProperties.isFilterTransactionEntry()));
|
||||
sendMessage(data, partition);
|
||||
}
|
||||
} else {
|
||||
@@ -245,9 +242,9 @@ public class CanalRocketMQProducer 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
|
||||
@@ -264,12 +261,10 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
|
||||
final int index = i;
|
||||
template.submit(() -> {
|
||||
List<Message> messages = flatMessagePart.stream()
|
||||
.map(flatMessage -> new Message(topicName,
|
||||
((RocketMQProducerConfig) this.mqProperties).getTag(),
|
||||
JSON.toJSONBytes(flatMessage,
|
||||
JSONWriter.Feature.WriteNulls,
|
||||
JSONWriter.Feature.LargeObject)))
|
||||
.collect(Collectors.toList());
|
||||
.map(flatMessage -> new Message(topicName,
|
||||
((RocketMQProducerConfig) this.mqProperties).getTag(),
|
||||
JSON.toJSONBytes(flatMessage, JSONWriter.Feature.WriteNulls, JSONWriter.Feature.LargeObject)))
|
||||
.collect(Collectors.toList());
|
||||
// 批量发送
|
||||
sendMessage(messages, index);
|
||||
});
|
||||
@@ -281,10 +276,10 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
|
||||
} else {
|
||||
final int partition = destination.getPartition() != null ? destination.getPartition() : 0;
|
||||
List<Message> messages = flatMessages.stream()
|
||||
.map(flatMessage -> new Message(topicName,
|
||||
((RocketMQProducerConfig) this.mqProperties).getTag(),
|
||||
JSON.toJSONBytes(flatMessage, JSONWriter.Feature.WriteNulls, JSONWriter.Feature.LargeObject)))
|
||||
.collect(Collectors.toList());
|
||||
.map(flatMessage -> new Message(topicName,
|
||||
((RocketMQProducerConfig) this.mqProperties).getTag(),
|
||||
JSON.toJSONBytes(flatMessage, JSONWriter.Feature.WriteNulls,JSONWriter.Feature.LargeObject)))
|
||||
.collect(Collectors.toList());
|
||||
// 批量发送
|
||||
sendMessage(messages, partition);
|
||||
}
|
||||
@@ -309,19 +304,14 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
|
||||
}
|
||||
}
|
||||
|
||||
@SuppressWarnings("deprecation")
|
||||
private void sendMessage(List<Message> messages, int partition) {
|
||||
@SuppressWarnings("deprecation") private void sendMessage(List<Message> messages, int partition) {
|
||||
if (messages.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
|
||||
// 获取一下messageQueue
|
||||
DefaultMQProducerImpl innerProducer = this.defaultMQProducer.getDefaultMQProducerImpl();
|
||||
String topic = messages.get(0).getTopic();
|
||||
if (StringUtils.isNotBlank(this.defaultMQProducer.getNamespace())) {
|
||||
topic = this.defaultMQProducer.getNamespace() + NAMESPACE_SEPARATOR + topic;
|
||||
}
|
||||
TopicPublishInfo topicInfo = innerProducer.getTopicPublishInfoTable().get(topic);
|
||||
TopicPublishInfo topicInfo = innerProducer.getTopicPublishInfoTable().get(messages.get(0).getTopic());
|
||||
if (topicInfo == null) {
|
||||
for (Message message : messages) {
|
||||
sendMessage(message, partition);
|
||||
@@ -356,8 +346,7 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void stop() {
|
||||
@Override public void stop() {
|
||||
logger.info("## Stop RocketMQ producer##");
|
||||
this.defaultMQProducer.shutdown();
|
||||
if (sendPartitionExecutor != null) {
|
||||
|
||||
@@ -819,7 +819,7 @@ public class QueryLogEvent extends LogEvent {
|
||||
case Q_MASTER_DATA_WRITTEN_CODE:
|
||||
return "Q_MASTER_DATA_WRITTEN_CODE";
|
||||
case Q_INVOKER:
|
||||
return "Q_INVOKER";
|
||||
return "case Q_INVOKER";
|
||||
case Q_MICROSECONDS:
|
||||
return "Q_MICROSECONDS";
|
||||
case Q_UPDATED_DB_NAMES:
|
||||
|
||||
@@ -5,11 +5,6 @@
|
||||
# enable gtid use true/false
|
||||
canal.instance.gtidon=false
|
||||
|
||||
# rds oss binlog
|
||||
canal.instance.rds.accesskey=
|
||||
canal.instance.rds.secretkey=
|
||||
canal.instance.rds.instanceId=
|
||||
|
||||
# position info
|
||||
canal.instance.master.address=127.0.0.1:3306
|
||||
canal.instance.master.journal.name=
|
||||
@@ -17,15 +12,10 @@ canal.instance.master.position=
|
||||
canal.instance.master.timestamp=
|
||||
canal.instance.master.gtid=
|
||||
|
||||
# ssl
|
||||
#canal.instance.master.sslMode=DISABLED
|
||||
#canal.instance.master.tlsVersions=
|
||||
#canal.instance.master.trustCertificateKeyStoreType=
|
||||
#canal.instance.master.trustCertificateKeyStoreUrl=
|
||||
#canal.instance.master.trustCertificateKeyStorePassword=
|
||||
#canal.instance.master.clientCertificateKeyStoreType=
|
||||
#canal.instance.master.clientCertificateKeyStoreUrl=
|
||||
#canal.instance.master.clientCertificateKeyStorePassword=
|
||||
# rds oss binlog
|
||||
canal.instance.rds.accesskey=
|
||||
canal.instance.rds.secretkey=
|
||||
canal.instance.rds.instanceId=
|
||||
|
||||
# table meta tsdb info
|
||||
canal.instance.tsdb.enable=true
|
||||
|
||||
@@ -36,4 +36,4 @@
|
||||
<property name="secretkey" value="${canal.aliyun.secretkey:}" />
|
||||
<property name="instanceId" value="${canal.instance.rds.instanceId:}" />
|
||||
</bean>
|
||||
</beans>
|
||||
</beans>
|
||||
@@ -111,6 +111,7 @@
|
||||
<property name="defaultConnectionTimeoutInSeconds" value="${canal.instance.network.soTimeout:30}" />
|
||||
|
||||
<!-- 解析编码 -->
|
||||
<!-- property name="connectionCharsetNumber" value="${canal.instance.connectionCharsetNumber:33}" /-->
|
||||
<property name="connectionCharset" value="${canal.instance.connectionCharset:UTF-8}" />
|
||||
|
||||
<!-- 解析位点记录 -->
|
||||
@@ -139,18 +140,6 @@
|
||||
<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
|
||||
<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
|
||||
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
|
||||
<property name="sslInfo">
|
||||
<bean class="com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo">
|
||||
<property name="sslMode" value="${canal.instance.master.sslMode:DISABLED}"/>
|
||||
<property name="tlsVersions" value="${canal.instance.master.tlsVersions:}"/>
|
||||
<property name="trustCertificateKeyStoreType" value="${canal.instance.master.trustCertificateKeyStoreType:}"/>
|
||||
<property name="trustCertificateKeyStoreUrl" value="${canal.instance.master.trustCertificateKeyStoreUrl:}"/>
|
||||
<property name="trustCertificateKeyStorePassword" value="${canal.instance.master.trustCertificateKeyStorePassword:}"/>
|
||||
<property name="clientCertificateKeyStoreType" value="${canal.instance.master.clientCertificateKeyStoreType:}"/>
|
||||
<property name="clientCertificateKeyStoreUrl" value="${canal.instance.master.clientCertificateKeyStoreUrl:}"/>
|
||||
<property name="clientCertificateKeyStorePassword" value="${canal.instance.master.clientCertificateKeyStorePassword:}"/>
|
||||
</bean>
|
||||
</property>
|
||||
</bean>
|
||||
</property>
|
||||
<property name="standbyInfo">
|
||||
@@ -161,18 +150,6 @@
|
||||
<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
|
||||
<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
|
||||
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
|
||||
<property name="sslInfo">
|
||||
<bean class="com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo">
|
||||
<property name="sslMode" value="${canal.instance.standby.sslMode:DISABLED}"/>
|
||||
<property name="tlsVersions" value="${canal.instance.standby.tlsVersions:}"/>
|
||||
<property name="trustCertificateKeyStoreType" value="${canal.instance.standby.trustCertificateKeyStoreType:}"/>
|
||||
<property name="trustCertificateKeyStoreUrl" value="${canal.instance.standby.trustCertificateKeyStoreUrl:}"/>
|
||||
<property name="trustCertificateKeyStorePassword" value="${canal.instance.standby.trustCertificateKeyStorePassword:}"/>
|
||||
<property name="clientCertificateKeyStoreType" value="${canal.instance.standby.clientCertificateKeyStoreType:}"/>
|
||||
<property name="clientCertificateKeyStoreUrl" value="${canal.instance.standby.clientCertificateKeyStoreUrl:}"/>
|
||||
<property name="clientCertificateKeyStorePassword" value="${canal.instance.standby.clientCertificateKeyStorePassword:}"/>
|
||||
</bean>
|
||||
</property>
|
||||
</bean>
|
||||
</property>
|
||||
|
||||
|
||||
@@ -97,6 +97,7 @@
|
||||
<property name="defaultConnectionTimeoutInSeconds" value="${canal.instance.network.soTimeout:30}" />
|
||||
|
||||
<!-- 解析编码 -->
|
||||
<!-- property name="connectionCharsetNumber" value="${canal.instance.connectionCharsetNumber:33}" /-->
|
||||
<property name="connectionCharset" value="${canal.instance.connectionCharset:UTF-8}" />
|
||||
|
||||
<!-- 解析位点记录 -->
|
||||
@@ -125,18 +126,6 @@
|
||||
<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
|
||||
<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
|
||||
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
|
||||
<property name="sslInfo">
|
||||
<bean class="com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo">
|
||||
<property name="sslMode" value="${canal.instance.master.sslMode:DISABLED}"/>
|
||||
<property name="tlsVersions" value="${canal.instance.master.tlsVersions:}"/>
|
||||
<property name="trustCertificateKeyStoreType" value="${canal.instance.master.trustCertificateKeyStoreType:}"/>
|
||||
<property name="trustCertificateKeyStoreUrl" value="${canal.instance.master.trustCertificateKeyStoreUrl:}"/>
|
||||
<property name="trustCertificateKeyStorePassword" value="${canal.instance.master.trustCertificateKeyStorePassword:}"/>
|
||||
<property name="clientCertificateKeyStoreType" value="${canal.instance.master.clientCertificateKeyStoreType:}"/>
|
||||
<property name="clientCertificateKeyStoreUrl" value="${canal.instance.master.clientCertificateKeyStoreUrl:}"/>
|
||||
<property name="clientCertificateKeyStorePassword" value="${canal.instance.master.clientCertificateKeyStorePassword:}"/>
|
||||
</bean>
|
||||
</property>
|
||||
</bean>
|
||||
</property>
|
||||
<property name="standbyInfo">
|
||||
@@ -147,18 +136,6 @@
|
||||
<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
|
||||
<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
|
||||
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
|
||||
<property name="sslInfo">
|
||||
<bean class="com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo">
|
||||
<property name="sslMode" value="${canal.instance.standby.sslMode:DISABLED}"/>
|
||||
<property name="tlsVersions" value="${canal.instance.standby.tlsVersions:}"/>
|
||||
<property name="trustCertificateKeyStoreType" value="${canal.instance.standby.trustCertificateKeyStoreType:}"/>
|
||||
<property name="trustCertificateKeyStoreUrl" value="${canal.instance.standby.trustCertificateKeyStoreUrl:}"/>
|
||||
<property name="trustCertificateKeyStorePassword" value="${canal.instance.standby.trustCertificateKeyStorePassword:}"/>
|
||||
<property name="clientCertificateKeyStoreType" value="${canal.instance.standby.clientCertificateKeyStoreType:}"/>
|
||||
<property name="clientCertificateKeyStoreUrl" value="${canal.instance.standby.clientCertificateKeyStoreUrl:}"/>
|
||||
<property name="clientCertificateKeyStorePassword" value="${canal.instance.standby.clientCertificateKeyStorePassword:}"/>
|
||||
</bean>
|
||||
</property>
|
||||
</bean>
|
||||
</property>
|
||||
|
||||
|
||||
@@ -103,6 +103,7 @@
|
||||
<property name="defaultConnectionTimeoutInSeconds" value="${canal.instance.network.soTimeout:30}" />
|
||||
|
||||
<!-- 解析编码 -->
|
||||
<!-- property name="connectionCharsetNumber" value="${canal.instance.connectionCharsetNumber:33}" /-->
|
||||
<property name="connectionCharset" value="${canal.instance.connectionCharset:UTF-8}" />
|
||||
|
||||
<!-- 解析位点记录 -->
|
||||
@@ -122,18 +123,6 @@
|
||||
<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
|
||||
<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
|
||||
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
|
||||
<property name="sslInfo">
|
||||
<bean class="com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo">
|
||||
<property name="sslMode" value="${canal.instance.master.sslMode:DISABLED}"/>
|
||||
<property name="tlsVersions" value="${canal.instance.master.tlsVersions:}"/>
|
||||
<property name="trustCertificateKeyStoreType" value="${canal.instance.master.trustCertificateKeyStoreType:}"/>
|
||||
<property name="trustCertificateKeyStoreUrl" value="${canal.instance.master.trustCertificateKeyStoreUrl:}"/>
|
||||
<property name="trustCertificateKeyStorePassword" value="${canal.instance.master.trustCertificateKeyStorePassword:}"/>
|
||||
<property name="clientCertificateKeyStoreType" value="${canal.instance.master.clientCertificateKeyStoreType:}"/>
|
||||
<property name="clientCertificateKeyStoreUrl" value="${canal.instance.master.clientCertificateKeyStoreUrl:}"/>
|
||||
<property name="clientCertificateKeyStorePassword" value="${canal.instance.master.clientCertificateKeyStorePassword:}"/>
|
||||
</bean>
|
||||
</property>
|
||||
</bean>
|
||||
</property>
|
||||
<property name="standbyInfo">
|
||||
@@ -144,18 +133,6 @@
|
||||
<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
|
||||
<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
|
||||
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
|
||||
<property name="sslInfo">
|
||||
<bean class="com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo">
|
||||
<property name="sslMode" value="${canal.instance.standby.sslMode:DISABLED}"/>
|
||||
<property name="tlsVersions" value="${canal.instance.standby.tlsVersions:}"/>
|
||||
<property name="trustCertificateKeyStoreType" value="${canal.instance.standby.trustCertificateKeyStoreType:}"/>
|
||||
<property name="trustCertificateKeyStoreUrl" value="${canal.instance.standby.trustCertificateKeyStoreUrl:}"/>
|
||||
<property name="trustCertificateKeyStorePassword" value="${canal.instance.standby.trustCertificateKeyStorePassword:}"/>
|
||||
<property name="clientCertificateKeyStoreType" value="${canal.instance.standby.clientCertificateKeyStoreType:}"/>
|
||||
<property name="clientCertificateKeyStoreUrl" value="${canal.instance.standby.clientCertificateKeyStoreUrl:}"/>
|
||||
<property name="clientCertificateKeyStorePassword" value="${canal.instance.standby.clientCertificateKeyStorePassword:}"/>
|
||||
</bean>
|
||||
</property>
|
||||
</bean>
|
||||
</property>
|
||||
|
||||
@@ -234,6 +211,7 @@
|
||||
<property name="defaultConnectionTimeoutInSeconds" value="${canal.instance.network.soTimeout:30}" />
|
||||
|
||||
<!-- 解析编码 -->
|
||||
<!-- property name="connectionCharsetNumber" value="${canal.instance.connectionCharsetNumber:33}" /-->
|
||||
<property name="connectionCharset" value="${canal.instance.connectionCharset:UTF-8}" />
|
||||
|
||||
<!-- 解析位点记录 -->
|
||||
@@ -253,18 +231,6 @@
|
||||
<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
|
||||
<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
|
||||
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
|
||||
<property name="sslInfo">
|
||||
<bean class="com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo">
|
||||
<property name="sslMode" value="${canal.instance.master.sslMode:DISABLED}"/>
|
||||
<property name="tlsVersions" value="${canal.instance.master.tlsVersions:}"/>
|
||||
<property name="trustCertificateKeyStoreType" value="${canal.instance.master.trustCertificateKeyStoreType:}"/>
|
||||
<property name="trustCertificateKeyStoreUrl" value="${canal.instance.master.trustCertificateKeyStoreUrl:}"/>
|
||||
<property name="trustCertificateKeyStorePassword" value="${canal.instance.master.trustCertificateKeyStorePassword:}"/>
|
||||
<property name="clientCertificateKeyStoreType" value="${canal.instance.master.clientCertificateKeyStoreType:}"/>
|
||||
<property name="clientCertificateKeyStoreUrl" value="${canal.instance.master.clientCertificateKeyStoreUrl:}"/>
|
||||
<property name="clientCertificateKeyStorePassword" value="${canal.instance.master.clientCertificateKeyStorePassword:}"/>
|
||||
</bean>
|
||||
</property>
|
||||
</bean>
|
||||
</property>
|
||||
<property name="standbyInfo">
|
||||
@@ -275,18 +241,6 @@
|
||||
<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
|
||||
<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
|
||||
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
|
||||
<property name="sslInfo">
|
||||
<bean class="com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo">
|
||||
<property name="sslMode" value="${canal.instance.standby.sslMode:DISABLED}"/>
|
||||
<property name="tlsVersions" value="${canal.instance.standby.tlsVersions:}"/>
|
||||
<property name="trustCertificateKeyStoreType" value="${canal.instance.standby.trustCertificateKeyStoreType:}"/>
|
||||
<property name="trustCertificateKeyStoreUrl" value="${canal.instance.standby.trustCertificateKeyStoreUrl:}"/>
|
||||
<property name="trustCertificateKeyStorePassword" value="${canal.instance.standby.trustCertificateKeyStorePassword:}"/>
|
||||
<property name="clientCertificateKeyStoreType" value="${canal.instance.standby.clientCertificateKeyStoreType:}"/>
|
||||
<property name="clientCertificateKeyStoreUrl" value="${canal.instance.standby.clientCertificateKeyStoreUrl:}"/>
|
||||
<property name="clientCertificateKeyStorePassword" value="${canal.instance.standby.clientCertificateKeyStorePassword:}"/>
|
||||
</bean>
|
||||
</property>
|
||||
</bean>
|
||||
</property>
|
||||
|
||||
|
||||
@@ -94,6 +94,7 @@
|
||||
<property name="defaultConnectionTimeoutInSeconds" value="${canal.instance.network.soTimeout:30}" />
|
||||
|
||||
<!-- 解析编码 -->
|
||||
<!-- property name="connectionCharsetNumber" value="${canal.instance.connectionCharsetNumber:33}" /-->
|
||||
<property name="connectionCharset" value="${canal.instance.connectionCharset:UTF-8}" />
|
||||
|
||||
<!-- 解析位点记录 -->
|
||||
@@ -113,18 +114,6 @@
|
||||
<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
|
||||
<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
|
||||
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
|
||||
<property name="sslInfo">
|
||||
<bean class="com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo">
|
||||
<property name="sslMode" value="${canal.instance.master.sslMode:DISABLED}"/>
|
||||
<property name="tlsVersions" value="${canal.instance.master.tlsVersions:}"/>
|
||||
<property name="trustCertificateKeyStoreType" value="${canal.instance.master.trustCertificateKeyStoreType:}"/>
|
||||
<property name="trustCertificateKeyStoreUrl" value="${canal.instance.master.trustCertificateKeyStoreUrl:}"/>
|
||||
<property name="trustCertificateKeyStorePassword" value="${canal.instance.master.trustCertificateKeyStorePassword:}"/>
|
||||
<property name="clientCertificateKeyStoreType" value="${canal.instance.master.clientCertificateKeyStoreType:}"/>
|
||||
<property name="clientCertificateKeyStoreUrl" value="${canal.instance.master.clientCertificateKeyStoreUrl:}"/>
|
||||
<property name="clientCertificateKeyStorePassword" value="${canal.instance.master.clientCertificateKeyStorePassword:}"/>
|
||||
</bean>
|
||||
</property>
|
||||
</bean>
|
||||
</property>
|
||||
<property name="standbyInfo">
|
||||
@@ -135,18 +124,6 @@
|
||||
<property name="pwdPublicKey" value="${canal.instance.pwdPublicKey:retl}" />
|
||||
<property name="enableDruid" value="${canal.instance.enableDruid:false}" />
|
||||
<property name="defaultDatabaseName" value="${canal.instance.defaultDatabaseName:}" />
|
||||
<property name="sslInfo">
|
||||
<bean class="com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo">
|
||||
<property name="sslMode" value="${canal.instance.standby.sslMode:DISABLED}"/>
|
||||
<property name="tlsVersions" value="${canal.instance.standby.tlsVersions:}"/>
|
||||
<property name="trustCertificateKeyStoreType" value="${canal.instance.standby.trustCertificateKeyStoreType:}"/>
|
||||
<property name="trustCertificateKeyStoreUrl" value="${canal.instance.standby.trustCertificateKeyStoreUrl:}"/>
|
||||
<property name="trustCertificateKeyStorePassword" value="${canal.instance.standby.trustCertificateKeyStorePassword:}"/>
|
||||
<property name="clientCertificateKeyStoreType" value="${canal.instance.standby.clientCertificateKeyStoreType:}"/>
|
||||
<property name="clientCertificateKeyStoreUrl" value="${canal.instance.standby.clientCertificateKeyStoreUrl:}"/>
|
||||
<property name="clientCertificateKeyStorePassword" value="${canal.instance.standby.clientCertificateKeyStorePassword:}"/>
|
||||
</bean>
|
||||
</property>
|
||||
</bean>
|
||||
</property>
|
||||
|
||||
|
||||
@@ -1,7 +1,5 @@
|
||||
package com.alibaba.otter.canal.parse.driver.mysql;
|
||||
|
||||
import static com.alibaba.otter.canal.parse.driver.mysql.packets.Capability.CLIENT_SSL;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.security.DigestException;
|
||||
@@ -12,12 +10,12 @@ import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket;
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.packets.client.*;
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.packets.client.AuthSwitchResponsePacket;
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.packets.client.ClientAuthenticationPacket;
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.packets.client.QuitCommandPacket;
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.packets.server.*;
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannel;
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.socket.SocketChannelPool;
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo;
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslMode;
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.utils.MSC;
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.utils.MySQLPasswordEncrypter;
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager;
|
||||
@@ -30,17 +28,18 @@ import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager;
|
||||
*/
|
||||
public class MysqlConnector {
|
||||
|
||||
public static final int timeout = 5 * 1000; // 5s
|
||||
private static final Logger logger = LoggerFactory.getLogger(MysqlConnector.class);
|
||||
private InetSocketAddress address;
|
||||
private String username;
|
||||
private String password;
|
||||
private SslInfo sslInfo;
|
||||
|
||||
private byte charsetNumber = 33;
|
||||
private String defaultSchema;
|
||||
private int soTimeout = 30 * 1000;
|
||||
private int connTimeout = 5 * 1000;
|
||||
private int receiveBufferSize = 16 * 1024;
|
||||
private int sendBufferSize = 16 * 1024;
|
||||
|
||||
private SocketChannel channel;
|
||||
private volatile boolean dumping = false;
|
||||
// mysql connectionId
|
||||
@@ -49,6 +48,8 @@ public class MysqlConnector {
|
||||
// serverVersion
|
||||
private String serverVersion;
|
||||
|
||||
public static final int timeout = 5 * 1000; // 5s
|
||||
|
||||
public MysqlConnector(){
|
||||
}
|
||||
|
||||
@@ -61,25 +62,20 @@ public class MysqlConnector {
|
||||
this.password = password;
|
||||
}
|
||||
|
||||
public MysqlConnector(InetSocketAddress address, String username, String password, String defaultSchema){
|
||||
public MysqlConnector(InetSocketAddress address, String username, String password, byte charsetNumber,
|
||||
String defaultSchema){
|
||||
this(address, username, password);
|
||||
|
||||
this.charsetNumber = charsetNumber;
|
||||
this.defaultSchema = defaultSchema;
|
||||
}
|
||||
|
||||
public MysqlConnector(InetSocketAddress address, String username, String password, String defaultSchema,
|
||||
SslInfo sslInfo){
|
||||
this(address, username, password, defaultSchema);
|
||||
this.sslInfo = sslInfo;
|
||||
}
|
||||
|
||||
public void connect() throws IOException {
|
||||
if (connected.compareAndSet(false, true)) {
|
||||
try {
|
||||
channel = SocketChannelPool.open(address);
|
||||
logger.info("connect MysqlConnection to {}...", address);
|
||||
negotiate(channel);
|
||||
printSslStatus();
|
||||
} catch (Exception e) {
|
||||
disconnect();
|
||||
throw new IOException("connect " + this.address + " failure", e);
|
||||
@@ -89,28 +85,6 @@ public class MysqlConnector {
|
||||
}
|
||||
}
|
||||
|
||||
private void printSslStatus() {
|
||||
try {
|
||||
MysqlQueryExecutor executor = new MysqlQueryExecutor(this);
|
||||
ResultSetPacket result = executor.query("SHOW STATUS LIKE 'Ssl_version'");
|
||||
String sslVersion = "";
|
||||
if (result.getFieldValues() != null && result.getFieldValues().size() >= 2) {
|
||||
sslVersion = result.getFieldValues().get(1);
|
||||
}
|
||||
result = executor.query("SHOW STATUS LIKE 'Ssl_cipher'");
|
||||
String sslCipher = "";
|
||||
if (result.getFieldValues() != null && result.getFieldValues().size() >= 2) {
|
||||
sslCipher = result.getFieldValues().get(1);
|
||||
}
|
||||
logger.info("connect MysqlConnection in sslMode {}, Ssl_version:{}, Ssl_cipher:{}",
|
||||
(sslInfo != null ? sslInfo.getSslMode() : SslMode.DISABLED),
|
||||
sslVersion,
|
||||
sslCipher);
|
||||
} catch (Exception e) {
|
||||
logger.warn("Can't show SSL status, server may not standard MySQL server", e);
|
||||
}
|
||||
}
|
||||
|
||||
public void reconnect() throws IOException {
|
||||
disconnect();
|
||||
connect();
|
||||
@@ -157,6 +131,7 @@ public class MysqlConnector {
|
||||
|
||||
public MysqlConnector fork() {
|
||||
MysqlConnector connector = new MysqlConnector();
|
||||
connector.setCharsetNumber(getCharsetNumber());
|
||||
connector.setDefaultSchema(getDefaultSchema());
|
||||
connector.setAddress(getAddress());
|
||||
connector.setPassword(password);
|
||||
@@ -165,7 +140,6 @@ public class MysqlConnector {
|
||||
connector.setSendBufferSize(getSendBufferSize());
|
||||
connector.setSoTimeout(getSoTimeout());
|
||||
connector.setConnTimeout(connTimeout);
|
||||
connector.setSslInfo(getSslInfo());
|
||||
return connector;
|
||||
}
|
||||
|
||||
@@ -196,23 +170,6 @@ public class MysqlConnector {
|
||||
}
|
||||
HandshakeInitializationPacket handshakePacket = new HandshakeInitializationPacket();
|
||||
handshakePacket.fromBytes(body);
|
||||
byte serverCharsetNumber = (handshakePacket.serverCharsetNumber != 0) ? handshakePacket.serverCharsetNumber : 33;
|
||||
SslMode sslMode = sslInfo != null ? sslInfo.getSslMode() : SslMode.DISABLED;
|
||||
if (sslMode != SslMode.DISABLED) {
|
||||
boolean serverSupportSsl = (handshakePacket.serverCapabilities & CLIENT_SSL) > 0;
|
||||
if (!serverSupportSsl) {
|
||||
throw new IOException("MySQL Server does not support SSL: " + address + " serverCapabilities: "
|
||||
+ handshakePacket.serverCapabilities);
|
||||
}
|
||||
byte[] sslPacket = new SslRequestCommandPacket(serverCharsetNumber).toBytes();
|
||||
HeaderPacket sslHeader = new HeaderPacket();
|
||||
sslHeader.setPacketBodyLength(sslPacket.length);
|
||||
sslHeader.setPacketSequenceNumber((byte) (header.getPacketSequenceNumber() + 1));
|
||||
header.setPacketSequenceNumber((byte) (header.getPacketSequenceNumber() + 1));
|
||||
PacketManager.writePkg(channel, sslHeader.toBytes(), sslPacket);
|
||||
channel = SocketChannelPool.connectSsl(channel, sslInfo);
|
||||
this.channel = channel;
|
||||
}
|
||||
if (handshakePacket.protocolVersion != MSC.DEFAULT_PROTOCOL_VERSION) {
|
||||
// HandshakeV9
|
||||
auth323(channel, (byte) (header.getPacketSequenceNumber() + 1), handshakePacket.seed);
|
||||
@@ -222,27 +179,15 @@ public class MysqlConnector {
|
||||
connectionId = handshakePacket.threadId; // 记录一下connection
|
||||
serverVersion = handshakePacket.serverVersion; // 记录serverVersion
|
||||
logger.info("handshake initialization packet received, prepare the client authentication packet to send");
|
||||
// 某些老协议的 server 默认不返回 auth plugin，需要使用默认的 mysql_native_password
|
||||
String authPluginName = (handshakePacket.authPluginName != null
|
||||
&& handshakePacket.authPluginName.length > 0) ? new String(
|
||||
handshakePacket.authPluginName) : "mysql_native_password";
|
||||
logger.info("auth plugin: {}", authPluginName);
|
||||
boolean isSha2Password = false;
|
||||
ClientAuthenticationPacket clientAuth;
|
||||
if ("caching_sha2_password".equals(authPluginName)) {
|
||||
clientAuth = new ClientAuthenticationSHA2Packet();
|
||||
isSha2Password = true;
|
||||
} else {
|
||||
clientAuth = new ClientAuthenticationPacket();
|
||||
}
|
||||
clientAuth.setCharsetNumber(serverCharsetNumber);
|
||||
ClientAuthenticationPacket clientAuth = new ClientAuthenticationPacket();
|
||||
clientAuth.setCharsetNumber(charsetNumber);
|
||||
|
||||
clientAuth.setUsername(username);
|
||||
clientAuth.setPassword(password);
|
||||
clientAuth.setServerCapabilities(handshakePacket.serverCapabilities);
|
||||
clientAuth.setDatabaseName(defaultSchema);
|
||||
clientAuth.setScrumbleBuff(joinAndCreateScrumbleBuff(handshakePacket));
|
||||
clientAuth.setAuthPluginName(authPluginName.getBytes());
|
||||
clientAuth.setAuthPluginName("mysql_native_password".getBytes());
|
||||
|
||||
byte[] clientAuthPkgBody = clientAuth.toBytes();
|
||||
HeaderPacket h = new HeaderPacket();
|
||||
@@ -260,64 +205,57 @@ public class MysqlConnector {
|
||||
assert body != null;
|
||||
byte marker = body[0];
|
||||
if (marker == -2 || marker == 1) {
|
||||
if (isSha2Password && body[1] == 3) {
|
||||
// sha2 auth ok
|
||||
logger.info("caching_sha2_password auth success.");
|
||||
header = PacketManager.readHeader(channel, 4);
|
||||
body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout);
|
||||
byte[] authData = null;
|
||||
String pluginName = null;
|
||||
if (marker == 1) {
|
||||
AuthSwitchRequestMoreData packet = new AuthSwitchRequestMoreData();
|
||||
packet.fromBytes(body);
|
||||
authData = packet.authData;
|
||||
} else {
|
||||
byte[] authData = null;
|
||||
String pluginName = null;
|
||||
if (marker == 1) {
|
||||
AuthSwitchRequestMoreData packet = new AuthSwitchRequestMoreData();
|
||||
packet.fromBytes(body);
|
||||
authData = packet.authData;
|
||||
} else {
|
||||
AuthSwitchRequestPacket packet = new AuthSwitchRequestPacket();
|
||||
packet.fromBytes(body);
|
||||
authData = packet.authData;
|
||||
pluginName = packet.authName;
|
||||
logger.info("auth switch pluginName is {}.", pluginName);
|
||||
}
|
||||
AuthSwitchRequestPacket packet = new AuthSwitchRequestPacket();
|
||||
packet.fromBytes(body);
|
||||
authData = packet.authData;
|
||||
pluginName = packet.authName;
|
||||
logger.info("auth switch pluginName is {}.", pluginName);
|
||||
}
|
||||
|
||||
byte[] encryptedPassword = null;
|
||||
if ("mysql_clear_password".equals(pluginName)) {
|
||||
encryptedPassword = getPassword().getBytes();
|
||||
header = authSwitchAfterAuth(encryptedPassword, header);
|
||||
byte[] encryptedPassword = null;
|
||||
if ("mysql_clear_password".equals(pluginName)) {
|
||||
encryptedPassword = getPassword().getBytes();
|
||||
header = authSwitchAfterAuth(encryptedPassword, header);
|
||||
body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout);
|
||||
} else if ("mysql_native_password".equals(pluginName)) {
|
||||
try {
|
||||
encryptedPassword = MySQLPasswordEncrypter.scramble411(getPassword().getBytes(), authData);
|
||||
} catch (NoSuchAlgorithmException e) {
|
||||
throw new RuntimeException("can't encrypt password that will be sent to MySQL server.", e);
|
||||
}
|
||||
header = authSwitchAfterAuth(encryptedPassword, header);
|
||||
body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout);
|
||||
} else if ("caching_sha2_password".equals(pluginName)) {
|
||||
byte[] scramble = authData;
|
||||
try {
|
||||
encryptedPassword = MySQLPasswordEncrypter.scrambleCachingSha2(getPassword().getBytes(), scramble);
|
||||
} catch (DigestException e) {
|
||||
throw new RuntimeException("can't encrypt password that will be sent to MySQL server.", e);
|
||||
}
|
||||
header = authSwitchAfterAuth(encryptedPassword, header);
|
||||
body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout);
|
||||
assert body != null;
|
||||
if (body[0] == 0x01 && body[1] == 0x04) {
|
||||
// fixed issue https://github.com/alibaba/canal/pull/4767, support mysql 8.0.30+
|
||||
header = cachingSha2PasswordFullAuth(channel, header, getPassword().getBytes(), scramble);
|
||||
body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout);
|
||||
} else if (pluginName == null || "mysql_native_password".equals(pluginName)) {
|
||||
try {
|
||||
encryptedPassword = MySQLPasswordEncrypter.scramble411(getPassword().getBytes(), authData);
|
||||
} catch (NoSuchAlgorithmException e) {
|
||||
throw new RuntimeException("can't encrypt password that will be sent to MySQL server.", e);
|
||||
}
|
||||
header = authSwitchAfterAuth(encryptedPassword, header);
|
||||
body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout);
|
||||
} else if ("caching_sha2_password".equals(pluginName)) {
|
||||
byte[] scramble = authData;
|
||||
try {
|
||||
encryptedPassword = MySQLPasswordEncrypter.scrambleCachingSha2(getPassword().getBytes(),
|
||||
scramble);
|
||||
} catch (DigestException e) {
|
||||
throw new RuntimeException("can't encrypt password that will be sent to MySQL server.", e);
|
||||
}
|
||||
header = authSwitchAfterAuth(encryptedPassword, header);
|
||||
body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout);
|
||||
assert body != null;
|
||||
if (body[0] == 0x01 && body[1] == 0x04) {
|
||||
// fixed issue https://github.com/alibaba/canal/pull/4767, support mysql 8.0.30+
|
||||
header = cachingSha2PasswordFullAuth(channel, header, getPassword().getBytes(), scramble);
|
||||
body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout);
|
||||
} else {
|
||||
header = PacketManager.readHeader(channel, 4);
|
||||
body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout);
|
||||
}
|
||||
} else {
|
||||
header = authSwitchAfterAuth(encryptedPassword, header);
|
||||
header = PacketManager.readHeader(channel, 4);
|
||||
body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout);
|
||||
}
|
||||
} else {
|
||||
header = authSwitchAfterAuth(encryptedPassword, header);
|
||||
body = PacketManager.readBytes(channel, header.getPacketBodyLength(), timeout);
|
||||
}
|
||||
}
|
||||
assert body != null;
|
||||
if (body[0] < 0) {
|
||||
if (body[0] == -1) {
|
||||
ErrorPacket err = new ErrorPacket();
|
||||
@@ -442,6 +380,14 @@ public class MysqlConnector {
|
||||
this.username = username;
|
||||
}
|
||||
|
||||
public byte getCharsetNumber() {
|
||||
return charsetNumber;
|
||||
}
|
||||
|
||||
public void setCharsetNumber(byte charsetNumber) {
|
||||
this.charsetNumber = charsetNumber;
|
||||
}
|
||||
|
||||
public String getDefaultSchema() {
|
||||
return defaultSchema;
|
||||
}
|
||||
@@ -482,6 +428,10 @@ public class MysqlConnector {
|
||||
this.channel = channel;
|
||||
}
|
||||
|
||||
public void setPassword(String password) {
|
||||
this.password = password;
|
||||
}
|
||||
|
||||
public long getConnectionId() {
|
||||
return connectionId;
|
||||
}
|
||||
@@ -510,19 +460,8 @@ public class MysqlConnector {
|
||||
return password;
|
||||
}
|
||||
|
||||
public void setPassword(String password) {
|
||||
this.password = password;
|
||||
}
|
||||
|
||||
public String getServerVersion() {
|
||||
return serverVersion;
|
||||
}
|
||||
|
||||
public SslInfo getSslInfo() {
|
||||
return sslInfo;
|
||||
}
|
||||
|
||||
public void setSslInfo(SslInfo sslInfo) {
|
||||
this.sslInfo = sslInfo;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,65 +0,0 @@
|
||||
package com.alibaba.otter.canal.parse.driver.mysql.packets.client;
|
||||
|
||||
import java.io.ByteArrayOutputStream;
|
||||
import java.io.IOException;
|
||||
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.packets.Capability;
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.utils.ByteHelper;
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.utils.MSC;
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.utils.MySQLPasswordEncrypter;
|
||||
|
||||
import org.apache.commons.lang.StringUtils;
|
||||
|
||||
public class ClientAuthenticationSHA2Packet extends ClientAuthenticationPacket {
|
||||
|
||||
@Override
|
||||
public byte[] toBytes() throws IOException {
|
||||
ByteArrayOutputStream out = new ByteArrayOutputStream();
|
||||
// 1. write client_flags
|
||||
int clientCapability = 0;
|
||||
clientCapability |= Capability.CLIENT_LONG_PASSWORD;
|
||||
clientCapability |= Capability.CLIENT_LONG_FLAG;
|
||||
clientCapability |= Capability.CLIENT_PROTOCOL_41;
|
||||
clientCapability |= Capability.CLIENT_TRANSACTIONS;
|
||||
clientCapability |= Capability.CLIENT_SECURE_CONNECTION;
|
||||
clientCapability |= Capability.CLIENT_MULTI_STATEMENTS;
|
||||
clientCapability |= Capability.CLIENT_PLUGIN_AUTH;
|
||||
clientCapability |= Capability.CLIENT_PLUGIN_AUTH_LENENC_CLIENT_DATA;
|
||||
if (getDatabaseName() != null) {
|
||||
clientCapability |= Capability.CLIENT_CONNECT_WITH_DB;
|
||||
}
|
||||
ByteHelper.writeUnsignedIntLittleEndian(clientCapability, out);
|
||||
|
||||
// 2. write max_packet_size
|
||||
ByteHelper.writeUnsignedIntLittleEndian(MSC.MAX_PACKET_LENGTH, out);
|
||||
// 3. write charset_number
|
||||
out.write(getCharsetNumber());
|
||||
// 4. write (filler) always 0x00...
|
||||
out.write(new byte[23]);
|
||||
// 5. write (Null-Terminated String) user
|
||||
ByteHelper.writeNullTerminatedString(getUsername(), out);
|
||||
// 6. write (Length Coded Binary) scramble_buff (1 + x bytes)
|
||||
if (StringUtils.isEmpty(getPassword())) {
|
||||
out.write(0x00);
|
||||
} else {
|
||||
try {
|
||||
byte[] encryptedPassword = MySQLPasswordEncrypter.scrambleCachingSha2(getPassword().getBytes(),
|
||||
getScrumbleBuff());
|
||||
ByteHelper.writeBinaryCodedLengthBytes(encryptedPassword, out);
|
||||
} catch (Exception e) {
|
||||
throw new IOException("can't encrypt password that will be sent to MySQL server.", e);
|
||||
}
|
||||
}
|
||||
// 7 . (Null-Terminated String) databasename (optional)
|
||||
if (getDatabaseName() != null) {
|
||||
ByteHelper.writeNullTerminatedString(getDatabaseName(), out);
|
||||
}
|
||||
// 8 . (Null-Terminated String) auth plugin name (optional)
|
||||
if (getAuthPluginName() != null) {
|
||||
ByteHelper.writeNullTerminated(getAuthPluginName(), out);
|
||||
}
|
||||
// end write
|
||||
return out.toByteArray();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,41 +0,0 @@
|
||||
package com.alibaba.otter.canal.parse.driver.mysql.packets.client;
|
||||
|
||||
import java.io.ByteArrayOutputStream;
|
||||
import java.io.IOException;
|
||||
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.packets.Capability;
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.packets.IPacket;
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.utils.ByteHelper;
|
||||
|
||||
/**
|
||||
* @author 枭博
|
||||
* @date 2024/05/14
|
||||
*/
|
||||
public class SslRequestCommandPacket implements IPacket {
|
||||
|
||||
private final int serverCharsetNumber;
|
||||
|
||||
public SslRequestCommandPacket(int serverCharsetNumber){
|
||||
this.serverCharsetNumber = serverCharsetNumber;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void fromBytes(byte[] data) throws IOException {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
@Override
|
||||
public byte[] toBytes() throws IOException {
|
||||
ByteArrayOutputStream out = new ByteArrayOutputStream();
|
||||
int clientCapabilities = Capability.CLIENT_LONG_FLAG | Capability.CLIENT_PROTOCOL_41
|
||||
| Capability.CLIENT_SECURE_CONNECTION | Capability.CLIENT_PLUGIN_AUTH
|
||||
| Capability.CLIENT_SSL;
|
||||
ByteHelper.writeUnsignedIntLittleEndian(clientCapabilities, out);
|
||||
ByteHelper.writeUnsignedIntLittleEndian(0, out);
|
||||
out.write(serverCharsetNumber);
|
||||
for (int i = 0; i < 23; i++) {
|
||||
out.write(0);
|
||||
}
|
||||
return out.toByteArray();
|
||||
}
|
||||
}
|
||||
@@ -10,8 +10,6 @@ import java.net.SocketException;
|
||||
import java.net.SocketTimeoutException;
|
||||
import java.nio.channels.ClosedByInterruptException;
|
||||
|
||||
import javax.net.ssl.SSLSocket;
|
||||
|
||||
/**
|
||||
* 使用BIO进行dump
|
||||
*
|
||||
@@ -19,18 +17,16 @@ import javax.net.ssl.SSLSocket;
|
||||
*/
|
||||
public class BioSocketChannel implements SocketChannel {
|
||||
|
||||
static final int DEFAULT_CONNECT_TIMEOUT = 10 * 1000;
|
||||
static final int SO_TIMEOUT = 1000;
|
||||
private Socket socket;
|
||||
private InputStream input;
|
||||
private OutputStream output;
|
||||
private final boolean ssl;
|
||||
static final int DEFAULT_CONNECT_TIMEOUT = 10 * 1000;
|
||||
static final int SO_TIMEOUT = 1000;
|
||||
private Socket socket;
|
||||
private InputStream input;
|
||||
private OutputStream output;
|
||||
|
||||
BioSocketChannel(Socket socket) throws IOException{
|
||||
this.socket = socket;
|
||||
this.input = new BufferedInputStream(socket.getInputStream(), 16384);
|
||||
this.output = socket.getOutputStream();
|
||||
this.ssl = (socket instanceof SSLSocket);
|
||||
}
|
||||
|
||||
public void write(byte[]... buf) throws IOException {
|
||||
@@ -92,9 +88,9 @@ public class BioSocketChannel implements SocketChannel {
|
||||
}
|
||||
}
|
||||
if (remain > 0 && accTimeout >= timeout) {
|
||||
throw new SocketTimeoutException(
|
||||
"Timeout occurred, failed to read total " + readSize + " bytes in " + timeout
|
||||
+ " milliseconds, actual read only " + (readSize - remain) + " bytes");
|
||||
throw new SocketTimeoutException("Timeout occurred, failed to read total " + readSize + " bytes in "
|
||||
+ timeout + " milliseconds, actual read only " + (readSize - remain)
|
||||
+ " bytes");
|
||||
}
|
||||
return data;
|
||||
}
|
||||
@@ -138,14 +134,6 @@ public class BioSocketChannel implements SocketChannel {
|
||||
return false;
|
||||
}
|
||||
|
||||
public boolean isSsl() {
|
||||
return ssl;
|
||||
}
|
||||
|
||||
public Socket getSocket() {
|
||||
return socket;
|
||||
}
|
||||
|
||||
public SocketAddress getRemoteSocketAddress() {
|
||||
Socket socket = this.socket;
|
||||
if (socket != null) {
|
||||
@@ -167,6 +155,16 @@ public class BioSocketChannel implements SocketChannel {
|
||||
public void close() {
|
||||
Socket socket = this.socket;
|
||||
if (socket != null) {
|
||||
try {
|
||||
socket.shutdownInput();
|
||||
} catch (IOException e) {
|
||||
// Ignore, could not do anymore
|
||||
}
|
||||
try {
|
||||
socket.shutdownOutput();
|
||||
} catch (IOException e) {
|
||||
// Ignore, could not do anymore
|
||||
}
|
||||
try {
|
||||
socket.close();
|
||||
} catch (IOException e) {
|
||||
|
||||
@@ -1,28 +1,7 @@
|
||||
package com.alibaba.otter.canal.parse.driver.mysql.socket;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.net.Socket;
|
||||
import java.net.SocketAddress;
|
||||
import java.net.URL;
|
||||
import java.security.InvalidAlgorithmParameterException;
|
||||
import java.security.KeyStore;
|
||||
import java.security.cert.*;
|
||||
import java.util.*;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import javax.naming.InvalidNameException;
|
||||
import javax.naming.ldap.LdapName;
|
||||
import javax.naming.ldap.Rdn;
|
||||
import javax.net.ssl.*;
|
||||
import javax.security.auth.x500.X500Principal;
|
||||
|
||||
import org.apache.commons.lang.StringUtils;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo;
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslMode;
|
||||
|
||||
/**
|
||||
* @author luoyaogui 实现channel的管理（监听连接、读数据、回收） 2016-12-28
|
||||
@@ -30,295 +9,14 @@ import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslMode;
|
||||
*/
|
||||
public abstract class BioSocketChannelPool {
|
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(BioSocketChannelPool.class);
|
||||
|
||||
public static BioSocketChannel open(SocketAddress address) throws Exception {
|
||||
Socket socket = createSocket(address);
|
||||
return new BioSocketChannel(socket);
|
||||
}
|
||||
|
||||
public static BioSocketChannel openSsl(Socket socket, SslInfo sslInfo) throws Exception {
|
||||
SslMode sslMode = sslInfo.getSslMode();
|
||||
switch (sslMode) {
|
||||
case REQUIRED:
|
||||
case PREFERRED:
|
||||
case VERIFY_CA:
|
||||
case VERIFY_IDENTITY:
|
||||
SSLSocket sslSocket = createSslSocket(socket, sslInfo);
|
||||
return new BioSocketChannel(sslSocket);
|
||||
default:
|
||||
throw new UnsupportedOperationException("Unsupported ssl mode: " + sslMode);
|
||||
}
|
||||
}
|
||||
|
||||
private static Socket createSocket(SocketAddress address) throws IOException {
|
||||
Socket socket;
|
||||
socket = new Socket();
|
||||
Socket socket = new Socket();
|
||||
socket.setSoTimeout(BioSocketChannel.SO_TIMEOUT);
|
||||
socket.setTcpNoDelay(true);
|
||||
socket.setKeepAlive(true);
|
||||
socket.setReuseAddress(true);
|
||||
socket.connect(address, BioSocketChannel.DEFAULT_CONNECT_TIMEOUT);
|
||||
return socket;
|
||||
}
|
||||
|
||||
/**
|
||||
* from JDBC driver com.mysql.cj.protocol.ExportControlled#performTlsHandshake
|
||||
* com.mysql.cj.protocol.ExportControlled#getSSLContext
|
||||
*
|
||||
* @param socket
|
||||
* @param sslInfo
|
||||
* @return
|
||||
* @throws Exception
|
||||
*/
|
||||
private static SSLSocket createSslSocket(Socket socket, SslInfo sslInfo) throws Exception {
|
||||
SslMode sslMode = sslInfo.getSslMode();
|
||||
boolean verifyServerCert = sslMode == SslMode.VERIFY_CA || sslMode == SslMode.VERIFY_IDENTITY;
|
||||
|
||||
String clientCertificateKeyStoreUrl = sslInfo.getClientCertificateKeyStoreUrl();
|
||||
String clientCertificateKeyStoreType = sslInfo.getClientCertificateKeyStoreType() != null ? sslInfo
|
||||
.getClientCertificateKeyStoreType() : "JKS";
|
||||
String clientCertificateKeyStorePassword = sslInfo.getClientCertificateKeyStorePassword();
|
||||
String trustCertificateKeyStoreUrl = sslInfo.getTrustCertificateKeyStoreUrl();
|
||||
String trustCertificateKeyStoreType = sslInfo.getTrustCertificateKeyStoreType() != null ? sslInfo
|
||||
.getTrustCertificateKeyStoreType() : "JKS";
|
||||
String trustCertificateKeyStorePassword = sslInfo.getTrustCertificateKeyStorePassword();
|
||||
boolean fallbackToDefaultTrustStore = true;
|
||||
String hostName = sslMode == SslMode.VERIFY_IDENTITY ? socket.getInetAddress().getHostName() : null;
|
||||
|
||||
SSLContext sslContext = getSSLContext(clientCertificateKeyStoreUrl,
|
||||
clientCertificateKeyStoreType,
|
||||
clientCertificateKeyStorePassword,
|
||||
trustCertificateKeyStoreUrl,
|
||||
trustCertificateKeyStoreType,
|
||||
trustCertificateKeyStorePassword,
|
||||
fallbackToDefaultTrustStore,
|
||||
verifyServerCert,
|
||||
hostName);
|
||||
SSLSocketFactory socketFactory = sslContext.getSocketFactory();
|
||||
|
||||
SSLSocket sslSocket = (SSLSocket) socketFactory
|
||||
.createSocket(socket, socket.getInetAddress().getHostName(), socket.getPort(), true);
|
||||
|
||||
String[] protocolArr = null;
|
||||
if (StringUtils.isNotEmpty(sslInfo.getTlsVersions())) {
|
||||
protocolArr = StringUtils.split(sslInfo.getTlsVersions(), ",");
|
||||
}
|
||||
if (protocolArr == null || protocolArr.length == 0) {
|
||||
protocolArr = new String[] { "TLSv1.2", "TLSv1.3" };
|
||||
}
|
||||
logger.info("SSL protocol: {}", StringUtils.join(protocolArr, ","));
|
||||
sslSocket.setEnabledProtocols(protocolArr);
|
||||
sslSocket.setSoTimeout(BioSocketChannel.SO_TIMEOUT);
|
||||
sslSocket.startHandshake();
|
||||
logger.info("SSL socket handshake success.");
|
||||
return sslSocket;
|
||||
}
|
||||
|
||||
private static SSLContext getSSLContext(String clientCertificateKeyStoreUrl, String clientCertificateKeyStoreType,
|
||||
String clientCertificateKeyStorePassword,
|
||||
String trustCertificateKeyStoreUrl, String trustCertificateKeyStoreType,
|
||||
String trustCertificateKeyStorePassword,
|
||||
boolean fallbackToDefaultTrustStore, boolean verifyServerCert,
|
||||
String hostName) throws Exception {
|
||||
|
||||
KeyManager[] kms = null;
|
||||
List<TrustManager> tms = new ArrayList<>();
|
||||
|
||||
TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
|
||||
KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
|
||||
|
||||
if (StringUtils.isNotEmpty(clientCertificateKeyStoreUrl)) {
|
||||
InputStream ksIS = null;
|
||||
try {
|
||||
if (StringUtils.isNotEmpty(clientCertificateKeyStoreType)) {
|
||||
KeyStore clientKeyStore = KeyStore.getInstance(clientCertificateKeyStoreType);
|
||||
URL ksURL = new URL(clientCertificateKeyStoreUrl);
|
||||
char[] password = (clientCertificateKeyStorePassword == null) ? new char[0] : clientCertificateKeyStorePassword
|
||||
.toCharArray();
|
||||
ksIS = ksURL.openStream();
|
||||
clientKeyStore.load(ksIS, password);
|
||||
kmf.init(clientKeyStore, password);
|
||||
kms = kmf.getKeyManagers();
|
||||
}
|
||||
} finally {
|
||||
if (ksIS != null) {
|
||||
try {
|
||||
ksIS.close();
|
||||
} catch (IOException e) {
|
||||
// can't close input stream, but keystore can be properly initialized so we
|
||||
// shouldn't throw this exception
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
InputStream trustStoreIS = null;
|
||||
try {
|
||||
String trustStoreType = "";
|
||||
char[] trustStorePassword = null;
|
||||
KeyStore trustKeyStore = null;
|
||||
|
||||
if (StringUtils.isNotEmpty(trustCertificateKeyStoreUrl)
|
||||
&& StringUtils.isNotEmpty(trustCertificateKeyStoreType)) {
|
||||
trustStoreType = trustCertificateKeyStoreType;
|
||||
trustStorePassword = (trustCertificateKeyStorePassword == null) ? new char[0] : trustCertificateKeyStorePassword
|
||||
.toCharArray();
|
||||
trustStoreIS = new URL(trustCertificateKeyStoreUrl).openStream();
|
||||
|
||||
trustKeyStore = KeyStore.getInstance(trustStoreType);
|
||||
trustKeyStore.load(trustStoreIS, trustStorePassword);
|
||||
}
|
||||
|
||||
if (trustKeyStore != null || fallbackToDefaultTrustStore) {
|
||||
tmf.init(trustKeyStore);
|
||||
// (trustKeyStore == null) initializes the TrustManagerFactory with the default
|
||||
// truststore.
|
||||
|
||||
// building the customized list of TrustManagers from original one if it's
|
||||
// available
|
||||
TrustManager[] origTms = tmf.getTrustManagers();
|
||||
|
||||
for (TrustManager tm : origTms) {
|
||||
// wrap X509TrustManager or put original if non-X509 TrustManager
|
||||
tms.add(tm instanceof X509TrustManager ? new X509TrustManagerWrapper((X509TrustManager) tm,
|
||||
verifyServerCert,
|
||||
hostName) : tm);
|
||||
}
|
||||
}
|
||||
|
||||
} finally {
|
||||
if (trustStoreIS != null) {
|
||||
try {
|
||||
trustStoreIS.close();
|
||||
} catch (IOException e) {
|
||||
// can't close input stream, but keystore can be properly initialized so we
|
||||
// shouldn't throw this exception
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// if original TrustManagers are not available then putting one
|
||||
// X509TrustManagerWrapper which take care only
|
||||
// about expiration check
|
||||
if (tms.size() == 0) {
|
||||
tms.add(new X509TrustManagerWrapper(verifyServerCert, hostName));
|
||||
}
|
||||
|
||||
SSLContext sslContext = SSLContext.getInstance("TLS");
|
||||
sslContext.init(kms, tms.toArray(new TrustManager[tms.size()]), null);
|
||||
return sslContext;
|
||||
}
|
||||
|
||||
public static class X509TrustManagerWrapper implements X509TrustManager {
|
||||
|
||||
private X509TrustManager origTm = null;
|
||||
private boolean verifyServerCert = false;
|
||||
private String hostName = null;
|
||||
private CertificateFactory certFactory = null;
|
||||
private PKIXParameters validatorParams = null;
|
||||
private CertPathValidator validator = null;
|
||||
|
||||
public X509TrustManagerWrapper(X509TrustManager tm, boolean verifyServerCertificate,
|
||||
String hostName) throws CertificateException{
|
||||
this.origTm = tm;
|
||||
this.verifyServerCert = verifyServerCertificate;
|
||||
this.hostName = hostName;
|
||||
|
||||
if (verifyServerCertificate) {
|
||||
try {
|
||||
Set<TrustAnchor> anch = Arrays.stream(tm.getAcceptedIssuers())
|
||||
.map(c -> new TrustAnchor(c, null))
|
||||
.collect(Collectors.toSet());
|
||||
this.validatorParams = new PKIXParameters(anch);
|
||||
this.validatorParams.setRevocationEnabled(false);
|
||||
this.validator = CertPathValidator.getInstance("PKIX");
|
||||
this.certFactory = CertificateFactory.getInstance("X.509");
|
||||
} catch (Exception e) {
|
||||
throw new CertificateException(e);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public X509TrustManagerWrapper(boolean verifyServerCertificate, String hostName){
|
||||
this.verifyServerCert = verifyServerCertificate;
|
||||
this.hostName = hostName;
|
||||
}
|
||||
|
||||
public X509Certificate[] getAcceptedIssuers() {
|
||||
return this.origTm != null ? this.origTm.getAcceptedIssuers() : new X509Certificate[0];
|
||||
}
|
||||
|
||||
public void checkServerTrusted(X509Certificate[] chain, String authType) throws CertificateException {
|
||||
for (int i = 0; i < chain.length; i++) {
|
||||
chain[i].checkValidity();
|
||||
}
|
||||
|
||||
if (this.validatorParams != null) {
|
||||
X509CertSelector certSelect = new X509CertSelector();
|
||||
certSelect.setSerialNumber(chain[0].getSerialNumber());
|
||||
|
||||
try {
|
||||
CertPath certPath = this.certFactory.generateCertPath(Arrays.asList(chain));
|
||||
// Validate against truststore
|
||||
CertPathValidatorResult result = this.validator.validate(certPath, this.validatorParams);
|
||||
// Check expiration for the CA used to validate this path
|
||||
((PKIXCertPathValidatorResult) result).getTrustAnchor().getTrustedCert().checkValidity();
|
||||
|
||||
} catch (InvalidAlgorithmParameterException e) {
|
||||
throw new CertificateException(e);
|
||||
} catch (CertPathValidatorException e) {
|
||||
throw new CertificateException(e);
|
||||
}
|
||||
}
|
||||
|
||||
if (this.verifyServerCert) {
|
||||
if (this.origTm != null) {
|
||||
this.origTm.checkServerTrusted(chain, authType);
|
||||
} else {
|
||||
throw new CertificateException(
|
||||
"Can't verify server certificate because no trust manager is found.");
|
||||
}
|
||||
|
||||
// verify server certificate identity
|
||||
if (this.hostName != null) {
|
||||
logger.info("verify hostName: {}", this.hostName);
|
||||
Set<String> expectHostNames = new HashSet<>();
|
||||
for (X509Certificate certificate : chain) {
|
||||
String dn = certificate.getSubjectX500Principal().getName(X500Principal.RFC2253);
|
||||
String cn = null;
|
||||
try {
|
||||
LdapName ldapDN = new LdapName(dn);
|
||||
for (Rdn rdn : ldapDN.getRdns()) {
|
||||
if (rdn.getType().equalsIgnoreCase("CN")) {
|
||||
cn = rdn.getValue().toString();
|
||||
break;
|
||||
}
|
||||
}
|
||||
} catch (InvalidNameException e) {
|
||||
throw new CertificateException(
|
||||
"Failed to retrieve the Common Name (CN) from the server certificate.");
|
||||
}
|
||||
expectHostNames.add(cn);
|
||||
}
|
||||
|
||||
if (!expectHostNames.contains(this.hostName)) {
|
||||
throw new CertificateException(
|
||||
"Server certificate identity check failed. The certificate Common Name "
|
||||
+ expectHostNames.stream()
|
||||
.map(h -> "'" + h + "'")
|
||||
.collect(Collectors.joining(", "))
|
||||
+ " does not match with '" + this.hostName + "'.");
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public void checkClientTrusted(X509Certificate[] chain, String authType) throws CertificateException {
|
||||
this.origTm.checkClientTrusted(chain, authType);
|
||||
}
|
||||
return new BioSocketChannel(socket);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,14 +1,8 @@
|
||||
package com.alibaba.otter.canal.parse.driver.mysql.socket;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.SocketAddress;
|
||||
|
||||
import org.apache.commons.lang.StringUtils;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo;
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslMode;
|
||||
|
||||
/**
|
||||
* @author agapple 2018年3月12日 下午10:46:22
|
||||
@@ -16,8 +10,6 @@ import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslMode;
|
||||
*/
|
||||
public abstract class SocketChannelPool {
|
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(SocketChannelPool.class);
|
||||
|
||||
public static SocketChannel open(SocketAddress address) throws Exception {
|
||||
String type = chooseSocketChannel();
|
||||
if ("netty".equalsIgnoreCase(type)) {
|
||||
@@ -25,32 +17,7 @@ public abstract class SocketChannelPool {
|
||||
} else {
|
||||
return BioSocketChannelPool.open(address);
|
||||
}
|
||||
}
|
||||
|
||||
public static SocketChannel connectSsl(SocketChannel channel, SslInfo sslInfo) throws IOException {
|
||||
SslMode sslMode = sslInfo.getSslMode();
|
||||
String type = chooseSocketChannel();
|
||||
if ("netty".equalsIgnoreCase(type)) {
|
||||
throw new UnsupportedOperationException("canal socketChannel netty not support ssl mode: " + sslMode);
|
||||
} else {
|
||||
SocketAddress remoteSocketAddress = channel.getRemoteSocketAddress();
|
||||
try {
|
||||
return BioSocketChannelPool.openSsl(((BioSocketChannel) channel).getSocket(), sslInfo);
|
||||
} catch (Exception e) {
|
||||
if (sslMode == SslMode.PREFERRED) {
|
||||
// still use non ssl channel
|
||||
logger.info("{} still use non SSL channel due to SSL connect failed.", remoteSocketAddress, e);
|
||||
return channel;
|
||||
}
|
||||
IOException ioe;
|
||||
if (e instanceof IOException) {
|
||||
ioe = (IOException) e;
|
||||
} else {
|
||||
ioe = new IOException(e);
|
||||
}
|
||||
throw ioe;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static String chooseSocketChannel() {
|
||||
|
||||
@@ -1,98 +0,0 @@
|
||||
package com.alibaba.otter.canal.parse.driver.mysql.ssl;
|
||||
|
||||
/**
|
||||
* @author 枭博
|
||||
* @date 2024/05/14
|
||||
*/
|
||||
public class SslInfo {
|
||||
|
||||
private SslMode sslMode = SslMode.DISABLED;
|
||||
private String tlsVersions; // 和 enabledTLSProtocols 同含义，TLSv1.2,TLSv1.3
|
||||
private String trustCertificateKeyStoreType; // trustStore 证书类型，支持 JKS (默认) 和 PKCS12
|
||||
private String trustCertificateKeyStoreUrl; // trustStore 证书
|
||||
private String trustCertificateKeyStorePassword; // trustStore 证书密码
|
||||
private String clientCertificateKeyStoreType; // client 证书类型，支持 JKS (默认) 和 PKCS12
|
||||
private String clientCertificateKeyStoreUrl; // client 证书
|
||||
private String clientCertificateKeyStorePassword; // client 证书密码
|
||||
|
||||
public SslInfo(SslMode sslMode, String tlsVersions, String trustCertificateKeyStoreType,
|
||||
String trustCertificateKeyStoreUrl, String trustCertificateKeyStorePassword,
|
||||
String clientCertificateKeyStoreType, String clientCertificateKeyStoreUrl,
|
||||
String clientCertificateKeyStorePassword) {
|
||||
this.sslMode = sslMode;
|
||||
this.tlsVersions = tlsVersions;
|
||||
this.trustCertificateKeyStoreType = trustCertificateKeyStoreType;
|
||||
this.trustCertificateKeyStoreUrl = trustCertificateKeyStoreUrl;
|
||||
this.trustCertificateKeyStorePassword = trustCertificateKeyStorePassword;
|
||||
this.clientCertificateKeyStoreType = clientCertificateKeyStoreType;
|
||||
this.clientCertificateKeyStoreUrl = clientCertificateKeyStoreUrl;
|
||||
this.clientCertificateKeyStorePassword = clientCertificateKeyStorePassword;
|
||||
}
|
||||
|
||||
public SslInfo() {
|
||||
}
|
||||
|
||||
public SslMode getSslMode() {
|
||||
return sslMode;
|
||||
}
|
||||
|
||||
public void setSslMode(SslMode sslMode) {
|
||||
this.sslMode = sslMode;
|
||||
}
|
||||
|
||||
public String getTlsVersions() {
|
||||
return tlsVersions;
|
||||
}
|
||||
|
||||
public void setTlsVersions(String tlsVersions) {
|
||||
this.tlsVersions = tlsVersions;
|
||||
}
|
||||
|
||||
public String getTrustCertificateKeyStoreType() {
|
||||
return trustCertificateKeyStoreType;
|
||||
}
|
||||
|
||||
public void setTrustCertificateKeyStoreType(String trustCertificateKeyStoreType) {
|
||||
this.trustCertificateKeyStoreType = trustCertificateKeyStoreType;
|
||||
}
|
||||
|
||||
public String getTrustCertificateKeyStoreUrl() {
|
||||
return trustCertificateKeyStoreUrl;
|
||||
}
|
||||
|
||||
public void setTrustCertificateKeyStoreUrl(String trustCertificateKeyStoreUrl) {
|
||||
this.trustCertificateKeyStoreUrl = trustCertificateKeyStoreUrl;
|
||||
}
|
||||
|
||||
public String getTrustCertificateKeyStorePassword() {
|
||||
return trustCertificateKeyStorePassword;
|
||||
}
|
||||
|
||||
public void setTrustCertificateKeyStorePassword(String trustCertificateKeyStorePassword) {
|
||||
this.trustCertificateKeyStorePassword = trustCertificateKeyStorePassword;
|
||||
}
|
||||
|
||||
public String getClientCertificateKeyStoreType() {
|
||||
return clientCertificateKeyStoreType;
|
||||
}
|
||||
|
||||
public void setClientCertificateKeyStoreType(String clientCertificateKeyStoreType) {
|
||||
this.clientCertificateKeyStoreType = clientCertificateKeyStoreType;
|
||||
}
|
||||
|
||||
public String getClientCertificateKeyStoreUrl() {
|
||||
return clientCertificateKeyStoreUrl;
|
||||
}
|
||||
|
||||
public void setClientCertificateKeyStoreUrl(String clientCertificateKeyStoreUrl) {
|
||||
this.clientCertificateKeyStoreUrl = clientCertificateKeyStoreUrl;
|
||||
}
|
||||
|
||||
public String getClientCertificateKeyStorePassword() {
|
||||
return clientCertificateKeyStorePassword;
|
||||
}
|
||||
|
||||
public void setClientCertificateKeyStorePassword(String clientCertificateKeyStorePassword) {
|
||||
this.clientCertificateKeyStorePassword = clientCertificateKeyStorePassword;
|
||||
}
|
||||
}
|
||||
@@ -1,26 +0,0 @@
|
||||
package com.alibaba.otter.canal.parse.driver.mysql.ssl;
|
||||
|
||||
public enum SslMode {
|
||||
|
||||
/**
|
||||
* 关闭SSL
|
||||
*/
|
||||
DISABLED,
|
||||
/**
|
||||
* 尝试SSL传输
|
||||
*/
|
||||
PREFERRED,
|
||||
/**
|
||||
* 要求SSL传输，不校验证书
|
||||
*/
|
||||
REQUIRED,
|
||||
/**
|
||||
* 要求SSL传输，校验证书，不校验证书里的域名
|
||||
*/
|
||||
VERIFY_CA,
|
||||
/**
|
||||
* 要求SSL传输，校验证书，校验证书里的域名
|
||||
*/
|
||||
VERIFY_IDENTITY
|
||||
|
||||
}
|
||||
@@ -276,7 +276,8 @@ public class CanalInstanceWithManager extends AbstractCanalInstance {
|
||||
mysqlEventParser.setDestination(destination);
|
||||
// 编码参数
|
||||
mysqlEventParser.setConnectionCharset(parameters.getConnectionCharset());
|
||||
// 网络相关参数1
|
||||
mysqlEventParser.setConnectionCharsetNumber(parameters.getConnectionCharsetNumber());
|
||||
// 网络相关参数
|
||||
mysqlEventParser.setDefaultConnectionTimeoutInSeconds(parameters.getDefaultConnectionTimeoutInSeconds());
|
||||
mysqlEventParser.setSendBufferSize(parameters.getSendBufferSize());
|
||||
mysqlEventParser.setReceiveBufferSize(parameters.getReceiveBufferSize());
|
||||
@@ -290,15 +291,13 @@ public class CanalInstanceWithManager extends AbstractCanalInstance {
|
||||
mysqlEventParser.setMasterInfo(new AuthenticationInfo(dbAddresses.get(0),
|
||||
parameters.getDbUsername(),
|
||||
parameters.getDbPassword(),
|
||||
parameters.getDefaultDatabaseName(),
|
||||
parameters.getSslInfo()));
|
||||
parameters.getDefaultDatabaseName()));
|
||||
|
||||
if (dbAddresses.size() > 1) {
|
||||
mysqlEventParser.setStandbyInfo(new AuthenticationInfo(dbAddresses.get(1),
|
||||
parameters.getDbUsername(),
|
||||
parameters.getDbPassword(),
|
||||
parameters.getDefaultDatabaseName(),
|
||||
parameters.getSslInfo()));
|
||||
parameters.getDefaultDatabaseName()));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -370,6 +369,7 @@ public class CanalInstanceWithManager extends AbstractCanalInstance {
|
||||
localBinlogEventParser.setDestination(destination);
|
||||
localBinlogEventParser.setBufferSize(parameters.getReceiveBufferSize());
|
||||
localBinlogEventParser.setConnectionCharset(parameters.getConnectionCharset());
|
||||
localBinlogEventParser.setConnectionCharsetNumber(parameters.getConnectionCharsetNumber());
|
||||
localBinlogEventParser.setDirectory(parameters.getLocalBinlogDirectory());
|
||||
localBinlogEventParser.setProfilingEnabled(false);
|
||||
localBinlogEventParser.setDetectingEnable(parameters.getDetectingEnable());
|
||||
@@ -381,8 +381,7 @@ public class CanalInstanceWithManager extends AbstractCanalInstance {
|
||||
localBinlogEventParser.setMasterInfo(new AuthenticationInfo(dbAddresses.get(0),
|
||||
parameters.getDbUsername(),
|
||||
parameters.getDbPassword(),
|
||||
parameters.getDefaultDatabaseName(),
|
||||
parameters.getSslInfo()));
|
||||
parameters.getDefaultDatabaseName()));
|
||||
}
|
||||
|
||||
eventParser = localBinlogEventParser;
|
||||
|
||||
@@ -2,7 +2,6 @@ package com.alibaba.otter.canal.instance.manager;
|
||||
|
||||
import java.util.Properties;
|
||||
|
||||
import org.apache.commons.lang.StringUtils;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.BeanFactory;
|
||||
@@ -18,7 +17,7 @@ import com.alibaba.otter.canal.parse.CanalEventParser;
|
||||
|
||||
/**
|
||||
* 基于manager生成对应的{@linkplain CanalInstance}
|
||||
*
|
||||
*
|
||||
* @author jianghang 2012-7-12 下午05:37:09
|
||||
* @version 1.0.0
|
||||
*/
|
||||
@@ -70,9 +69,6 @@ public class PlainCanalInstanceGenerator implements CanalInstanceGenerator {
|
||||
// ================ setter / getter ================
|
||||
|
||||
private BeanFactory getBeanFactory(String springXml) {
|
||||
if (!StringUtils.startsWithIgnoreCase(springXml, "classpath:")) {
|
||||
springXml = "classpath:" + springXml;
|
||||
}
|
||||
ApplicationContext applicationContext = new ClassPathXmlApplicationContext(springXml);
|
||||
return applicationContext;
|
||||
}
|
||||
|
||||
@@ -9,8 +9,6 @@ import org.apache.commons.lang.StringUtils;
|
||||
import org.apache.commons.lang.builder.ToStringBuilder;
|
||||
|
||||
import com.alibaba.otter.canal.common.utils.CanalToStringStyle;
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo;
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslMode;
|
||||
|
||||
/**
|
||||
* canal运行相关参数
|
||||
@@ -58,6 +56,8 @@ public class CanalParameter implements Serializable {
|
||||
private Integer defaultConnectionTimeoutInSeconds = 30; // sotimeout
|
||||
private Integer receiveBufferSize = 64 * 1024;
|
||||
private Integer sendBufferSize = 64 * 1024;
|
||||
// 编码信息
|
||||
private Byte connectionCharsetNumber = (byte) 33;
|
||||
private String connectionCharset = "UTF-8";
|
||||
|
||||
// 数据库信息
|
||||
@@ -66,21 +66,6 @@ public class CanalParameter implements Serializable {
|
||||
private String dbUsername; // 数据库用户
|
||||
private String dbPassword; // 数据库密码
|
||||
|
||||
private String sslMode = SslMode.DISABLED.name();
|
||||
private String tlsVersions; // 和
|
||||
// enabledTLSProtocols
|
||||
// 同含义，TLSv1.2,TLSv1.3
|
||||
private String trustCertificateKeyStoreType; // trustStore
|
||||
// 证书类型，支持 JKS (默认)
|
||||
// 和 PKCS12
|
||||
private String trustCertificateKeyStoreUrl; // trustStore 证书路径
|
||||
private String trustCertificateKeyStorePassword; // trustStore 证书密码
|
||||
private String clientCertificateKeyStoreType; // client 证书类型，支持
|
||||
// JKS (默认) 和
|
||||
// PKCS12
|
||||
private String clientCertificateKeyStoreUrl; // client 证书路径
|
||||
private String clientCertificateKeyStorePassword; // client 证书密码
|
||||
|
||||
// binlog链接信息
|
||||
private IndexMode indexMode;
|
||||
private List<String> positions; // 数据库positions信息
|
||||
@@ -123,39 +108,10 @@ public class CanalParameter implements Serializable {
|
||||
private InetSocketAddress masterAddress; // 主库信息
|
||||
private String masterUsername; // 帐号
|
||||
private String masterPassword; // 密码
|
||||
private String masterSslMode = SslMode.DISABLED.name();
|
||||
private String masterTlsVersions; // 和
|
||||
// enabledTLSProtocols
|
||||
// 同含义，TLSv1.2,TLSv1.3
|
||||
private String masterTrustCertificateKeyStoreType; // trustStore
|
||||
// 证书类型，支持 JKS (默认)
|
||||
// 和 PKCS12
|
||||
private String masterTrustCertificateKeyStoreUrl; // trustStore 证书路径
|
||||
private String masterTrustCertificateKeyStorePassword; // trustStore 证书密码
|
||||
private String masterClientCertificateKeyStoreType; // client 证书类型，支持
|
||||
// JKS (默认) 和
|
||||
// PKCS12
|
||||
private String masterClientCertificateKeyStoreUrl; // client 证书路径
|
||||
private String masterClientCertificateKeyStorePassword; // client 证书密码
|
||||
|
||||
private InetSocketAddress standbyAddress; // 备库信息
|
||||
private String standbyUsername; // 帐号
|
||||
private String standbyPassword;
|
||||
private String standbySslMode = SslMode.DISABLED.name();
|
||||
private String standbyTlsVersions; // 和
|
||||
// enabledTLSProtocols
|
||||
// 同含义，TLSv1.2,TLSv1.3
|
||||
private String standbyTrustCertificateKeyStoreType; // trustStore
|
||||
// 证书类型，支持 JKS (默认)
|
||||
// 和 PKCS12
|
||||
private String standbyTrustCertificateKeyStoreUrl; // trustStore 证书路径
|
||||
private String standbyTrustCertificateKeyStorePassword; // trustStore 证书密码
|
||||
private String standbyClientCertificateKeyStoreType; // client 证书类型，支持
|
||||
// JKS (默认) 和
|
||||
// PKCS12
|
||||
private String standbyClientCertificateKeyStoreUrl; // client 证书路径
|
||||
private String standbyClientCertificateKeyStorePassword; // client 证书密码
|
||||
|
||||
private String masterLogfileName = null; // master起始位置
|
||||
private Long masterLogfileOffest = null;
|
||||
private Long masterTimestamp = null;
|
||||
@@ -164,11 +120,11 @@ public class CanalParameter implements Serializable {
|
||||
private Long standbyTimestamp = null;
|
||||
private Boolean parallel = Boolean.FALSE;
|
||||
|
||||
// 自定义alarmHandler类全路径
|
||||
//自定义alarmHandler类全路径
|
||||
private String alarmHandlerClass = null;
|
||||
// 自定义alarmHandler插件文件夹路径
|
||||
//自定义alarmHandler插件文件夹路径
|
||||
private String alarmHandlerPluginDir = null;
|
||||
// 是否支持多流消费
|
||||
//是否支持多流消费
|
||||
private Boolean multiStreamEnable = Boolean.FALSE;
|
||||
|
||||
public static enum RunMode {
|
||||
@@ -227,7 +183,6 @@ public class CanalParameter implements Serializable {
|
||||
}
|
||||
|
||||
public static enum StorageMode {
|
||||
|
||||
/** 内存存储模式 */
|
||||
MEMORY,
|
||||
/** 文件存储模式 */
|
||||
@@ -250,7 +205,6 @@ public class CanalParameter implements Serializable {
|
||||
}
|
||||
|
||||
public static enum StorageScavengeMode {
|
||||
|
||||
/** 在存储满的时候触发 */
|
||||
ON_FULL,
|
||||
/** 在每次有ack请求时触发 */
|
||||
@@ -278,7 +232,6 @@ public class CanalParameter implements Serializable {
|
||||
}
|
||||
|
||||
public static enum SourcingType {
|
||||
|
||||
/** mysql DB */
|
||||
MYSQL,
|
||||
/** localBinLog */
|
||||
@@ -306,7 +259,6 @@ public class CanalParameter implements Serializable {
|
||||
}
|
||||
|
||||
public static enum MetaMode {
|
||||
|
||||
/** 内存存储模式 */
|
||||
MEMORY,
|
||||
/** 文件存储模式 */
|
||||
@@ -334,7 +286,6 @@ public class CanalParameter implements Serializable {
|
||||
}
|
||||
|
||||
public static enum IndexMode {
|
||||
|
||||
/** 内存存储模式 */
|
||||
MEMORY,
|
||||
/** 文件存储模式 */
|
||||
@@ -368,7 +319,6 @@ public class CanalParameter implements Serializable {
|
||||
}
|
||||
|
||||
public static enum BatchMode {
|
||||
|
||||
/** 对象数量 */
|
||||
ITEMSIZE,
|
||||
|
||||
@@ -583,6 +533,14 @@ public class CanalParameter implements Serializable {
|
||||
this.sendBufferSize = sendBufferSize;
|
||||
}
|
||||
|
||||
public Byte getConnectionCharsetNumber() {
|
||||
return connectionCharsetNumber;
|
||||
}
|
||||
|
||||
public void setConnectionCharsetNumber(Byte connectionCharsetNumber) {
|
||||
this.connectionCharsetNumber = connectionCharsetNumber;
|
||||
}
|
||||
|
||||
public String getConnectionCharset() {
|
||||
return connectionCharset;
|
||||
}
|
||||
@@ -773,134 +731,6 @@ public class CanalParameter implements Serializable {
|
||||
this.dbPassword = dbPassword;
|
||||
}
|
||||
|
||||
public SslInfo getSslInfo() {
|
||||
if (dbUsername == null) {
|
||||
if (masterUsername != null) {
|
||||
return new SslInfo(SslMode.valueOf(masterSslMode),
|
||||
masterTlsVersions,
|
||||
masterTrustCertificateKeyStoreType,
|
||||
masterTrustCertificateKeyStoreUrl,
|
||||
masterTrustCertificateKeyStorePassword,
|
||||
masterClientCertificateKeyStoreType,
|
||||
masterClientCertificateKeyStoreUrl,
|
||||
masterClientCertificateKeyStorePassword);
|
||||
} else {
|
||||
return new SslInfo(SslMode.valueOf(standbySslMode),
|
||||
standbyTlsVersions,
|
||||
standbyTrustCertificateKeyStoreType,
|
||||
standbyTrustCertificateKeyStoreUrl,
|
||||
standbyTrustCertificateKeyStorePassword,
|
||||
standbyClientCertificateKeyStoreType,
|
||||
standbyClientCertificateKeyStoreUrl,
|
||||
standbyClientCertificateKeyStorePassword);
|
||||
}
|
||||
}
|
||||
return new SslInfo(SslMode.valueOf(sslMode),
|
||||
tlsVersions,
|
||||
trustCertificateKeyStoreType,
|
||||
trustCertificateKeyStoreUrl,
|
||||
trustCertificateKeyStorePassword,
|
||||
clientCertificateKeyStoreType,
|
||||
clientCertificateKeyStoreUrl,
|
||||
clientCertificateKeyStorePassword);
|
||||
}
|
||||
|
||||
public void setSslMode(String sslMode) {
|
||||
this.sslMode = sslMode;
|
||||
}
|
||||
|
||||
public void setTlsVersions(String tlsVersions) {
|
||||
this.tlsVersions = tlsVersions;
|
||||
}
|
||||
|
||||
public void setTrustCertificateKeyStoreType(String trustCertificateKeyStoreType) {
|
||||
this.trustCertificateKeyStoreType = trustCertificateKeyStoreType;
|
||||
}
|
||||
|
||||
public void setTrustCertificateKeyStoreUrl(String trustCertificateKeyStoreUrl) {
|
||||
this.trustCertificateKeyStoreUrl = trustCertificateKeyStoreUrl;
|
||||
}
|
||||
|
||||
public void setTrustCertificateKeyStorePassword(String trustCertificateKeyStorePassword) {
|
||||
this.trustCertificateKeyStorePassword = trustCertificateKeyStorePassword;
|
||||
}
|
||||
|
||||
public void setClientCertificateKeyStoreType(String clientCertificateKeyStoreType) {
|
||||
this.clientCertificateKeyStoreType = clientCertificateKeyStoreType;
|
||||
}
|
||||
|
||||
public void setClientCertificateKeyStoreUrl(String clientCertificateKeyStoreUrl) {
|
||||
this.clientCertificateKeyStoreUrl = clientCertificateKeyStoreUrl;
|
||||
}
|
||||
|
||||
public void setClientCertificateKeyStorePassword(String clientCertificateKeyStorePassword) {
|
||||
this.clientCertificateKeyStorePassword = clientCertificateKeyStorePassword;
|
||||
}
|
||||
|
||||
public void setMasterSslMode(String masterSslMode) {
|
||||
this.masterSslMode = masterSslMode;
|
||||
}
|
||||
|
||||
public void setMasterTlsVersions(String masterTlsVersions) {
|
||||
this.masterTlsVersions = masterTlsVersions;
|
||||
}
|
||||
|
||||
public void setMasterTrustCertificateKeyStoreType(String masterTrustCertificateKeyStoreType) {
|
||||
this.masterTrustCertificateKeyStoreType = masterTrustCertificateKeyStoreType;
|
||||
}
|
||||
|
||||
public void setMasterTrustCertificateKeyStoreUrl(String masterTrustCertificateKeyStoreUrl) {
|
||||
this.masterTrustCertificateKeyStoreUrl = masterTrustCertificateKeyStoreUrl;
|
||||
}
|
||||
|
||||
public void setMasterTrustCertificateKeyStorePassword(String masterTrustCertificateKeyStorePassword) {
|
||||
this.masterTrustCertificateKeyStorePassword = masterTrustCertificateKeyStorePassword;
|
||||
}
|
||||
|
||||
public void setMasterClientCertificateKeyStoreType(String masterClientCertificateKeyStoreType) {
|
||||
this.masterClientCertificateKeyStoreType = masterClientCertificateKeyStoreType;
|
||||
}
|
||||
|
||||
public void setMasterClientCertificateKeyStoreUrl(String masterClientCertificateKeyStoreUrl) {
|
||||
this.masterClientCertificateKeyStoreUrl = masterClientCertificateKeyStoreUrl;
|
||||
}
|
||||
|
||||
public void setMasterClientCertificateKeyStorePassword(String masterClientCertificateKeyStorePassword) {
|
||||
this.masterClientCertificateKeyStorePassword = masterClientCertificateKeyStorePassword;
|
||||
}
|
||||
|
||||
public void setStandbySslMode(String standbySslMode) {
|
||||
this.standbySslMode = standbySslMode;
|
||||
}
|
||||
|
||||
public void setStandbyTlsVersions(String standbyTlsVersions) {
|
||||
this.standbyTlsVersions = standbyTlsVersions;
|
||||
}
|
||||
|
||||
public void setStandbyTrustCertificateKeyStoreType(String standbyTrustCertificateKeyStoreType) {
|
||||
this.standbyTrustCertificateKeyStoreType = standbyTrustCertificateKeyStoreType;
|
||||
}
|
||||
|
||||
public void setStandbyTrustCertificateKeyStoreUrl(String standbyTrustCertificateKeyStoreUrl) {
|
||||
this.standbyTrustCertificateKeyStoreUrl = standbyTrustCertificateKeyStoreUrl;
|
||||
}
|
||||
|
||||
public void setStandbyTrustCertificateKeyStorePassword(String standbyTrustCertificateKeyStorePassword) {
|
||||
this.standbyTrustCertificateKeyStorePassword = standbyTrustCertificateKeyStorePassword;
|
||||
}
|
||||
|
||||
public void setStandbyClientCertificateKeyStoreType(String standbyClientCertificateKeyStoreType) {
|
||||
this.standbyClientCertificateKeyStoreType = standbyClientCertificateKeyStoreType;
|
||||
}
|
||||
|
||||
public void setStandbyClientCertificateKeyStoreUrl(String standbyClientCertificateKeyStoreUrl) {
|
||||
this.standbyClientCertificateKeyStoreUrl = standbyClientCertificateKeyStoreUrl;
|
||||
}
|
||||
|
||||
public void setStandbyClientCertificateKeyStorePassword(String standbyClientCertificateKeyStorePassword) {
|
||||
this.standbyClientCertificateKeyStorePassword = standbyClientCertificateKeyStorePassword;
|
||||
}
|
||||
|
||||
public List<String> getPositions() {
|
||||
if (positions == null) {
|
||||
positions = new ArrayList<>();
|
||||
|
||||
@@ -1,6 +1,5 @@
|
||||
package com.alibaba.otter.canal.instance.spring;
|
||||
|
||||
import org.apache.commons.lang.StringUtils;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.BeanFactory;
|
||||
@@ -45,9 +44,6 @@ public class SpringCanalInstanceGenerator implements CanalInstanceGenerator {
|
||||
}
|
||||
|
||||
private BeanFactory getBeanFactory(String springXml) {
|
||||
if (!StringUtils.startsWithIgnoreCase(springXml, "classpath:")) {
|
||||
springXml = "classpath:" + springXml;
|
||||
}
|
||||
ApplicationContext applicationContext = new ClassPathXmlApplicationContext(springXml);
|
||||
return applicationContext;
|
||||
}
|
||||
|
||||
@@ -24,7 +24,7 @@ public class DefaultSpringInstanceTest {
|
||||
@Before
|
||||
public void start() {
|
||||
System.setProperty("canal.instance.destination", "retl");
|
||||
context = new ClassPathXmlApplicationContext(new String[] { "classpath:spring/default-instance.xml" });
|
||||
context = new ClassPathXmlApplicationContext(new String[] { "spring/default-instance.xml" });
|
||||
}
|
||||
|
||||
@After
|
||||
|
||||
@@ -24,7 +24,7 @@ public class GroupSpringInstanceTest {
|
||||
@Before
|
||||
public void start() {
|
||||
System.setProperty("canal.instance.destination", "retl");
|
||||
context = new ClassPathXmlApplicationContext(new String[] { "classpath:spring/group-instance.xml" });
|
||||
context = new ClassPathXmlApplicationContext(new String[] { "spring/group-instance.xml" });
|
||||
}
|
||||
|
||||
@After
|
||||
|
||||
@@ -24,7 +24,7 @@ public class MemorySpringInstanceTest {
|
||||
@Before
|
||||
public void start() {
|
||||
System.setProperty("canal.instance.destination", "retl");
|
||||
context = new ClassPathXmlApplicationContext(new String[] { "classpath:spring/memory-instance.xml" });
|
||||
context = new ClassPathXmlApplicationContext(new String[] { "spring/memory-instance.xml" });
|
||||
}
|
||||
|
||||
@After
|
||||
|
||||
@@ -114,6 +114,7 @@
|
||||
<property name="defaultConnectionTimeoutInSeconds" value="${canal.instance.network.soTimeout:30}" />
|
||||
|
||||
<!-- 解析编码 -->
|
||||
<!-- property name="connectionCharsetNumber" value="${canal.instance.connectionCharsetNumber:33}" /-->
|
||||
<property name="connectionCharset" value="${canal.instance.connectionCharset:UTF-8}" />
|
||||
|
||||
<!-- 解析位点记录 -->
|
||||
|
||||
@@ -100,6 +100,7 @@
|
||||
<property name="defaultConnectionTimeoutInSeconds" value="${canal.instance.network.soTimeout:30}" />
|
||||
|
||||
<!-- 解析编码 -->
|
||||
<!-- property name="connectionCharsetNumber" value="${canal.instance.connectionCharsetNumber:33}" /-->
|
||||
<property name="connectionCharset" value="${canal.instance.connectionCharset:UTF-8}" />
|
||||
|
||||
<!-- 解析位点记录 -->
|
||||
|
||||
@@ -106,6 +106,7 @@
|
||||
<property name="defaultConnectionTimeoutInSeconds" value="${canal.instance.network.soTimeout:30}" />
|
||||
|
||||
<!-- 解析编码 -->
|
||||
<!-- property name="connectionCharsetNumber" value="${canal.instance.connectionCharsetNumber:33}" /-->
|
||||
<property name="connectionCharset" value="${canal.instance.connectionCharset:UTF-8}" />
|
||||
|
||||
<!-- 解析位点记录 -->
|
||||
@@ -185,6 +186,7 @@
|
||||
<property name="defaultConnectionTimeoutInSeconds" value="${canal.instance.network.soTimeout:30}" />
|
||||
|
||||
<!-- 解析编码 -->
|
||||
<!-- property name="connectionCharsetNumber" value="${canal.instance.connectionCharsetNumber:33}" /-->
|
||||
<property name="connectionCharset" value="${canal.instance.connectionCharset:UTF-8}" />
|
||||
|
||||
<!-- 解析位点记录 -->
|
||||
|
||||
@@ -97,6 +97,7 @@
|
||||
<property name="defaultConnectionTimeoutInSeconds" value="${canal.instance.network.soTimeout:30}" />
|
||||
|
||||
<!-- 解析编码 -->
|
||||
<!-- property name="connectionCharsetNumber" value="${canal.instance.connectionCharsetNumber:33}" /-->
|
||||
<property name="connectionCharset" value="${canal.instance.connectionCharset:UTF-8}" />
|
||||
|
||||
<!-- 解析位点记录 -->
|
||||
|
||||
@@ -33,6 +33,7 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
|
||||
protected TableMetaTSDB tableMetaTSDB;
|
||||
|
||||
// 编码信息
|
||||
protected byte connectionCharsetNumber = (byte) 33;
|
||||
protected Charset connectionCharset = Charset.forName("UTF-8");
|
||||
protected boolean filterQueryDcl = false;
|
||||
protected boolean filterQueryDml = false;
|
||||
@@ -204,6 +205,10 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
|
||||
|
||||
// ============================ setter / getter =========================
|
||||
|
||||
public void setConnectionCharsetNumber(byte connectionCharsetNumber) {
|
||||
this.connectionCharsetNumber = connectionCharsetNumber;
|
||||
}
|
||||
|
||||
public void setConnectionCharsetStd(Charset connectionCharset) {
|
||||
this.connectionCharset = connectionCharset;
|
||||
}
|
||||
|
||||
@@ -116,6 +116,7 @@ public class LocalBinlogEventParser extends AbstractMysqlEventParser implements
|
||||
MysqlConnection connection = new MysqlConnection(runningInfo.getAddress(),
|
||||
runningInfo.getUsername(),
|
||||
runningInfo.getPassword(),
|
||||
connectionCharsetNumber,
|
||||
runningInfo.getDefaultDatabaseName());
|
||||
connection.getConnector().setReceiveBufferSize(64 * 1024);
|
||||
connection.getConnector().setSendBufferSize(64 * 1024);
|
||||
|
||||
@@ -27,7 +27,6 @@ import com.alibaba.otter.canal.parse.driver.mysql.packets.client.RegisterSlaveCo
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.packets.client.SemiAckCommandPacket;
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ErrorPacket;
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ResultSetPacket;
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo;
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager;
|
||||
import com.alibaba.otter.canal.parse.exception.CanalParseException;
|
||||
import com.alibaba.otter.canal.parse.inbound.ErosaConnection;
|
||||
@@ -74,27 +73,14 @@ public class MysqlConnection implements ErosaConnection {
|
||||
connector.setConnTimeout(connTimeout);
|
||||
}
|
||||
|
||||
public MysqlConnection(InetSocketAddress address, String username, String password, String defaultSchema){
|
||||
public MysqlConnection(InetSocketAddress address, String username, String password, byte charsetNumber,
|
||||
String defaultSchema){
|
||||
authInfo = new AuthenticationInfo();
|
||||
authInfo.setAddress(address);
|
||||
authInfo.setUsername(username);
|
||||
authInfo.setPassword(password);
|
||||
authInfo.setDefaultDatabaseName(defaultSchema);
|
||||
connector = new MysqlConnector(address, username, password, defaultSchema);
|
||||
// 将connection里面的参数透传下
|
||||
connector.setSoTimeout(soTimeout);
|
||||
connector.setConnTimeout(connTimeout);
|
||||
}
|
||||
|
||||
public MysqlConnection(InetSocketAddress address, String username, String password, String defaultSchema,
|
||||
SslInfo sslInfo){
|
||||
authInfo = new AuthenticationInfo();
|
||||
authInfo.setAddress(address);
|
||||
authInfo.setUsername(username);
|
||||
authInfo.setPassword(password);
|
||||
authInfo.setDefaultDatabaseName(defaultSchema);
|
||||
authInfo.setSslInfo(sslInfo);
|
||||
connector = new MysqlConnector(address, username, password, defaultSchema, sslInfo);
|
||||
connector = new MysqlConnector(address, username, password, charsetNumber, defaultSchema);
|
||||
// 将connection里面的参数透传下
|
||||
connector.setSoTimeout(soTimeout);
|
||||
connector.setConnTimeout(connTimeout);
|
||||
@@ -603,32 +589,6 @@ public class MysqlConnection implements ErosaConnection {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* MySQL 8.4版本开始部分命令出现变化
|
||||
* https://dev.mysql.com/doc/relnotes/mysql/8.4/en/news-8-4-0.html#mysqld-8-4-0-deprecation-removal
|
||||
*
|
||||
* @param major
|
||||
* @param minor
|
||||
* @return
|
||||
*/
|
||||
public boolean atLeast(int major, int minor) {
|
||||
if (isMariaDB()) {
|
||||
return false;
|
||||
}
|
||||
String version = connector.getServerVersion();
|
||||
if (StringUtils.isNotEmpty(version)) {
|
||||
String[] parts = version.split("\\.");
|
||||
int majorVer = Integer.parseInt(parts[0]);
|
||||
int minorVer = Integer.parseInt(parts[1]);
|
||||
return (majorVer > major) || (majorVer == major && minorVer >= minor);
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
public boolean atLeastMySQL84() {
|
||||
return atLeast(8, 4);
|
||||
}
|
||||
|
||||
private void accumulateReceivedBytes(long x) {
|
||||
if (receivedBinlogBytes != null) {
|
||||
receivedBinlogBytes.addAndGet(x);
|
||||
@@ -653,7 +613,7 @@ public class MysqlConnection implements ErosaConnection {
|
||||
|
||||
private String value;
|
||||
|
||||
private BinlogFormat(String value) {
|
||||
private BinlogFormat(String value){
|
||||
this.value = value;
|
||||
}
|
||||
|
||||
@@ -693,7 +653,7 @@ public class MysqlConnection implements ErosaConnection {
|
||||
|
||||
private String value;
|
||||
|
||||
private BinlogImage(String value) {
|
||||
private BinlogImage(String value){
|
||||
this.value = value;
|
||||
}
|
||||
|
||||
|
||||
@@ -305,8 +305,8 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
|
||||
MysqlConnection connection = new MysqlConnection(runningInfo.getAddress(),
|
||||
runningInfo.getUsername(),
|
||||
runningInfo.getPassword(),
|
||||
runningInfo.getDefaultDatabaseName(),
|
||||
runningInfo.getSslInfo());
|
||||
connectionCharsetNumber,
|
||||
runningInfo.getDefaultDatabaseName());
|
||||
connection.getConnector().setReceiveBufferSize(receiveBufferSize);
|
||||
connection.getConnector().setSendBufferSize(sendBufferSize);
|
||||
connection.getConnector().setSoTimeout(defaultConnectionTimeoutInSeconds * 1000);
|
||||
@@ -668,20 +668,13 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
|
||||
* 查询当前的binlog位置
|
||||
*/
|
||||
private EntryPosition findEndPosition(MysqlConnection mysqlConnection) {
|
||||
String showSql = "show master status";
|
||||
try {
|
||||
if (mysqlConnection.atLeastMySQL84()) {
|
||||
// 8.4新语法
|
||||
showSql = "show binary log status";
|
||||
} else if (multiStreamEnable) {
|
||||
// 兼容polardb-x的多流binlog
|
||||
showSql = "show master status with " + destination;
|
||||
}
|
||||
String showSql = multiStreamEnable ? "show master status with " + destination : "show master status";
|
||||
ResultSetPacket packet = mysqlConnection.query(showSql);
|
||||
List<String> fields = packet.getFieldValues();
|
||||
if (CollectionUtils.isEmpty(fields)) {
|
||||
throw new CanalParseException(
|
||||
"command : '" + showSql + "' has an error! pls check. you need (at least one of) the SUPER,REPLICATION CLIENT privilege(s) for this operation");
|
||||
"command : 'show master status' has an error! pls check. you need (at least one of) the SUPER,REPLICATION CLIENT privilege(s) for this operation");
|
||||
}
|
||||
EntryPosition endPosition = new EntryPosition(fields.get(0), Long.valueOf(fields.get(1)));
|
||||
if (isGTIDMode() && fields.size() > 4) {
|
||||
@@ -697,7 +690,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
|
||||
}
|
||||
return endPosition;
|
||||
} catch (IOException e) {
|
||||
throw new CanalParseException("command : '" + showSql + "' has an error!", e);
|
||||
throw new CanalParseException("command : 'show master status' has an error!", e);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -728,12 +721,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
|
||||
@SuppressWarnings("unused")
|
||||
private SlaveEntryPosition findSlavePosition(MysqlConnection mysqlConnection) {
|
||||
try {
|
||||
String showSql = "show slave stauts";
|
||||
if (mysqlConnection.atLeastMySQL84()) {
|
||||
// 兼容mysql 8.4
|
||||
showSql = "show replica status";
|
||||
}
|
||||
ResultSetPacket packet = mysqlConnection.query(showSql);
|
||||
ResultSetPacket packet = mysqlConnection.query("show slave status");
|
||||
List<FieldPacket> names = packet.getFieldDescriptors();
|
||||
List<String> fields = packet.getFieldValues();
|
||||
if (CollectionUtils.isEmpty(fields)) {
|
||||
|
||||
@@ -10,8 +10,8 @@ public class DdlResult {
|
||||
|
||||
private String schemaName;
|
||||
private String tableName;
|
||||
private String oriSchemaName; // rename ddl中的源表
|
||||
private String oriTableName; // rename ddl中的目标表
|
||||
private String oriSchemaName; // rename ddl中的源表
|
||||
private String oriTableName; // rename ddl中的目标表
|
||||
private EventType type;
|
||||
private DdlResult renameTableResult; // 多个rename table的存储
|
||||
|
||||
@@ -102,8 +102,7 @@ public class DdlResult {
|
||||
DdlResult ddlResult = this;
|
||||
StringBuilder sb = new StringBuilder();
|
||||
do {
|
||||
sb.append(String.format(
|
||||
"DdlResult [schemaName=%s , tableName=%s , oriSchemaName=%s , oriTableName=%s , type=%s ];",
|
||||
sb.append(String.format("DdlResult [schemaName=%s , tableName=%s , oriSchemaName=%s , oriTableName=%s , type=%s ];",
|
||||
ddlResult.schemaName,
|
||||
ddlResult.tableName,
|
||||
ddlResult.oriSchemaName,
|
||||
|
||||
@@ -4,18 +4,39 @@ import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
|
||||
import org.apache.commons.lang.StringUtils;
|
||||
|
||||
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
|
||||
import com.alibaba.polardbx.druid.sql.SQLUtils;
|
||||
import com.alibaba.polardbx.druid.sql.ast.SQLExpr;
|
||||
import com.alibaba.polardbx.druid.sql.ast.SQLStatement;
|
||||
import com.alibaba.polardbx.druid.sql.ast.expr.SQLIdentifierExpr;
|
||||
import com.alibaba.polardbx.druid.sql.ast.expr.SQLPropertyExpr;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.*;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableAddConstraint;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableAddIndex;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableDropConstraint;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableDropIndex;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableDropKey;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableItem;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableRename;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableStatement;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLConstraint;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLCreateDatabaseStatement;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLCreateIndexStatement;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLCreateTableStatement;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLDeleteStatement;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLDropDatabaseStatement;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLDropIndexStatement;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLDropTableStatement;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLExprTableSource;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLInsertStatement;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLTableSource;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLTruncateStatement;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLUnique;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLUpdateStatement;
|
||||
import com.alibaba.polardbx.druid.sql.dialect.mysql.ast.statement.MySqlRenameTableStatement;
|
||||
import com.alibaba.polardbx.druid.sql.parser.ParserException;
|
||||
import com.alibaba.polardbx.druid.util.JdbcConstants;
|
||||
import org.apache.commons.lang.StringUtils;
|
||||
|
||||
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
|
||||
|
||||
/**
|
||||
* @author agapple 2017年7月27日 下午4:05:34
|
||||
|
||||
@@ -51,6 +51,7 @@ public class RdsBinlogEventParserProxy extends MysqlEventParser {
|
||||
rdsLocalBinlogEventParser.setDestination(destination);
|
||||
rdsLocalBinlogEventParser.setAlarmHandler(this.getAlarmHandler());
|
||||
rdsLocalBinlogEventParser.setConnectionCharsetStd(this.connectionCharset);
|
||||
rdsLocalBinlogEventParser.setConnectionCharsetNumber(this.connectionCharsetNumber);
|
||||
rdsLocalBinlogEventParser.setEnableTsdb(this.enableTsdb);
|
||||
rdsLocalBinlogEventParser.setEventBlackFilter(this.eventBlackFilter);
|
||||
rdsLocalBinlogEventParser.setFilterQueryDcl(this.filterQueryDcl);
|
||||
|
||||
@@ -195,7 +195,7 @@ public abstract class AbstractRequest<T> {
|
||||
private final HttpResponse executeHttpRequest(HttpGet getMethod, String host) throws Exception {
|
||||
SSLContext sslContext = SSLContexts.custom().loadTrustMaterial(null, (TrustStrategy) (arg0, arg1) -> true).build();
|
||||
SSLConnectionSocketFactory sslsf = new SSLConnectionSocketFactory(sslContext,
|
||||
new String[] { "TLSv1.2" },
|
||||
new String[] { "TLSv1" },
|
||||
null,
|
||||
SSLConnectionSocketFactory.ALLOW_ALL_HOSTNAME_VERIFIER);
|
||||
Registry registry = RegistryBuilder.create()
|
||||
|
||||
@@ -207,11 +207,6 @@ public class DatabaseTableMeta implements TableMetaTSDB {
|
||||
}
|
||||
|
||||
for (String schema : schemas) {
|
||||
// 如果schema命中黑名单，直接跳过
|
||||
// 解决部分数据库可以看到database，但实际表级别无权限的情况
|
||||
if (blackFilter != null && blackFilter.filter(schema + ".%")) {
|
||||
continue;
|
||||
}
|
||||
// filter views
|
||||
packet = connection.query("show full tables from `" + schema + "` where Table_type = 'BASE TABLE'");
|
||||
columnSize = packet.getFieldDescriptors().size();
|
||||
|
||||
@@ -3,22 +3,26 @@ package com.alibaba.otter.canal.parse.inbound.mysql.tsdb;
|
||||
import java.util.*;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
import org.apache.commons.lang.StringUtils;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import com.alibaba.otter.canal.parse.inbound.TableMeta;
|
||||
import com.alibaba.otter.canal.parse.inbound.TableMeta.FieldMeta;
|
||||
import com.alibaba.otter.canal.parse.inbound.mysql.ddl.DruidDdlParser;
|
||||
import com.alibaba.otter.canal.protocol.position.EntryPosition;
|
||||
import com.alibaba.polardbx.druid.DbType;
|
||||
import com.alibaba.polardbx.druid.sql.SQLUtils;
|
||||
import com.alibaba.polardbx.druid.sql.ast.SQLDataType;
|
||||
import com.alibaba.polardbx.druid.sql.ast.SQLDataTypeImpl;
|
||||
import com.alibaba.polardbx.druid.sql.ast.SQLExpr;
|
||||
import com.alibaba.polardbx.druid.sql.ast.SQLStatement;
|
||||
import com.alibaba.polardbx.druid.sql.ast.expr.*;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.*;
|
||||
import com.alibaba.polardbx.druid.sql.ast.expr.SQLCharExpr;
|
||||
import com.alibaba.polardbx.druid.sql.ast.expr.SQLIdentifierExpr;
|
||||
import com.alibaba.polardbx.druid.sql.ast.expr.SQLMethodInvokeExpr;
|
||||
import com.alibaba.polardbx.druid.sql.ast.expr.SQLNullExpr;
|
||||
import com.alibaba.polardbx.druid.sql.ast.expr.SQLPropertyExpr;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLColumnConstraint;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLColumnDefinition;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLColumnPrimaryKey;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLColumnUniqueKey;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLCreateTableStatement;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLNotNullConstraint;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLNullConstraint;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLSelectOrderByItem;
|
||||
import com.alibaba.polardbx.druid.sql.ast.statement.SQLTableElement;
|
||||
import com.alibaba.polardbx.druid.sql.dialect.mysql.ast.MySqlPrimaryKey;
|
||||
import com.alibaba.polardbx.druid.sql.dialect.mysql.ast.MySqlUnique;
|
||||
import com.alibaba.polardbx.druid.sql.dialect.mysql.ast.expr.MySqlOrderingExpr;
|
||||
@@ -28,6 +32,13 @@ import com.alibaba.polardbx.druid.sql.repository.SchemaRepository;
|
||||
import com.alibaba.polardbx.druid.sql.visitor.SQLASTOutputVisitor;
|
||||
import com.alibaba.polardbx.druid.sql.visitor.VisitorFeature;
|
||||
import com.alibaba.polardbx.druid.util.JdbcConstants;
|
||||
import org.apache.commons.lang.StringUtils;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import com.alibaba.otter.canal.parse.inbound.TableMeta;
|
||||
import com.alibaba.otter.canal.parse.inbound.TableMeta.FieldMeta;
|
||||
import com.alibaba.otter.canal.parse.inbound.mysql.ddl.DruidDdlParser;
|
||||
import com.alibaba.otter.canal.protocol.position.EntryPosition;
|
||||
|
||||
/**
|
||||
* 基于DDL维护的内存表结构
|
||||
@@ -39,7 +50,7 @@ public class MemoryTableMeta implements TableMetaTSDB {
|
||||
|
||||
private Logger logger = LoggerFactory.getLogger(MemoryTableMeta.class);
|
||||
private Map<List<String>, TableMeta> tableMetas = new ConcurrentHashMap<>();
|
||||
private SchemaRepository repository;
|
||||
private SchemaRepository repository;
|
||||
|
||||
public MemoryTableMeta(){
|
||||
repository = new SchemaRepository(JdbcConstants.MYSQL);
|
||||
|
||||
@@ -30,9 +30,6 @@ public class TableMetaTSDBBuilder {
|
||||
if (applicationContext == null) {
|
||||
synchronized (contexts) {
|
||||
if (applicationContext == null) {
|
||||
if (!StringUtils.startsWithIgnoreCase(springXml, "classpath:")) {
|
||||
springXml = "classpath:" + springXml;
|
||||
}
|
||||
applicationContext = new ClassPathXmlApplicationContext(springXml);
|
||||
contexts.put(destination, applicationContext);
|
||||
}
|
||||
|
||||
@@ -2,12 +2,10 @@ package com.alibaba.otter.canal.parse.support;
|
||||
|
||||
import java.net.InetSocketAddress;
|
||||
|
||||
import com.alibaba.otter.canal.common.utils.CommonUtils;
|
||||
import org.apache.commons.lang.builder.ToStringBuilder;
|
||||
import org.apache.commons.lang.builder.ToStringStyle;
|
||||
|
||||
import com.alibaba.otter.canal.common.utils.CommonUtils;
|
||||
import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo;
|
||||
|
||||
/**
|
||||
* 数据库认证信息
|
||||
*
|
||||
@@ -16,15 +14,16 @@ import com.alibaba.otter.canal.parse.driver.mysql.ssl.SslInfo;
|
||||
*/
|
||||
public class AuthenticationInfo {
|
||||
|
||||
|
||||
|
||||
private InetSocketAddress address; // 主库信息
|
||||
private String username; // 帐号
|
||||
private String password; // 密码
|
||||
private String defaultDatabaseName;// 默认链接的数据库
|
||||
private String pwdPublicKey; // 公钥
|
||||
private boolean enableDruid; // 是否使用druid加密解密数据库密码
|
||||
private SslInfo sslInfo;
|
||||
private String pwdPublicKey; //公钥
|
||||
private boolean enableDruid; //是否使用druid加密解密数据库密码
|
||||
|
||||
public void initPwd() throws Exception {
|
||||
public void initPwd() throws Exception{
|
||||
if (enableDruid) {
|
||||
this.password = CommonUtils.decryptDruidPassword(pwdPublicKey, password);
|
||||
}
|
||||
@@ -38,19 +37,13 @@ public class AuthenticationInfo {
|
||||
this(address, username, password, "");
|
||||
}
|
||||
|
||||
private AuthenticationInfo(InetSocketAddress address, String username, String password, String defaultDatabaseName){
|
||||
public AuthenticationInfo(InetSocketAddress address, String username, String password, String defaultDatabaseName){
|
||||
this.address = address;
|
||||
this.username = username;
|
||||
this.password = password;
|
||||
this.defaultDatabaseName = defaultDatabaseName;
|
||||
}
|
||||
|
||||
public AuthenticationInfo(InetSocketAddress address, String username, String password, String defaultDatabaseName,
|
||||
SslInfo sslInfo){
|
||||
this(address, username, password, defaultDatabaseName);
|
||||
this.sslInfo = sslInfo;
|
||||
}
|
||||
|
||||
public InetSocketAddress getAddress() {
|
||||
return address;
|
||||
}
|
||||
@@ -99,14 +92,6 @@ public class AuthenticationInfo {
|
||||
this.enableDruid = enableDruid;
|
||||
}
|
||||
|
||||
public SslInfo getSslInfo() {
|
||||
return sslInfo;
|
||||
}
|
||||
|
||||
public void setSslInfo(SslInfo sslInfo) {
|
||||
this.sslInfo = sslInfo;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return ToStringBuilder.reflectionToString(this, ToStringStyle.DEFAULT_STYLE);
|
||||
|
||||
6
pom.xml
6
pom.xml
@@ -99,13 +99,13 @@
|
||||
<java_target_version>1.8</java_target_version>
|
||||
<file_encoding>UTF-8</file_encoding>
|
||||
<javadoc_skip>true</javadoc_skip>
|
||||
<spring_version>5.3.26</spring_version>
|
||||
<spring_version>5.3.9</spring_version>
|
||||
<log4j_version>2.17.0</log4j_version>
|
||||
<rocketmq_version>4.9.8</rocketmq_version>
|
||||
<rocketmq_version>4.8.0</rocketmq_version>
|
||||
<rabbitmq_version>5.18.0</rabbitmq_version>
|
||||
<mq_amqp_client>1.0.3</mq_amqp_client>
|
||||
<kafka_version>2.4.0</kafka_version>
|
||||
<pulsar_version>2.11.4</pulsar_version>
|
||||
<pulsar_version>2.8.1</pulsar_version>
|
||||
<mysql_driver_version>5.1.48</mysql_driver_version>
|
||||
<maven-jacoco-plugin.version>0.8.3</maven-jacoco-plugin.version>
|
||||
<maven-surefire.version>2.22.1</maven-surefire.version>
|
||||
|
||||
@@ -253,6 +253,7 @@ public class CanalServerTest {
|
||||
|
||||
parameter.setDefaultConnectionTimeoutInSeconds(30);
|
||||
parameter.setConnectionCharset("UTF-8");
|
||||
parameter.setConnectionCharsetNumber((byte) 33);
|
||||
parameter.setReceiveBufferSize(8 * 1024);
|
||||
parameter.setSendBufferSize(8 * 1024);
|
||||
|
||||
|
||||
@@ -40,6 +40,7 @@ public class CanalServerWithEmbedded_FileModeTest extends BaseCanalServerWithEmb
|
||||
|
||||
parameter.setDefaultConnectionTimeoutInSeconds(30);
|
||||
parameter.setConnectionCharset("UTF-8");
|
||||
parameter.setConnectionCharsetNumber((byte) 33);
|
||||
parameter.setReceiveBufferSize(8 * 1024);
|
||||
parameter.setSendBufferSize(8 * 1024);
|
||||
|
||||
|
||||
@@ -43,6 +43,7 @@ public class CanalServerWithEmbedded_StandaloneTest extends BaseCanalServerWithE
|
||||
|
||||
parameter.setDefaultConnectionTimeoutInSeconds(30);
|
||||
parameter.setConnectionCharset("UTF-8");
|
||||
parameter.setConnectionCharsetNumber((byte) 33);
|
||||
parameter.setReceiveBufferSize(8 * 1024);
|
||||
parameter.setSendBufferSize(8 * 1024);
|
||||
|
||||
|
||||
@@ -55,6 +55,7 @@ public class CanalServerWithEmbedded_StandbyTest extends BaseCanalServerWithEmbe
|
||||
|
||||
parameter.setDefaultConnectionTimeoutInSeconds(30);
|
||||
parameter.setConnectionCharset("UTF-8");
|
||||
parameter.setConnectionCharsetNumber((byte) 33);
|
||||
parameter.setReceiveBufferSize(8 * 1024);
|
||||
parameter.setSendBufferSize(8 * 1024);
|
||||
|
||||
|
||||
Reference in New Issue
Block a user