Merge pull request #1054 from duhengforever/canal_rocketmq_flatmessage

[ISSUE1027]Add flat message support to rocketmq

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java

@@ -92,8 +92,10 @@ public class CanalAdapterLoader {
                     }
                     canalOuterAdapterGroups.add(canalOuterAdapters);
                     if (StringUtils.isBlank(topic.getMqMode()) || "rocketmq".equalsIgnoreCase(topic.getMqMode())) {
-                        CanalAdapterRocketMQWorker rocketMQWorker = new CanalAdapterRocketMQWorker(canalClientConfig
-                            .getBootstrapServers(), topic.getTopic(), group.getGroupId(), canalOuterAdapterGroups);
+                        CanalAdapterRocketMQWorker rocketMQWorker = new CanalAdapterRocketMQWorker(canalClientConfig.getBootstrapServers(),
+                            topic.getTopic(),
+                            group.getGroupId(),
+                            canalOuterAdapterGroups, canalClientConfig.getFlatMessage());
                         canalMQWorker.put(topic.getTopic() + "-rocketmq-" + group.getGroupId(), rocketMQWorker);
                         rocketMQWorker.start();
                     } else if ("kafka".equalsIgnoreCase(topic.getMqMode())) {

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRocketMQWorker.java

@@ -1,5 +1,6 @@
 package com.alibaba.otter.canal.adapter.launcher.loader;
 
+import com.alibaba.otter.canal.protocol.FlatMessage;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -14,25 +15,27 @@ import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnectorProvider;
 import com.alibaba.otter.canal.protocol.Message;
 
 /**
- * kafka对应的client适配器工作线程
+ * rocketmq对应的client适配器工作线程
  *
- * @author rewerma 2018-8-19 下午11:30:49
  * @version 1.0.0
  */
 public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
 
     private RocketMQCanalConnector connector;
 
-    private String                 topic;
+    private String topic;
+
+    private boolean flatMessage;
 
     public CanalAdapterRocketMQWorker(String nameServers, String topic, String groupId,
-                                      List<List<OuterAdapter>> canalOuterAdapters){
+        List<List<CanalOuterAdapter>> canalOuterAdapters, boolean flatMessage) {
         logger.info("RocketMQ consumer config topic:{}, nameServer:{}, groupId:{}", topic, nameServers, groupId);
         this.canalOuterAdapters = canalOuterAdapters;
         this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size());
         this.topic = topic;
+        this.flatMessage = flatMessage;
         this.canalDestination = topic;
-        connector = RocketMQCanalConnectorProvider.newRocketMQConnector(nameServers, topic, groupId);
+        connector = RocketMQCanalConnectorProvider.newRocketMQConnector(nameServers, topic, groupId, flatMessage);
     }
 
     @Override
@@ -54,29 +57,46 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
                 logger.info("=============> Subscribe topic: {} succeed<=============", this.topic);
                 while (running) {
                     try {
-                        // switcher.get(); //等待开关开启
-
-                        final Message message = connector.getWithoutAck(1);
+                        Object message = null;
+                        if (!flatMessage) {
+                            message = connector.getWithoutAck(1);
+                        } else {
+                            message = connector.getFlatMessageWithoutAck();
+                        }
                         if (message != null) {
-                            executor.submit(() -> {
-                                try {
-                                    if (logger.isDebugEnabled()) {
-                                        logger.debug("topic: {} batchId: {} batchSize: {} ",
-                                            topic,
-                                            message.getId(),
-                                            message.getEntries().size());
-                                    }
-                                    long begin = System.currentTimeMillis();
-                                    writeOut(message);
-                                    long now = System.currentTimeMillis();
-                                    if ((System.currentTimeMillis() - begin) > 5 * 60 * 1000) {
-                                        logger.error("topic: {} batchId {} elapsed time: {} ms",
-                                            topic,
-                                            message.getId(),
-                                            now - begin);
+                            final Object msg = message;
+                            executor.submit(new Runnable() {
+                                @Override
+                                public void run() {
+                                    try {
+                                        if (logger.isDebugEnabled()) {
+                                            logger.debug("topic: {} batchId: {} batchSize: {} ",
+                                                topic,
+                                                message.getId(),
+                                                message.getEntries().size());
+                                        }
+                                        if (msg != null) {
+                                            long begin = System.currentTimeMillis();
+                                            if (msg instanceof Message) {
+                                                Message receive = (Message) msg;
+                                                writeOut(receive, topic);
+                                                connector.ack(receive.getId());
+                                            } else {
+                                                FlatMessage receive = (FlatMessage) msg;
+                                                writeOut(receive);
+                                                connector.ack(receive.getId());
+                                            }
+                                            long now = System.currentTimeMillis();
+                                            if ((now - begin) > 5 * 60 * 1000) {
+                                                logger.error("topic: {} batchId {} elapsed time: {} ms",
+                                                    topic,
+                                                    message.getId(),
+                                                    now - begin);
+                                            }
+                                        }
+                                    } catch (Exception e) {
+                                        logger.error(e.getMessage(), e);
                                     }
-                                } catch (Exception e) {
-                                    logger.error(e.getMessage(), e);
                                 }
                                 connector.ack(message.getId());
                             });

client-launcher/src/main/resources/canal-client.yml

@@ -0,0 +1,24 @@
+#canalServerHost: 127.0.0.1:11111
+#zookeeperHosts: slave1:2181
+bootstrapServers: slave1:6667,slave2:6667 #or rocketmq nameservers:host1:9876;host2:9876
+flatMessage: false
+
+#canalInstances:
+#- instance: example
+#  adapterGroups:
+#  - outAdapters:
+#    - name: logger
+#    - name: hbase
+#      hosts: slave1:2181
+#      properties: {znodeParent: "/hbase-unsecure"}
+
+mqTopics:
+- mqMode: rocketmq
+  topic: example
+  groups:
+  - groupId: example
+    outAdapters:
+    - name: logger
+#    - name: hbase
+#      hosts: slave1:2181
+#      properties: {znodeParent: "/hbase-unsecure"}

client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java

@@ -1,5 +1,7 @@
 package com.alibaba.otter.canal.client.rocketmq;
 
+import com.alibaba.fastjson.JSON;
+import com.alibaba.otter.canal.protocol.FlatMessage;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.BlockingQueue;
@@ -24,21 +26,23 @@ import com.alibaba.otter.canal.protocol.exception.CanalClientException;
 
 public class RocketMQCanalConnector implements CanalConnector {
 
-    private static final Logger                          logger              = LoggerFactory.getLogger(RocketMQCanalConnector.class);
+    private static final Logger logger = LoggerFactory.getLogger(RocketMQCanalConnector.class);
 
-    private String                                       nameServer;
-    private String                                       topic;
-    private String                                       groupName;
-    private volatile boolean                             connected           = false;
-    private DefaultMQPushConsumer                        rocketMQConsumer;
-    private BlockingQueue<ConsumerBatchMessage<Message>> messageBlockingQueue;
-    Map<Long, ConsumerBatchMessage<Message>>             messageCache;
-    private long                                         batchProcessTimeout = 10000;
+    private String nameServer;
+    private String topic;
+    private String groupName;
+    private volatile boolean connected = false;
+    private DefaultMQPushConsumer rocketMQConsumer;
+    private BlockingQueue<ConsumerBatchMessage> messageBlockingQueue;
+    Map<Long, ConsumerBatchMessage> messageCache;
+    private long batchProcessTimeout = 3000;
+    private boolean flatMessage;
 
-    public RocketMQCanalConnector(String nameServer, String topic, String groupName){
+    public RocketMQCanalConnector(String nameServer, String topic, String groupName, boolean flatMessage) {
         this.nameServer = nameServer;
         this.topic = topic;
         this.groupName = groupName;
+        this.flatMessage = flatMessage;
         messageBlockingQueue = new LinkedBlockingQueue<>();
         messageCache = new ConcurrentHashMap<>();
     }
@@ -70,11 +74,12 @@ public class RocketMQCanalConnector implements CanalConnector {
             if (rocketMQConsumer == null) {
                 this.connect();
             }
-            rocketMQConsumer.subscribe(topic, "*");
+            rocketMQConsumer.subscribe(this.topic, "*");
             rocketMQConsumer.registerMessageListener(new MessageListenerOrderly() {
 
                 @Override
-                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> messageExts, ConsumeOrderlyContext context) {
+                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> messageExts,
+                    ConsumeOrderlyContext context) {
                     context.setAutoCommit(true);
                     boolean isSuccess = process(messageExts);
                     if (isSuccess) {
@@ -93,17 +98,33 @@ public class RocketMQCanalConnector implements CanalConnector {
     }
 
     private boolean process(List<MessageExt> messageExts) {
-        BlockingQueue<Message> messageList = new LinkedBlockingQueue<>();
+        logger.info("Get Message:{}", messageExts);
+        BlockingQueue messageList = new LinkedBlockingQueue<>();
         for (MessageExt messageExt : messageExts) {
             byte[] data = messageExt.getBody();
-            Message message = CanalMessageDeserializer.deserializer(data);
-            try {
-                messageList.put(message);
-            } catch (InterruptedException ex) {
-                logger.error("Add message error");
+            if (data != null){
+                try {
+                    if (!flatMessage) {
+                        Message message = CanalMessageDeserializer.deserializer(data);
+                        messageList.put(message);
+                    } else {
+                        FlatMessage flatMessage = JSON.parseObject(data, FlatMessage.class);
+                        messageList.put(flatMessage);
+                    }
+                } catch (Exception ex) {
+                    logger.error("Add message error", ex);
+                    throw new CanalClientException(ex);
+                }
+            }else{
+                logger.warn("Received message data is null");
             }
         }
-        ConsumerBatchMessage<Message> batchMessage = new ConsumerBatchMessage<>(messageList);
+        ConsumerBatchMessage batchMessage;
+        if (!flatMessage) {
+            batchMessage = new ConsumerBatchMessage<Message>(messageList);
+        } else {
+            batchMessage = new ConsumerBatchMessage<FlatMessage>(messageList);
+        }
         try {
             messageBlockingQueue.put(batchMessage);
         } catch (InterruptedException e) {
@@ -131,17 +152,28 @@ public class RocketMQCanalConnector implements CanalConnector {
         this.rocketMQConsumer.unsubscribe(this.topic);
     }
 
+    /**
+     * 暂时不支持batchSize 参数
+     *
+     * @param batchSize 暂时不支持
+     * @return
+     * @throws CanalClientException
+     */
     @Override
     public Message get(int batchSize) throws CanalClientException {
         Message message = getWithoutAck(batchSize);
-        ack(message.getId());
+        if (message != null) {
+            ack(message.getId());
+        }
         return message;
     }
 
     @Override
     public Message get(int batchSize, Long timeout, TimeUnit unit) throws CanalClientException {
         Message message = getWithoutAck(batchSize, timeout, unit);
-        ack(message.getId());
+        if (message != null) {
+            ack(message.getId());
+        }
         return message;
     }
 
@@ -155,6 +187,23 @@ public class RocketMQCanalConnector implements CanalConnector {
         return null;
     }
 
+    private FlatMessage getFlatMessage(ConsumerBatchMessage consumerBatchMessage) {
+        BlockingQueue<FlatMessage> messageList = consumerBatchMessage.getData();
+        if (messageList != null & messageList.size() > 0) {
+            FlatMessage message = messageList.poll();
+            messageCache.put(message.getId(), consumerBatchMessage);
+            return message;
+        }
+        return null;
+    }
+
+    /**
+     * 暂时不支持该参数设置
+     *
+     * @param batchSize
+     * @return
+     * @throws CanalClientException
+     */
     @Override
     public Message getWithoutAck(int batchSize) throws CanalClientException {
         ConsumerBatchMessage batchMessage = messageBlockingQueue.poll();
@@ -168,11 +217,51 @@ public class RocketMQCanalConnector implements CanalConnector {
     public Message getWithoutAck(int batchSize, Long timeout, TimeUnit unit) throws CanalClientException {
         try {
             ConsumerBatchMessage batchMessage = messageBlockingQueue.poll(timeout, unit);
-            return getMessage(batchMessage);
+            if (batchMessage != null) {
+                return getMessage(batchMessage);
+            }
         } catch (InterruptedException ex) {
             logger.warn("Get message timeout", ex);
             throw new CanalClientException("Failed to fetch the data after: " + timeout);
         }
+        return null;
+    }
+
+    public FlatMessage getFlatMessageWithoutAck() {
+        return getFlatMessageWithoutAck(null, null);
+    }
+
+    public FlatMessage getFlatMessageWithoutAck(Long timeout,
+        TimeUnit unit) throws CanalClientException {
+        try {
+            ConsumerBatchMessage batchMessage = null;
+            if (timeout == null || timeout == 0) {
+                batchMessage = messageBlockingQueue.poll();
+            } else {
+                batchMessage = messageBlockingQueue.poll(timeout, unit);
+            }
+            if (batchMessage != null) {
+                return getFlatMessage(batchMessage);
+            }
+        } catch (InterruptedException ex) {
+            logger.warn("Get flat message timeout", ex);
+            throw new CanalClientException("Failed to fetch the flat message data after: " + timeout);
+        }
+        return null;
+    }
+
+    public FlatMessage getFlatMessage() throws CanalClientException {
+        FlatMessage message = getFlatMessageWithoutAck(null, null);
+        if (message != null) {
+            ack(message.getId());
+        }
+        return message;
+    }
+
+    public FlatMessage getFlatMessage(Long timeout, TimeUnit unit) throws CanalClientException {
+        FlatMessage message = getFlatMessageWithoutAck(timeout, unit);
+        ack(message.getId());
+        return message;
     }
 
     @Override
@@ -180,6 +269,7 @@ public class RocketMQCanalConnector implements CanalConnector {
         ConsumerBatchMessage batchMessage = messageCache.get(batchId);
         if (batchMessage != null) {
             batchMessage.ack();
+            messageCache.remove(batchId);
         }
     }
 

client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnectorProvider.java

@@ -14,7 +14,11 @@ public class RocketMQCanalConnectorProvider {
      * @return {@link RocketMQCanalConnector}
      */
     public static RocketMQCanalConnector newRocketMQConnector(String nameServers, String topic, String groupId) {
-        return new RocketMQCanalConnector(nameServers, topic, groupId);
+        return new RocketMQCanalConnector(nameServers, topic, groupId, false);
     }
 
+    public static RocketMQCanalConnector newRocketMQConnector(String nameServers, String topic, String groupId,
+        boolean flatMessage) {
+        return new RocketMQCanalConnector(nameServers, topic, groupId, flatMessage);
+    }
 }

deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java

@@ -57,6 +57,7 @@ public class CanalLauncher {
 
             CanalMQProducer canalMQProducer = null;
             String serverMode = controller.getProperty(properties, CanalConstants.CANAL_SERVER_MODE);
+            serverMode = "rocketmq";
             if (serverMode.equalsIgnoreCase("kafka")) {
                 canalMQProducer = new CanalKafkaProducer();
             } else if (serverMode.equalsIgnoreCase("rocketmq")) {

deployer/src/main/resources/mq.yml

@@ -8,7 +8,7 @@ bufferMemory: 33554432
 canalBatchSize: 50
 # Canal get数据的超时时间, 单位: 毫秒, 空为不限超时
 canalGetTimeout: 100
-flatMessage: true
+flatMessage: false
 
 canalDestinations:
   - canalDestination: example

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

@@ -1,10 +1,13 @@
 package com.alibaba.otter.canal.rocketmq;
 
+import com.alibaba.fastjson.JSON;
 import com.alibaba.otter.canal.common.CanalMessageSerializer;
 import com.alibaba.otter.canal.common.MQProperties;
+import com.alibaba.otter.canal.protocol.FlatMessage;
 import com.alibaba.otter.canal.server.exception.CanalServerException;
 import com.alibaba.otter.canal.spi.CanalMQProducer;
 import java.util.List;
+import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.rocketmq.client.exception.MQBrokerException;
 import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.client.producer.DefaultMQProducer;
@@ -19,10 +22,13 @@ public class CanalRocketMQProducer implements CanalMQProducer {
 
     private static final Logger logger = LoggerFactory.getLogger(CanalRocketMQProducer.class);
 
-    private DefaultMQProducer   defaultMQProducer;
+    private DefaultMQProducer defaultMQProducer;
+
+    private MQProperties mqProperties;
 
     @Override
     public void init(MQProperties rocketMQProperties) {
+        this.mqProperties = rocketMQProperties;
         defaultMQProducer = new DefaultMQProducer();
         defaultMQProducer.setNamesrvAddr(rocketMQProperties.getServers());
         defaultMQProducer.setProducerGroup(rocketMQProperties.getProducerGroup());
@@ -37,25 +43,78 @@ public class CanalRocketMQProducer implements CanalMQProducer {
 
     @Override
     public void send(final MQProperties.CanalDestination destination, com.alibaba.otter.canal.protocol.Message data,
-                     Callback callback) {
-        try {
-            Message message = new Message(destination.getTopic(), CanalMessageSerializer.serializer(data));
-            this.defaultMQProducer.send(message, new MessageQueueSelector() {
-
-                @Override
-                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
-                    int partition = 0;
+        Callback callback) {
+        if (!mqProperties.getFlatMessage()) {
+            try {
+                Message message = new Message(destination.getTopic(), CanalMessageSerializer.serializer(data));
+                logger.debug("send message:{} to destination:{}, partition: {}", message, destination.getCanalDestination(), destination.getPartition());
+                this.defaultMQProducer.send(message, new MessageQueueSelector() {
+                    @Override
+                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
+                        int partition = 0;
+                        if (destination.getPartition() != null) {
+                            partition = destination.getPartition();
+                        }
+                        return mqs.get(partition);
+                    }
+                }, null);
+                callback.commit();
+            } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
+                logger.error("Send message error!", e);
+                callback.rollback();
+            }
+        } else {
+            List<FlatMessage> flatMessages = FlatMessage.messageConverter(data);
+            if (flatMessages != null) {
+                for (FlatMessage flatMessage : flatMessages) {
                     if (destination.getPartition() != null) {
-                        partition = destination.getPartition();
+                        try {
+                            logger.info("send flat message: {} to topic: {} fixed partition: {}", JSON.toJSONString(flatMessage),destination.getTopic(), destination.getPartition());
+                            Message message = new Message(destination.getTopic(), JSON.toJSONString(flatMessage).getBytes());
+                            this.defaultMQProducer.send(message, new MessageQueueSelector() {
+                                @Override
+                                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
+                                    return mqs.get(destination.getPartition());
+                                }
+                            }, null);
+                        } catch (Exception e) {
+                            logger.error("send flat message to fixed partition error", e);
+                            callback.rollback();
+                        }
+                    } else {
+                        if (destination.getPartitionHash() != null
+                            && !destination.getPartitionHash().isEmpty()) {
+                            FlatMessage[] partitionFlatMessage = FlatMessage.messagePartition(flatMessage,
+                                destination.getPartitionsNum(),
+                                destination.getPartitionHash());
+                            int length = partitionFlatMessage.length;
+                            for (int i = 0; i < length; i++) {
+                                FlatMessage flatMessagePart = partitionFlatMessage[i];
+                                logger.debug("flatMessagePart: {}, partition: {}", JSON.toJSONString(flatMessagePart), i);
+                                final int index = i;
+                                try {
+                                    Message message = new Message(destination.getTopic(), JSON.toJSONString(flatMessagePart).getBytes());
+                                    this.defaultMQProducer.send(message, new MessageQueueSelector() {
+                                        @Override
+                                        public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
+                                            if (index > mqs.size()) {
+                                                throw new CanalServerException("partition number is error,config num:" + destination.getPartitionsNum() + ", mq num: " + mqs.size());
+                                            }
+                                            return mqs.get(index);
+                                        }
+                                    }, null);
+                                } catch (Exception e) {
+                                    logger.error("send flat message to hashed partition error", e);
+                                    callback.rollback();
+                                }
+
+                            }
+                        }
                     }
-                    return mqs.get(partition);
                 }
-            }, null);
-            callback.commit();
-        } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
-            logger.error("Send message error!", e);
-            callback.rollback();
+            }
         }
+
     }
 
     @Override