如何在Spring Integration中实现MongoDB批量插入操作

我们用Java DSL开发Spring Integration Flow，从远程SFTP读取CSV文件并将数据插入MongoDB。当前采用流式读取文件行的方式，需要实现MongoDB批量插入功能。查阅Spring Integration文档和示例后，未找到对应批量操作选项，也不确定如何实现预期行为。尝试过Aggregation，但没找到适合固定批量大小的解决方案。

现有代码示例 #

@Configuration
public class SampleConfiguration {

  ...

  @Bean
  MessagingTemplate messagingTemplate(ApplicationContext context) {
    MessagingTemplate messagingTemplate = new MessagingTemplate();
    messagingTemplate.setBeanFactory(context);
    return messagingTemplate;
  }

  @Bean
  IntegrationFlow sftpSource() {
    DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory();
    factory.setHost("localhost");
    factory.setPort(22);
    factory.setUser("foo");
    factory.setPassword("foo");
    factory.setAllowUnknownKeys(true);
    SftpRemoteFileTemplate template = new SftpRemoteFileTemplate(factory);
    return IntegrationFlow
        .from(Sftp.inboundStreamingAdapter(template, Comparator.comparing(DirEntry::getFilename))
                .remoteDirectory("upload")
                .patternFilter("*.csv")
                .maxFetchSize(1),
            spec -> spec.poller(Pollers.fixedRate(Duration.ofMillis(1000)))
                .autoStartup(true))
        .split(Files
            .splitter()
            .markers()
            .charset(StandardCharsets.UTF_8)
            .firstLineAsHeader("fileHeader")
            .applySequence(true))
        .filter(payload -> !(payload instanceof FileSplitter.FileMarker))
        .enrichHeaders(h -> h.errorChannel("errorChannel"))
        .handle((String payload, MessageHeaders headers) -> {
          String header = headers.get("fileHeader", String.class);
          String rowWithHeader = header + "\n" + payload;
          try (StringReader reader = new StringReader(rowWithHeader)) {
            CsvToBean<MyPojo> beanReader = new CsvToBeanBuilder<MyPojo>(reader)
                .withType(MyPojo.class)
                .withSeparator(';')
                .build();
            return beanReader.iterator().next();
          }
        })
        .handle(MongoDb
            .outboundGateway(mongoTemplate)
            .entityClass(MyPojo.class)
            .collectionNameFunction(m -> "mypojo")
            .collectionCallback(
                (collection, message) -> {
                  MyPojo myPojo = (MyPojo) message.getPayload();
                  Document document = new Document();
                  mongoTemplate.getConverter().write(myPojo, document);
                  return collection.insertOne(document);
                }))
        .channel("logChannel")
        .get();
  }

  @Bean
  IntegrationFlow logFiles() {
    return IntegrationFlow
        .from("logChannel")
        .handle(message -> log.info("message received: {}", message))
        .get();
  }

  @Bean
  IntegrationFlow logErrors() {
    return IntegrationFlow
        .from("errorChannel")
        .handle(message -> {
          MessagingException exception = (MessagingException) message.getPayload();
          log.error("error message received: {} for message {}", exception.getMessage(),
              exception.getFailedMessage());
        })
        .get();
  }
  ...
}

已尝试的聚合步骤 #

.aggregate(aggregatorSpec -> aggregatorSpec
            .correlationStrategy(message -> message
                .getHeaders()
                .get(FileHeaders.REMOTE_FILE))
            .releaseStrategy(group -> group.size() >= 5)
            .groupTimeout(200L)
            .sendPartialResultOnExpiry(true))
.handle(MongoDb...)

解决方案 #

调整聚合器与MongoDB批量插入逻辑 #

你的聚合思路是对的，但需要调整MongoDB处理环节，接收聚合后的List<MyPojo>并执行批量插入：

1. 完善聚合配置：确保聚合器按文件分组（通过FileHeaders.REMOTE_FILE关联），并按固定大小或超时释放分组。
2. 修改MongoDB处理器：将原来处理单条数据的逻辑改为处理批量数据，使用insertMany替代insertOne。

修改后的关键代码片段：

.aggregate(aggregatorSpec -> aggregatorSpec
        .correlationStrategy(message -> message.getHeaders().get(FileHeaders.REMOTE_FILE))
        .releaseStrategy(group -> group.size() >= 5) // 每5条数据批量插入
        .groupTimeout(200L) // 超时未达到批量阈值也触发插入
        .sendPartialResultOnExpiry(true)
        .outputProcessor(group -> group.getMessages()
                .stream()
                .map(msg -> (MyPojo) msg.getPayload())
                .collect(Collectors.toList()))) // 将聚合后的消息转换为MyPojo列表
.handle(MongoDb.outboundGateway(mongoTemplate)
        .entityClass(MyPojo.class)
        .collectionNameFunction(m -> "mypojo")
        .collectionCallback((collection, message) -> {
            List<MyPojo> myPojoList = (List<MyPojo>) message.getPayload();
            List<Document> documents = myPojoList.stream()
                    .map(pojo -> {
                        Document doc = new Document();
                        mongoTemplate.getConverter().write(pojo, doc);
                        return doc;
                    })
                    .collect(Collectors.toList());
            return collection.insertMany(documents); // 执行批量插入
        }))

关键说明 #

• 聚合器输出处理：通过outputProcessor将聚合后的消息组转换为List<MyPojo>，简化后续MongoDB处理逻辑。
• 批量插入逻辑：将列表中的每个MyPojo转换为Document，调用insertMany完成批量操作，提升插入效率。
• 分组策略：基于FileHeaders.REMOTE_FILE确保同一文件的数据被聚合在一起，避免跨文件批量插入。

内容的提问来源于stack exchange，提问作者tommaso.normani