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

Merge pull request #1419 from rewerma/master

Browse Source 
specified mq dynamic topic
This commit is contained in:
agapple
2019-01-16 10:27:25 +08:00
committed by GitHub
parent 7f9dce4b77 739f3b4fa1
commit e4177ab0ce
6 changed files with 182 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java
View File

@@ -73,10 +73,10 @@ public class RdbSyncService {
this.executorThreads = new ExecutorService[this.threads];
for (int i = 0; i < this.threads; i++) {
dmlsPartition[i] = new ArrayList<>();
batchExecutors[i] = new BatchExecutor(dataSource.getConnection());
batchExecutors[i] = new BatchExecutor(dataSource);
executorThreads[i] = Executors.newSingleThreadExecutor();
}
} catch (SQLException e) {
} catch (Exception e) {
throw new RuntimeException(e);
}
}
@@ -122,6 +122,12 @@ public class RdbSyncService {
}
});
}
for (BatchExecutor batchExecutor : batchExecutors) {
if (batchExecutor != null) {
batchExecutor.close();
}
}
}
/**
@@ -189,7 +195,7 @@ public class RdbSyncService {
} else if (type != null && type.equalsIgnoreCase("DELETE")) {
delete(batchExecutor, config, dml);
} else if (type != null && type.equalsIgnoreCase("TRUNCATE")) {
truncate(batchExecutor, config, dml);
truncate(batchExecutor, config);
}
if (logger.isDebugEnabled()) {
logger.debug("DML: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue));
@@ -290,15 +296,16 @@ public class RdbSyncService {
StringBuilder updateSql = new StringBuilder();
updateSql.append("UPDATE ").append(SyncUtil.getDbTableName(dbMapping)).append(" SET ");
List<Map<String, ?>> values = new ArrayList<>();
boolean hasMatched = false;
for (String srcColumnName : old.keySet()) {
List<String> targetColumnNames = new ArrayList<>();
columnsMap.forEach((targetColumn, srcColumn) -> {
if (srcColumnName.toLowerCase().equals(srcColumn)) {
if (srcColumnName.toLowerCase().equals(srcColumn.toLowerCase())) {
targetColumnNames.add(targetColumn);
}
});
if (!targetColumnNames.isEmpty()) {
hasMatched = true;
for (String targetColumnName : targetColumnNames) {
updateSql.append(targetColumnName).append("=?, ");
Integer type = ctype.get(Util.cleanColumn(targetColumnName).toLowerCase());
@@ -309,6 +316,10 @@ public class RdbSyncService {
}
}
}
if (!hasMatched) {
logger.warn("Did not matched any columns to update ");
return;
}
int len = updateSql.length();
updateSql.delete(len - 2, len).append(" WHERE ");
@@ -353,15 +364,13 @@ public class RdbSyncService {
*
* @param config
*/
private void truncate(BatchExecutor batchExecutor, MappingConfig config, SingleDml dml) throws SQLException {
if (dml.getIsTruncate()) {
DbMapping dbMapping = config.getDbMapping();
StringBuilder sql = new StringBuilder();
sql.append("TRUNCATE TABLE ").append(SyncUtil.getDbTableName(dbMapping));
batchExecutor.execute(sql.toString(), new ArrayList<>());
if (logger.isTraceEnabled()) {
logger.trace("Truncate target table, sql: {}", sql);
}
private void truncate(BatchExecutor batchExecutor, MappingConfig config) throws SQLException {
DbMapping dbMapping = config.getDbMapping();
StringBuilder sql = new StringBuilder();
sql.append("TRUNCATE TABLE ").append(SyncUtil.getDbTableName(dbMapping));
batchExecutor.execute(sql.toString(), new ArrayList<>());
if (logger.isTraceEnabled()) {
logger.trace("Truncate target table, sql: {}", sql);
}
}
@@ -477,7 +486,6 @@ public class RdbSyncService {
public void close() {
for (int i = 0; i < threads; i++) {
batchExecutors[i].close();
executorThreads[i].shutdown();
}
}

37
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/BatchExecutor.java
View File

@@ -9,22 +9,38 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import javax.sql.DataSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* sql批量执行器
*
* @author rewerma 2018-11-7 下午06:45:49
* @version 1.0.0
*/
public class BatchExecutor implements Closeable {
private static final Logger logger = LoggerFactory.getLogger(BatchExecutor.class);
private DataSource dataSource;
private Connection conn;
private AtomicInteger idx = new AtomicInteger(0);
public BatchExecutor(Connection conn) throws SQLException{
this.conn = conn;
this.conn.setAutoCommit(false);
public BatchExecutor(DataSource dataSource){
this.dataSource = dataSource;
}
public Connection getConn() {
if (conn == null) {
try {
conn = dataSource.getConnection();
this.conn.setAutoCommit(false);
} catch (SQLException e) {
logger.error(e.getMessage(), e);
}
}
return conn;
}
@@ -36,7 +52,7 @@ public class BatchExecutor implements Closeable {
}
public void execute(String sql, List<Map<String, ?>> values) throws SQLException {
PreparedStatement pstmt = conn.prepareStatement(sql);
PreparedStatement pstmt = getConn().prepareStatement(sql);
int len = values.size();
for (int i = 0; i < len; i++) {
int type = (Integer) values.get(i).get("type");
@@ -49,7 +65,7 @@ public class BatchExecutor implements Closeable {
}
public void commit() throws SQLException {
conn.commit();
getConn().commit();
if (logger.isTraceEnabled()) {
logger.trace("Batch executor commit " + idx.get() + " rows");
}
@@ -57,7 +73,7 @@ public class BatchExecutor implements Closeable {
}
public void rollback() throws SQLException {
conn.rollback();
getConn().rollback();
if (logger.isTraceEnabled()) {
logger.trace("Batch executor rollback " + idx.get() + " rows");
}
@@ -68,10 +84,11 @@ public class BatchExecutor implements Closeable {
public void close() {
if (conn != null) {
try {
if (conn != null) {
conn.close();
}
} catch (SQLException ioe) {
conn.close();
} catch (SQLException e) {
logger.error(e.getMessage(), e);
} finally {
conn = null;
}
}
}

14
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/SingleDml.java
View File

@@ -14,7 +14,6 @@ public class SingleDml {
private String type;
private Map<String, Object> data;
private Map<String, Object> old;
private boolean isTruncate = false;
public String getDestination() {
return destination;
@@ -64,18 +63,10 @@ public class SingleDml {
this.old = old;
}
public boolean getIsTruncate() {
return isTruncate;
}
public void setIsTruncate(boolean isTruncate) {
this.isTruncate = isTruncate;
}
public static List<SingleDml> dml2SingleDmls(Dml dml) {
int size = dml.getData() == null ? 0 : dml.getData().size();
List<SingleDml> singleDmls = new ArrayList<>();
if (size > 0) {
if (dml.getData() != null) {
int size = dml.getData().size();
for (int i = 0; i < size; i++) {
SingleDml singleDml = new SingleDml();
singleDml.setDestination(dml.getDestination());
@@ -94,7 +85,6 @@ public class SingleDml {
singleDml.setDatabase(dml.getDatabase());
singleDml.setTable(dml.getTable());
singleDml.setType(dml.getType());
singleDml.setIsTruncate(true);
singleDmls.add(singleDml);
}
return singleDmls;

115
server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java
View File

@@ -1,12 +1,6 @@
package com.alibaba.otter.canal.common;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.*;
import org.apache.commons.lang.StringUtils;
@@ -30,36 +24,46 @@ import com.google.protobuf.InvalidProtocolBufferException;
public class MQMessageUtils {
@SuppressWarnings("deprecation")
private static Map<String, List<PartitionData>> partitionDatas = MigrateMap.makeComputingMap(new MapMaker().softValues(),
new Function<String, List<PartitionData>>() {
private static Map<String, List<PartitionData>> partitionDatas = MigrateMap
.makeComputingMap(new MapMaker().softValues(), new Function<String, List<PartitionData>>() {
public List<PartitionData> apply(String pkHashConfigs) {
List<PartitionData> datas = Lists.newArrayList();
String[] pkHashConfigArray = StringUtils.split(pkHashConfigs,
",");
List<PartitionData> datas = Lists
.newArrayList();
String[] pkHashConfigArray = StringUtils
.split(pkHashConfigs, ",");
// schema.table:id^name
for (String pkHashConfig : pkHashConfigArray) {
PartitionData data = new PartitionData();
int i = pkHashConfig.lastIndexOf(":");
int i = pkHashConfig
.lastIndexOf(":");
if (i > 0) {
String pkStr = pkHashConfig.substring(i + 1);
if (pkStr.equalsIgnoreCase("$pk$")) {
String pkStr = pkHashConfig
.substring(i + 1);
if (pkStr.equalsIgnoreCase(
"$pk$")) {
data.hashMode.autoPkHash = true;
} else {
data.hashMode.pkNames = Lists.newArrayList(StringUtils.split(pkStr,
'^'));
data.hashMode.pkNames = Lists
.newArrayList(
StringUtils
.split(
pkStr,
'^'));
}
pkHashConfig = pkHashConfig.substring(0,
i);
pkHashConfig = pkHashConfig
.substring(0, i);
} else {
data.hashMode.tableHash = true;
}
if (!isWildCard(pkHashConfig)) {
if (!isWildCard(
pkHashConfig)) {
data.simpleName = pkHashConfig;
} else {
data.regexFilter = new AviaterRegexFilter(pkHashConfig);
data.regexFilter = new AviaterRegexFilter(
pkHashConfig);
}
datas.add(data);
}
@@ -69,24 +73,29 @@ public class MQMessageUtils {
});
@SuppressWarnings("deprecation")
private static Map<String, List<DynamicTopicData>> dynamicTopicDatas = MigrateMap.makeComputingMap(new MapMaker().softValues(),
new Function<String, List<DynamicTopicData>>() {
private static Map<String, List<DynamicTopicData>> dynamicTopicDatas = MigrateMap
.makeComputingMap(new MapMaker().softValues(), new Function<String, List<DynamicTopicData>>() {
public List<DynamicTopicData> apply(String pkHashConfigs) {
List<DynamicTopicData> datas = Lists.newArrayList();
String[] dynamicTopicArray = StringUtils.split(pkHashConfigs,
",");
List<DynamicTopicData> datas = Lists
.newArrayList();
String[] dynamicTopicArray = StringUtils
.split(pkHashConfigs, ",");
// schema.table
for (String dynamicTopic : dynamicTopicArray) {
DynamicTopicData data = new DynamicTopicData();
if (!isWildCard(dynamicTopic)) {
if (!isWildCard(
dynamicTopic)) {
data.simpleName = dynamicTopic;
} else {
if (dynamicTopic.contains("\\.")) {
data.tableRegexFilter = new AviaterRegexFilter(dynamicTopic);
if (dynamicTopic
.contains("\\.")) {
data.tableRegexFilter = new AviaterRegexFilter(
dynamicTopic);
} else {
data.schemaRegexFilter = new AviaterRegexFilter(dynamicTopic);
data.schemaRegexFilter = new AviaterRegexFilter(
dynamicTopic);
}
}
datas.add(data);
@@ -134,15 +143,22 @@ public class MQMessageUtils {
if (StringUtils.isEmpty(schemaName) || StringUtils.isEmpty(tableName)) {
put2MapMessage(messages, message.getId(), defaultTopic, entry);
} else {
if (matchDynamicTopic(schemaName + "." + tableName, dynamicTopicConfigs)) {
put2MapMessage(messages, message.getId(), schemaName + "." + tableName, entry);
} else if (matchDynamicTopic(schemaName, dynamicTopicConfigs)) {
put2MapMessage(messages, message.getId(), schemaName, entry);
Set<String> topics = matchTopics(schemaName + "." + tableName, dynamicTopicConfigs);
if (topics != null) {
for (String topic : topics) {
put2MapMessage(messages, message.getId(), topic, entry);
}
} else {
put2MapMessage(messages, message.getId(), defaultTopic, entry);
topics = matchTopics(schemaName, dynamicTopicConfigs);
if (topics != null) {
for (String topic : topics) {
put2MapMessage(messages, message.getId(), topic, entry);
}
} else {
put2MapMessage(messages, message.getId(), defaultTopic, entry);
}
}
}
}
return messages;
}
@@ -275,8 +291,9 @@ public class MQMessageUtils {
try {
rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
} catch (Exception e) {
throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:"
+ entry.toString(), e);
throw new RuntimeException(
"ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
e);
}
CanalEntry.EventType eventType = rowChange.getEventType();
@@ -480,6 +497,24 @@ public class MQMessageUtils {
return null;
}
private static Set<String> matchTopics(String name, String dynamicTopicConfigs) {
String[] router = StringUtils.split(dynamicTopicConfigs, ';');
Set<String> topics = new HashSet<>();
for (String item : router) {
int i = item.indexOf(":");
if (i > -1) {
String topic = item.substring(0, i).trim();
String topicConfigs = item.substring(i + 1).trim();
if (matchDynamicTopic(name, topicConfigs)) {
topics.add(topic);
}
} else if (matchDynamicTopic(name, item)) {
topics.add(name);
}
}
return topics.isEmpty() ? null : topics;
}
public static boolean matchDynamicTopic(String name, String dynamicTopicConfigs) {
if (StringUtils.isEmpty(dynamicTopicConfigs)) {
return false;
@@ -520,8 +555,8 @@ public class MQMessageUtils {
private static boolean isWildCard(String value) {
// not contaiins '.' ?
return StringUtils.containsAny(value, new char[] { '*', '?', '+', '|', '(', ')', '{', '}', '[', ']', '\\', '$',
'^' });
return StringUtils.containsAny(value,
new char[] { '*', '?', '+', '|', '(', ')', '{', '}', '[', ']', '\\', '$', '^' });
}
private static void put2MapMessage(Map<String, Message> messageMap, Long messageId, String topicName,

59
server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java
View File

@@ -79,27 +79,31 @@ public class CanalKafkaProducer implements CanalMQProducer {
@Override
public void send(MQProperties.CanalDestination canalDestination, Message message, Callback callback) {
try {
if (!StringUtils.isEmpty(canalDestination.getDynamicTopic())) {
// 动态topic
Map<String, Message> messageMap = MQMessageUtils
.messageTopics(message, canalDestination.getTopic(), canalDestination.getDynamicTopic());
if (!StringUtils.isEmpty(canalDestination.getDynamicTopic())) {
// 动态topic
Map<String, Message> messageMap = MQMessageUtils
.messageTopics(message, canalDestination.getTopic(), canalDestination.getDynamicTopic());
for (Map.Entry<String, Message> entry : messageMap.entrySet()) {
String topicName = entry.getKey().replace('.', '_');
Message messageSub = entry.getValue();
if (logger.isDebugEnabled()) {
logger.debug("## Send message to kafka topic: " + topicName);
for (Map.Entry<String, Message> entry : messageMap.entrySet()) {
String topicName = entry.getKey().replace('.', '_');
Message messageSub = entry.getValue();
if (logger.isDebugEnabled()) {
logger.debug("## Send message to kafka topic: " + topicName);
}
send(canalDestination, topicName, messageSub);
}
send(canalDestination, topicName, messageSub, callback);
} else {
send(canalDestination, canalDestination.getTopic(), message);
}
} else {
send(canalDestination, canalDestination.getTopic(), message, callback);
callback.commit();
} catch (Exception e) {
callback.rollback();
}
}
private void send(MQProperties.CanalDestination canalDestination, String topicName, Message message,
Callback callback) {
private void send(MQProperties.CanalDestination canalDestination, String topicName,
Message message) throws Exception {
// producer.beginTransaction();
if (!kafkaProperties.getFlatMessage()) {
try {
@@ -134,14 +138,16 @@ public class CanalKafkaProducer implements CanalMQProducer {
} catch (Exception e) {
logger.error(e.getMessage(), e);
// producer.abortTransaction();
callback.rollback();
return;
throw e;
}
} else {
// 发送扁平数据json
List<FlatMessage> flatMessages = MQMessageUtils.messageConverter(message);
if (flatMessages != null) {
int idx = 0;
int size = flatMessages.size();
for (FlatMessage flatMessage : flatMessages) {
idx++;
if (StringUtils.isEmpty(canalDestination.getPartitionHash())) {
try {
Integer partition = canalDestination.getPartition();
@@ -152,12 +158,15 @@ public class CanalKafkaProducer implements CanalMQProducer {
partition,
null,
JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue));
producer2.send(record);
if (idx != size) {
producer2.send(record);
} else {
producer2.send(record).get();
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
// producer.abortTransaction();
callback.rollback();
return;
throw e;
}
} else {
FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage,
@@ -173,12 +182,15 @@ public class CanalKafkaProducer implements CanalMQProducer {
i,
null,
JSON.toJSONString(flatMessagePart, SerializerFeature.WriteMapNullValue));
producer2.send(record);
if (idx != size) {
producer2.send(record);
} else {
producer2.send(record).get();
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
// producer.abortTransaction();
callback.rollback();
return;
throw e;
}
}
}
@@ -193,7 +205,6 @@ public class CanalKafkaProducer implements CanalMQProducer {
}
// producer.commitTransaction();
callback.commit();
}
}

40
server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java
View File

@@ -61,23 +61,28 @@ public class CanalRocketMQProducer implements CanalMQProducer {
@Override
public void send(final MQProperties.CanalDestination destination, com.alibaba.otter.canal.protocol.Message data,
Callback callback) {
if (!StringUtils.isEmpty(destination.getDynamicTopic())) {
// 动态topic
Map<String, com.alibaba.otter.canal.protocol.Message> messageMap = MQMessageUtils
.messageTopics(data, destination.getTopic(), destination.getDynamicTopic());
try {
if (!StringUtils.isEmpty(destination.getDynamicTopic())) {
// 动态topic
Map<String, com.alibaba.otter.canal.protocol.Message> messageMap = MQMessageUtils
.messageTopics(data, destination.getTopic(), destination.getDynamicTopic());
for (Map.Entry<String, com.alibaba.otter.canal.protocol.Message> entry : messageMap.entrySet()) {
String topicName = entry.getKey().replace('.', '_');
com.alibaba.otter.canal.protocol.Message messageSub = entry.getValue();
send(destination, topicName, messageSub, callback);
for (Map.Entry<String, com.alibaba.otter.canal.protocol.Message> entry : messageMap.entrySet()) {
String topicName = entry.getKey().replace('.', '_');
com.alibaba.otter.canal.protocol.Message messageSub = entry.getValue();
send(destination, topicName, messageSub);
}
} else {
send(destination, destination.getTopic(), data);
}
} else {
send(destination, destination.getTopic(), data, callback);
callback.commit();
} catch (Exception e) {
callback.rollback();
}
}
public void send(final MQProperties.CanalDestination destination, String topicName,
com.alibaba.otter.canal.protocol.Message data, Callback callback) {
com.alibaba.otter.canal.protocol.Message data) throws Exception {
if (!mqProperties.getFlatMessage()) {
try {
if (destination.getPartition() != null) {
@@ -132,8 +137,7 @@ public class CanalRocketMQProducer implements CanalMQProducer {
}, null);
} catch (Exception e) {
logger.error("send flat message to hashed partition error", e);
callback.rollback();
return;
throw e;
}
}
}
@@ -141,8 +145,7 @@ public class CanalRocketMQProducer implements CanalMQProducer {
}
} catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
logger.error("Send message error!", e);
callback.rollback();
return;
throw e;
}
} else {
List<FlatMessage> flatMessages = MQMessageUtils.messageConverter(data);
@@ -167,8 +170,7 @@ public class CanalRocketMQProducer implements CanalMQProducer {
}, null);
} catch (Exception e) {
logger.error("send flat message to fixed partition error", e);
callback.rollback();
return;
throw e;
}
} else {
if (destination.getPartitionHash() != null && !destination.getPartitionHash().isEmpty()) {
@@ -205,8 +207,7 @@ public class CanalRocketMQProducer implements CanalMQProducer {
}, null);
} catch (Exception e) {
logger.error("send flat message to hashed partition error", e);
callback.rollback();
return;
throw e;
}
}
}
@@ -216,7 +217,6 @@ public class CanalRocketMQProducer implements CanalMQProducer {
}
}
callback.commit();
if (logger.isDebugEnabled()) {
logger.debug("send message to rocket topic: {}", destination.getTopic());
}