Spring Integration RecursiveDirectoryScanner 给出太多文件打开异常

Question

我正在使用 Spring Integration RecursiveDirectoryScanner 递归地扫描目录以处理将放置在配置目录 (/home/test) 下的传入文件。

我经常收到以下错误：

    ERROR org.springframework.integration.handler.LoggingHandler - java.lang.IllegalArgumentException: java.nio.file.FileSystemException: /home/test: Too many open files
        at org.springframework.integration.file.RecursiveDirectoryScanner.listFiles(RecursiveDirectoryScanner.java:89)
        at org.springframework.integration.file.FileReadingMessageSource.scanInputDirectory(FileReadingMessageSource.java:387)
        at org.springframework.integration.file.FileReadingMessageSource.doReceive(FileReadingMessageSource.java:361)
        at org.springframework.integration.file.FileReadingMessageSource.doReceive(FileReadingMessageSource.java:90)
        at org.springframework.integration.endpoint.AbstractMessageSource.receive(AbstractMessageSource.java:134)
        at org.springframework.integration.endpoint.SourcePollingChannelAdapter.receiveMessage(SourcePollingChannelAdapter.java:224)
        at org.springframework.integration.endpoint.AbstractPollingEndpoint.doPoll(AbstractPollingEndpoint.java:245)
        at org.springframework.integration.endpoint.AbstractPollingEndpoint.access$000(AbstractPollingEndpoint.java:58)
        at org.springframework.integration.endpoint.AbstractPollingEndpoint$1.call(AbstractPollingEndpoint.java:190)
        at org.springframework.integration.endpoint.AbstractPollingEndpoint$1.call(AbstractPollingEndpoint.java:186)
        at org.springframework.integration.endpoint.AbstractPollingEndpoint$Poller$1.run(AbstractPollingEndpoint.java:353)
        at org.springframework.integration.util.ErrorHandlingTaskExecutor$1.run(ErrorHandlingTaskExecutor.java:55)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)

Caused by: java.nio.file.FileSystemException: /home/test: Too many open files
        at sun.nio.fs.UnixException.translateToIOException(UnixException.java:91)
        at sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:102)
        at sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:107)
        at sun.nio.fs.UnixFileSystemProvider.newDirectoryStream(UnixFileSystemProvider.java:427)
        at java.nio.file.Files.newDirectoryStream(Files.java:457)
        at java.nio.file.FileTreeWalker.visit(FileTreeWalker.java:300)
        at java.nio.file.FileTreeWalker.walk(FileTreeWalker.java:322)
        at java.nio.file.FileTreeIterator.<init>(FileTreeIterator.java:72)
        at java.nio.file.Files.walk(Files.java:3574)
        at org.springframework.integration.file.RecursiveDirectoryScanner.listFiles(RecursiveDirectoryScanner.java:73)

我的 Spring 集成流程如下：

XML 中的配置

 <task:executor id="pollerPool"
        pool-size="${pollerThreadPoolSize}"
        queue-capacity="${pollerThreadQueueCapacity}" rejection-policy="ABORT" />

    <task:executor id="fileHandlerPool"
        pool-size="${fileHandlerPoolSize}"
        queue-capacity="${fileHandlerPoolThreadQueueCapacity}" rejection-policy="CALLER_RUNS" />


     <bean id="iFilter" class="org.springframework.integration.file.filters.ChainFileListFilter">
        <constructor-arg>
             <list>
                  <bean id="lastModifiedFileListFilter" class="org.springframework.integration.file.filters.LastModifiedFileListFilter">
                     <property name="age" value="120" />
                  </bean>
                 <ref bean="acceptOnceFileListFilter"/>

                 <bean class="org.springframework.integration.file.filters.RegexPatternFileListFilter">
                    <constructor-arg value="^.*\.(txt|csv|xls|xlsx|asc)$"/>
                 </bean>

            </list>
        </constructor-arg>
      </bean>

     <bean id="acceptOnceFileListFilter" name="acceptOnceFileListFilter" class="org.springframework.integration.file.filters.AcceptOnceFileListFilter" primary="true" />

     <bean id="recursiveDirectoryScanner" class="org.springframework.integration.file.RecursiveDirectoryScanner">
        <property name="filter" ref="iFilter" />
        <property name="locker" ref="nioFileLocker" />
     </bean>

    <bean id="nioFileLocker" class="org.springframework.integration.file.locking.NioFileLocker" />

    <int-file:inbound-channel-adapter
        id="fileSource" channel="fileReceivedChannel" auto-startup="true"
        directory="file:${polling.directory}" 
        scanner="recursiveDirectoryScanner"  >
        <int:poller task-executor="pollerPool"
            fixed-rate="${pollerFixedRate}"
            receive-timeout="${pollerReceiveTimeout}">
        </int:poller>
    </int-file:inbound-channel-adapter>

动态参数如下：

• polling.directory=/home/test pollerThreadPoolSize=1 pollerThreadQueueCapacity=10 pollerFixedRate=5000 轮询接收超时=5000 文件处理程序池大小=2 fileHandlerPoolThreadQueueCapacity=100

编辑：

我确实在选择文件时出现在图片中的服务激活器中解锁文件。我从文件中获取一些信息并解锁它。

@Autowired
  NioFileLocker nioFileLocker;

   protected void doTransform(Message<?> message) throws Exception {
  MessageBuilder<File> payload = (MessageBuilder<File>) message.getPayload();
    File inFile = payload.getPayload();
   try {
      nioFileLocker.unlock(inFile);
    } catch (Exception e) {
      LOGGER.error("file not unlock");
    }
 }

配置有问题吗？如何确保此异常不再出现？

提前谢谢你。

Answer 1

我建议在没有NioFileLocker 的情况下测试您的解决方案。看起来您并没有使用它来解锁文件，但lock(File fileToLock) 确实在操作系统中保留了一些文件标记。

另一方面，文件锁在 UNIX 系统上不可靠。它仍然允许访问文件。至少是为了阅读。

为了更好的独占文件访问，我建议使用FileSystemPersistentAcceptOnceFileListFilter 和外部MetadataStore，而不是内存中的AcceptOnceFileListFilter。这样一来，您的应用程序只有一个实例可以访问该文件，并且根本不会再次处理该文件。