canal.mq.partitionsNum 配置后，rocketmq的queue都发了消息，现象好

原因应该找到了

// 并发构造 MQMessageUtils.EntryRowData[] datas = MQMessageUtils.buildMessageData(message, buildExecutor); // 串行分区 List flatMessages = MQMessageUtils.messageConverter(datas, message.getId()); // 初始化分区合并队列 if (destination.getPartitionHash() != null && !destination.getPartitionHash().isEmpty()) { List partitionFlatMessages = new ArrayList<>(); for (int i = 0; i < destination.getPartitionsNum(); i++) { partitionFlatMessages.add(new ArrayList<>()); }

        for (FlatMessage flatMessage : flatMessages) {
            FlatMessage[] partitionFlatMessage = MQMessageUtils.messagePartition(flatMessage,
                partitionNum,
                destination.getPartitionHash(),
                mqProperties.isDatabaseHash());
            int length = partitionFlatMessage.length;
            for (int i = 0; i < length; i++) {
                partitionFlatMessages.get(i).add(partitionFlatMessage[i]);
            }
        }

        ExecutorTemplate template = new ExecutorTemplate(sendExecutor);
        for (int i = 0; i < partitionFlatMessages.size(); i++) {
            final List<FlatMessage> flatMessagePart = partitionFlatMessages.get(i);
            **if (flatMessagePart != null) {**
                final int index = i;
                template.submit(() -> {
                    List<Message> messages = flatMessagePart.stream()
                        .map(flatMessage -> new Message(topicName, JSON.toJSONBytes(flatMessage,
                            SerializerFeature.WriteMapNullValue)))
                        .collect(Collectors.toList());
                    // 批量发送
                    sendMessage(messages, index);
                });
            }
        }

源码中 if (flatMessagePart != null) 这个判断 有问题；；这个是不可能为null的；调试后，[null]；（有值，只是值为null而已）

这个是上面的2个for循环导致的

改了源码，修复了此bug了 改动的地方

for (int i = 0; i < length; i++) { if (partitionFlatMessage[i] != null) {//增加null判断 partitionFlatMessages.get(i).add(partitionFlatMessage[i]); } }

改动的第二个地方

if (flatMessagePart != null && flatMessagePart.size() > 0) {//判断加上size要大于0 final int index = i; template.submit(() -> { List messages = flatMessagePart.stream() .map(flatMessage -> new Message(topicName, JSON.toJSONBytes(flatMessage, SerializerFeature.WriteMapNullValue))) .collect(Collectors.toList()); // 批量发送 sendMessage(messages, index); }); }

原回答者GitHub用户gujiachun