时间戳

首页 > 解决方案 > Spring集成文件拆分器锁

spring - Spring集成文件拆分器锁

问题描述

我有一个集成流程，它从名为“promo”的消息通道读取文件，然后拆分文件并对每个文件的特定行进行转换，然后由 ouboundAdapter 以追加模式逐行写入目录称为“生成”。然后，我有一个入站适配器，它使用固定延迟的轮询器从“生成”目录中读取，并将文件移动到目标位置以由不同的应用程序选择。如何确保从第二个流程中挑选的文件是完整的？我看到的情况是入站适配器移动到仍然由第一个流写入的目标文件。可以将锁添加到正在写入“生成”目录的文件中，以便第二个流程？

  private IntegrationFlow promoChannelSubscriber(File generatedDir, GenericSelector storesFilter) {
    return IntegrationFlows
      .from("promo")
      .filter(storesFilter)
      .split(
        Files.splitter()
          .applySequence(true)
      )
      .transform(this::enrichXstoreFileHeader)
      .handle(
        Files.outboundAdapter(generatedDir)
          .autoCreateDirectory(true)
          .fileExistsMode(APPEND)
          .appendNewLine(true)
      )
      .get();
  }

  private IntegrationFlow movePromotions(File generatedDir, Path destinationDirectory) {
    return IntegrationFlows.from(
      Files.inboundAdapter(generatedDir, comparingLong(File::lastModified))
        .regexFilter(fileNamePattern),
      cas -> cas.poller(
        Pollers.fixedDelay(2, TimeUnit.SECONDS)
          .maxMessagesPerPoll(promoConfig.getMaxMessagesPerPoll())
          .errorChannel(errorChannelName)
      ))
      .handle(
        Files.outboundAdapter(destinationDirectory.toFile())
          .deleteSourceFiles(true)
          .temporaryFileSuffix(".tmp")
      )
      .get();
  }

谢谢

标签： springspring-integration

解决方案

为什么使用 2 个流程？为什么在收到最后一个拆分时不执行第二个流程？

如果必须使用 2 个流，则需要写入与第二个流的过滤器不匹配的临时文件名。

然后，当文件完成后，将文件重命名为其最终名称。

您可以使用标记消息和路由器来检测文件结束情况。

有关示例，请参见file-split-ftp 示例应用程序。

推荐阅读