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

fixed issue #2258 , optimizer rabbitmq send perf & matchTopics only match one Topic

Browse Source
This commit is contained in:
agapple
2019-09-30 20:42:52 +08:00
parent d505b5ab9d
commit dd93e2b0c4
2 changed files with 30 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -565,9 +565,13 @@ public class MQMessageUtils {
String topicConfigs = item.substring(i + 1).trim(); String topicConfigs = item.substring(i + 1).trim();
if (matchDynamicTopic(name, topicConfigs)) { if (matchDynamicTopic(name, topicConfigs)) {
topics.add(topic); topics.add(topic);
// 匹配了一个就退出
break;
} }
} else if (matchDynamicTopic(name, item)) { } else if (matchDynamicTopic(name, item)) {
// 匹配了一个就退出
topics.add(name.toLowerCase()); topics.add(name.toLowerCase());
break;
} }
} }
return topics.isEmpty() ? null : topics; return topics.isEmpty() ? null : topics;

29
server/src/main/java/com/alibaba/otter/canal/rabbitmq/CanalRabbitMQProducer.java
View File

@@ -16,6 +16,7 @@ import com.alibaba.otter.canal.common.CanalMessageSerializer;
import com.alibaba.otter.canal.common.MQMessageUtils; import com.alibaba.otter.canal.common.MQMessageUtils;
import com.alibaba.otter.canal.common.MQMessageUtils.EntryRowData; import com.alibaba.otter.canal.common.MQMessageUtils.EntryRowData;
import com.alibaba.otter.canal.common.MQProperties; import com.alibaba.otter.canal.common.MQProperties;
import com.alibaba.otter.canal.common.utils.ExecutorTemplate;
import com.alibaba.otter.canal.protocol.FlatMessage; import com.alibaba.otter.canal.protocol.FlatMessage;
import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.server.exception.CanalServerException; import com.alibaba.otter.canal.server.exception.CanalServerException;
@@ -57,8 +58,9 @@ public class CanalRabbitMQProducer extends AbstractMQProducer implements CanalMQ
} }
@Override @Override
public void send(MQProperties.CanalDestination canalDestination, Message message, Callback callback) public void send(final MQProperties.CanalDestination canalDestination, Message message, Callback callback)
throws IOException { throws IOException {
ExecutorTemplate template = new ExecutorTemplate(executor);
try { try {
if (!StringUtils.isEmpty(canalDestination.getDynamicTopic())) { if (!StringUtils.isEmpty(canalDestination.getDynamicTopic())) {
// 动态topic // 动态topic
@@ -67,21 +69,32 @@ public class CanalRabbitMQProducer extends AbstractMQProducer implements CanalMQ
canalDestination.getDynamicTopic()); canalDestination.getDynamicTopic());
for (Map.Entry<String, com.alibaba.otter.canal.protocol.Message> entry : messageMap.entrySet()) { for (Map.Entry<String, com.alibaba.otter.canal.protocol.Message> entry : messageMap.entrySet()) {
String topicName = entry.getKey().replace('.', '_'); final String topicName = entry.getKey().replace('.', '_');
com.alibaba.otter.canal.protocol.Message messageSub = entry.getValue(); final com.alibaba.otter.canal.protocol.Message messageSub = entry.getValue();
template.submit(new Runnable() {
@Override
public void run() {
send(canalDestination, topicName, messageSub); send(canalDestination, topicName, messageSub);
} }
});
}
template.waitForResult();
} else { } else {
send(canalDestination, canalDestination.getTopic(), message); send(canalDestination, canalDestination.getTopic(), message);
} }
callback.commit(); callback.commit();
} catch (Throwable e) { } catch (Throwable e) {
logger.error(e.getMessage(), e);
callback.rollback(); callback.rollback();
} finally {
template.clear();
} }
} }
private void send(MQProperties.CanalDestination canalDestination, String topicName, Message messageSub) private void send(MQProperties.CanalDestination canalDestination, String topicName, Message messageSub) {
throws Exception {
if (!mqProperties.getFlatMessage()) { if (!mqProperties.getFlatMessage()) {
byte[] message = CanalMessageSerializer.serializer(messageSub, mqProperties.isFilterTransactionEntry()); byte[] message = CanalMessageSerializer.serializer(messageSub, mqProperties.isFilterTransactionEntry());
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {
@@ -108,9 +121,13 @@ public class CanalRabbitMQProducer extends AbstractMQProducer implements CanalMQ
} }
private void sendMessage(String queueName, byte[] message) throws Exception { private void sendMessage(String queueName, byte[] message) {
// tips: 目前逻辑中暂不处理对exchange处理，请在Console后台绑定 才可使用routekey // tips: 目前逻辑中暂不处理对exchange处理，请在Console后台绑定 才可使用routekey
try {
channel.basicPublish(mqProperties.getExchange(), queueName, null, message); channel.basicPublish(mqProperties.getExchange(), queueName, null, message);
} catch (Throwable e) {
throw new RuntimeException(e);
}
} }
@Override @Override