From dd93e2b0c4b4479e052da2c63d3065a2ad5108b2 Mon Sep 17 00:00:00 2001
From: agapple <jianghang115@gmail.com>
Date: Mon, 30 Sep 2019 20:42:52 +0800
Subject: [PATCH] fixed issue #2258 , optimizer rabbitmq send perf &
 matchTopics only match one Topic

---
 .../otter/canal/common/MQMessageUtils.java    |  4 +++
 .../canal/rabbitmq/CanalRabbitMQProducer.java | 35 ++++++++++++++-----
 2 files changed, 30 insertions(+), 9 deletions(-)

diff --git a/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java b/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java
index 5299f7e9..ae966260 100644
--- a/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java
+++ b/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java
@@ -565,9 +565,13 @@ public class MQMessageUtils {
                 String topicConfigs = item.substring(i + 1).trim();
                 if (matchDynamicTopic(name, topicConfigs)) {
                     topics.add(topic);
+                    // 匹配了一个就退出
+                    break;
                 }
             } else if (matchDynamicTopic(name, item)) {
+                // 匹配了一个就退出
                 topics.add(name.toLowerCase());
+                break;
             }
         }
         return topics.isEmpty() ? null : topics;
diff --git a/server/src/main/java/com/alibaba/otter/canal/rabbitmq/CanalRabbitMQProducer.java b/server/src/main/java/com/alibaba/otter/canal/rabbitmq/CanalRabbitMQProducer.java
index c8fd9552..8cf82caa 100644
--- a/server/src/main/java/com/alibaba/otter/canal/rabbitmq/CanalRabbitMQProducer.java
+++ b/server/src/main/java/com/alibaba/otter/canal/rabbitmq/CanalRabbitMQProducer.java
@@ -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.EntryRowData;
 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.Message;
 import com.alibaba.otter.canal.server.exception.CanalServerException;
@@ -57,8 +58,9 @@ public class CanalRabbitMQProducer extends AbstractMQProducer implements CanalMQ
     }
 
     @Override
-    public void send(MQProperties.CanalDestination canalDestination, Message message, Callback callback)
-                                                                                                        throws IOException {
+    public void send(final MQProperties.CanalDestination canalDestination, Message message, Callback callback)
+                                                                                                              throws IOException {
+        ExecutorTemplate template = new ExecutorTemplate(executor);
         try {
             if (!StringUtils.isEmpty(canalDestination.getDynamicTopic())) {
                 // 动态topic
@@ -67,21 +69,32 @@ public class CanalRabbitMQProducer extends AbstractMQProducer implements CanalMQ
                     canalDestination.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(canalDestination, topicName, messageSub);
+                    final String topicName = entry.getKey().replace('.', '_');
+                    final com.alibaba.otter.canal.protocol.Message messageSub = entry.getValue();
+
+                    template.submit(new Runnable() {
+
+                        @Override
+                        public void run() {
+                            send(canalDestination, topicName, messageSub);
+                        }
+                    });
                 }
+
+                template.waitForResult();
             } else {
                 send(canalDestination, canalDestination.getTopic(), message);
             }
             callback.commit();
         } catch (Throwable e) {
+            logger.error(e.getMessage(), e);
             callback.rollback();
+        } finally {
+            template.clear();
         }
     }
 
-    private void send(MQProperties.CanalDestination canalDestination, String topicName, Message messageSub)
-                                                                                                           throws Exception {
+    private void send(MQProperties.CanalDestination canalDestination, String topicName, Message messageSub) {
         if (!mqProperties.getFlatMessage()) {
             byte[] message = CanalMessageSerializer.serializer(messageSub, mqProperties.isFilterTransactionEntry());
             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
-        channel.basicPublish(mqProperties.getExchange(), queueName, null, message);
+        try {
+            channel.basicPublish(mqProperties.getExchange(), queueName, null, message);
+        } catch (Throwable e) {
+            throw new RuntimeException(e);
+        }
     }
 
     @Override