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 #2286 , ArithmeticException zero

Browse Source
This commit is contained in:
agapple
2019-10-10 15:37:17 +08:00
parent 31562919bb
commit b64a3e301a
1 changed files with 20 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -249,20 +249,27 @@ public class CanalRocketMQProducer extends AbstractMQProducer implements CanalMQ
// 批量发送 // 批量发送
List<MessageQueue> queues = topicInfo.getMessageQueueList(); List<MessageQueue> queues = topicInfo.getMessageQueueList();
int size = queues.size(); int size = queues.size();
MessageQueue queue = null; if (size <= 0) {
if (partition > queues.size()) { // 可能是第一次创建
queue = queues.get(partition % size); for (Message message : messages) {
} else { sendMessage(message, partition);
queue = queues.get(partition); }
} } else {
MessageQueue queue = null;
try { if (partition > size) {
SendResult sendResult = this.defaultMQProducer.send(messages, queue); queue = queues.get(partition % size);
if (logger.isDebugEnabled()) { } else {
logger.debug("Send Message Result: {}", sendResult); queue = queues.get(partition);
}
try {
SendResult sendResult = this.defaultMQProducer.send(messages, queue);
if (logger.isDebugEnabled()) {
logger.debug("Send Message Result: {}", sendResult);
}
} catch (Throwable e) {
throw new RuntimeException(e);
} }
} catch (Throwable e) {
throw new RuntimeException(e);
} }
} }
} }