通过多个实例从一个文件夹中读取文件

Question

我有一个文件夹，上游系统会在其中转储为我的应用程序输入的文件。我的应用程序读取这些文件，处理它们，然后在某个目录中输出。由于输入文件的数量很大，我决定运行我的应用程序的多个实例，并且所有实例都将为输入文件读取相同的输入目录。现在，由于多个实例会读取同一个文件夹，如何防止多个实例同时读取同一个文件？

PS：我想过使用带有 NIO 锁的 Spring Integration，但是一旦你锁定了文件，即使同一个线程也将无法读取文件。 Camel 会处理锁定，但它会锁定整个目录，并且不会在 2 个实例试图获取尚未锁定的文件的那段时间内处理场景。

任何建议都会有所帮助。

Answer 1

Spring Integration 为此提供了FileSystemPersistentAcceptOnceFileListFilter。您可以使用共享的ConcurrentMetadataStore impl 对其进行配置，例如RedisMetadataStore 或 ZookeeperMetadataStore。

只有一个应用程序实例会接受一个文件。

虽然有一些限制，所有文件只能由一个实例应用：

private void scanInputDirectory() {
    List<File> filteredFiles = this.scanner.listFiles(this.directory);
    Set<File> freshFiles = new LinkedHashSet<File>(filteredFiles);
    if (!freshFiles.isEmpty()) {
        this.toBeReceived.addAll(freshFiles);
        if (logger.isDebugEnabled()) {
            logger.debug("Added to queue: " + freshFiles);
        }
    }
}

您可以通过分布式MessageChannel 克服单例问题，例如SubscribableJmsChannel、PublishSubscribeAmqpChannel 或任何其他基于MessageStore 的队列通道解决方案。

Answer 2

我不会这样做，因为读取具有多个实例的文件夹只会使您的流程复杂化。

我会改为有一个单一的路由来读取文件并将内容放入队列（可以是 jms、amqp 等等）以及文件名作为标题。然后您可以对该队列进行并行处理并对目标文件夹进行并行写入。如果您需要阅读其他文件夹，这将使您的解决方案更易于处理和扩展。

Answer 3

您可以为应用程序实例已读取的每个文件创建一个影子文件

为避免并发访问同一文件的问题，您需要将检查文件可读性的过程集中在一个线程或一个进程中，该线程/进程将同步并与所有其他应用程序实例进行通信，

例如：

• Instance1 需要检查 file1 之前是否已被读取 => 它会检查此文件是否有影子文件，如果 true 将与集中式线程/进程通信以查看是否有任何其他实例已在处理此文件李>
• 集中式线程将允许实例处理文件 1 并将其引用存储在池列表中，该池列表将用于防止任何其他实例处理此文件
• Instance1 已完成文件处理，它还将创建影子文件，一旦完成，它将与集中线程通信以从池列表中删除该文件的引用

每个实例在访问文件之前都需要执行两个检查级别：

• 检查文件是否有同名的影子文件（表示之前已经读取过）
• 如果第一个条件为真，则使用集中式进程检查文件当前是否正在由任何其他实例处理。