java - 连接重置 FTP 后恢复文件传输

Question

我正在使用 Spring Integration 构建一个应用程序，该应用程序用于将文件从一个 FTP 服务器(源)发送到另一个 FTP 服务器(目标)。我首先使用入站适配器将文件从源发送到本地目录，然后使用出站适配器将文件从本地目录发送到目标。

我的代码似乎工作正常，我能够实现我的目标，但我的问题是，当在文件传输期间重置到目标 FTP 服务器的连接时，文件传输在连接后不会继续开始工作。

我使用了 Java 配置 inbound和 outbound适配器。谁能告诉我在连接重置后是否可以以某种方式恢复文件传输？

P.S:我是 Spring 的初学者，所以如果我在这里做错了什么，请纠正我。谢谢

AppConfig.java:

@Configuration
@Component
public class FileTransferServiceConfig {

    @Autowired
    private ConfigurationService configurationService;

    public static final String FILE_POLLING_DURATION = "5000";

    @Bean
    public SessionFactory<FTPFile> sourceFtpSessionFactory() {
        DefaultFtpSessionFactory sf = new DefaultFtpSessionFactory();
        sf.setHost(configurationService.getSourceHostName());
        sf.setPort(Integer.parseInt(configurationService.getSourcePort()));
        sf.setUsername(configurationService.getSourceUsername());
        sf.setPassword(configurationService.getSourcePassword());
        return new CachingSessionFactory<FTPFile>(sf);
    }

    @Bean
    public SessionFactory<FTPFile> targetFtpSessionFactory() {
        DefaultFtpSessionFactory sf = new DefaultFtpSessionFactory();
        sf.setHost(configurationService.getTargetHostName());
        sf.setPort(Integer.parseInt(configurationService.getTargetPort()));
        sf.setUsername(configurationService.getTargetUsername());
        sf.setPassword(configurationService.getTargetPassword());
        return new CachingSessionFactory<FTPFile>(sf);
    }

    @MessagingGateway
    public interface MyGateway {

         @Gateway(requestChannel = "toFtpChannel")
         void sendToFtp(Message message);

    }

    @Bean
    public FtpInboundFileSynchronizer ftpInboundFileSynchronizer() {
        FtpInboundFileSynchronizer fileSynchronizer = new FtpInboundFileSynchronizer(sourceFtpSessionFactory());
        fileSynchronizer.setDeleteRemoteFiles(false);
        fileSynchronizer.setRemoteDirectory(configurationService.getSourceDirectory());
        fileSynchronizer.setFilter(new FtpSimplePatternFileListFilter(
                configurationService.getFileMask()));
        return fileSynchronizer;
    }

    @Bean
    @InboundChannelAdapter(channel = "ftpChannel",
            poller = @Poller(fixedDelay = FILE_POLLING_DURATION ))
    public MessageSource<File> ftpMessageSource() {
        FtpInboundFileSynchronizingMessageSource source =
                new FtpInboundFileSynchronizingMessageSource(ftpInboundFileSynchronizer());
        source.setLocalDirectory(new File(configurationService.getLocalDirectory()));
        source.setAutoCreateLocalDirectory(true);
        source.setLocalFilter(new AcceptOnceFileListFilter<File>());
        return source;
    }



    @Bean
    @ServiceActivator(inputChannel = "ftpChannel")
    public MessageHandler targetHandler() {
        FtpMessageHandler handler = new FtpMessageHandler(targetFtpSessionFactory());
        handler.setRemoteDirectoryExpression(new LiteralExpression(
                configurationService.getTargetDirectory()));
        return handler;
    }    
}

Application.java:

@SpringBootApplication
public class Application {

    public static ConfigurableApplicationContext context;

    public static void main(String[] args) {
        context = new SpringApplicationBuilder(Application.class)
                .web(false)
                .run(args);
    }

    @Bean
    @ServiceActivator(inputChannel = "ftpChannel")
    public MessageHandler sourceHandler() {
        return new MessageHandler() {

            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                Object payload = message.getPayload();
                System.out.println("Payload: " + payload);
                if (payload instanceof File) {
                    File file = (File) payload;
                    System.out.println("Trying to send " + file.getName() + " to target");
                }
                MyGateway gateway = context.getBean(MyGateway.class);
                gateway.sendToFtp(message);
            }

        };
    }
}

Answer 1

首先，不清楚 sourceHandler 的用途，但您确实应该确保它已订阅(或 targetHandler)到正确的 channel 。

我以某种方式相信，在您的目标代码中，targetHandler确实订阅了toFtpChannel。

无论如何，这不相关。

我认为这里的问题正是由 AcceptOnceFileListFilter 和错误引起的。因此，出于性能原因，在目录扫描期间首先过滤工作并将所有本地文件加载到内存队列中。然后全部发送到 channel 进行处理。当我们到达 targetHandler 并收到异常时，我们只是默默地进入全局 errorChannel ，丢失了文件尚未传输的事实。内存中所有剩余文件都会发生这种情况。我认为传输无论如何都会恢复，但它仅适用于远程目录中的新文件。

我建议您将 ExpressionEvaluatingRequestHandlerAdvice 添加到 targetHandler 定义 (@ServiceActivator(adviceChain))，并在出现错误时调用 AcceptOnceFileListFilter.remove(File):

/**
 * Remove the specified file from the filter so it will pass on the next attempt.
 * @param f the element to remove.
 * @return true if the file was removed as a result of this call.
 */
boolean remove(F f);

通过这种方式，您可以从过滤器中删除失败的文件，并将在下一个轮询任务中拾取该文件。您必须使 AcceptOnceFileListFilter 才能从 onFailureExpression 访问它。该文件是请求消息的有效负载。

编辑

ExpressionEvaluatingRequestHandlerAdvice 的示例:

@Bean
public Advice expressionAdvice() {
    ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
    advice.setOnFailureExpressionString("@acceptOnceFileListFilter.remove(payload)");
    advice.setTrapException(true);
    return advice;
}

...

@ServiceActivator(inputChannel = "ftpChannel", adviceChain = "expressionAdvice")

您可以从他们的 JavaDocs 中获得所有其他信息。