使用 Spring Integration 从 FTP 获取文件答案

【问题标题】：Fetching file from FTP using Spring Integration使用 Spring Integration 从 FTP 获取文件
【发布时间】：2021-12-09 19:51:41
【问题描述】：

我有一个项目，我需要从远程 FTP 文件夹中获取 .csv 文件，问题是文件很大，可以说是 6-7 MB，即使尚未从第三个文件夹完全传输，投票也会开始读取它们党，这是一个例外。

我看到我们可以使用 LastModifiedFileListFilter 但不确定这是否是正确的解决方案。

这是我的代码示例。

@Bean
    public IntegrationFlow ftpInboundFlow() {
        return IntegrationFlows
                .from(Ftp.inboundAdapter(ftpSessionFactory())
                                .preserveTimestamp(true)
                                .remoteDirectory("/ftp/GE/Inbound")
                                .patternFilter("*.csv")
                                .deleteRemoteFiles(true)
                                .localDirectory(new File("inbound"))
                                .temporaryFileSuffix(TEMPORARY_FILE_SUFFIX),
                        e -> e.id("ftpInboundAdapter")
                                .poller(Pollers.fixedDelay(5000))
                                .autoStartup(true))
                .transform(e -> {
                    log.info("Sending CSV file " + e + " to FTP server");
                    return e;
                })
                .handle(Ftp.outboundAdapter(ftpSessionFactory())
                        .useTemporaryFileName(true)
                        .autoCreateDirectory(true)
                        .remoteDirectory("/ftp/GE/Inbound/history"))
                .get();
    }

例外：

Caused by: org.springframework.messaging.MessagingException: Failure occurred while copying '/ftp/GE/Inbound/OA_ex_PK_2020_2021.csv' from the remote to the local directory; nested exception is java.io.IOException: Failed to copy '/ftp/GE/Inbound/OA_ex_PK_2020_2021.csv'. Server replied with: 550 The process cannot access the file because it is being used by another process.

【问题讨论】：

标签： spring-integration spring-integration-dsl

【解决方案1】：

LastModifiedFileListFilter 用于本地文件系统。但是我认为一个想法很好，并且足以为 FTP 实现类似的“最后修改”。请参阅 FtpPersistentAcceptOnceFileListFilter 以及它如何从远程实体获取 modified 选项。

这样的FtpLastModifiedFileListFilter 必须是ChainFileListFilter 中的第一个，您必须提供到Ftp.inboundAdapter。第二个确实是你的new FtpSimplePatternFileListFilter("*.csv")。链中的最后一个必须是 AcceptOnceFileListFilter 或其 FTP 变体：FtpPersistentAcceptOnceFileListFilter。

FtpLastModifiedFileListFilter 的解决方案可以回馈给框架，因为您不是第一个要求类似解决方案的人。

另一种方法是通过tmp文件更改写入逻辑。因此，在这些 tmp 文件分别完全写入和重命名之前，您看不到最终文件。

另外一种解决方案就是忽略该异常，事情最终会自行解决。重点是FtpPersistentAcceptOnceFileListFilter，内部默认使用的是ReversibleFileListFilter<F>, ResettableFileListFilter<F>，所以当FtpInboundFileSynchronizer.copyFileToLocalDirectory()发生异常时，它有这样的逻辑：

    catch (RuntimeException | IOException e1) {
        if (filteringOneByOne) {
            resetFilterIfNecessary(file);
        }
        else {
            rollbackFromFileToListEnd(filteredFiles, file);
        }
        throw e1;
    }

因此，失败的文件将从过滤器存储中删除，因此它将在下一个轮询周期再次成为。

【讨论】：