连接重置 FTP 后继续传输文件

Question

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

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

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

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）到正确的频道。

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

反正不相关。

我认为这里的问题正是AcceptOnceFileListFilter 和错误。因此，出于性能原因，在目录扫描期间首先过滤工作并将所有本地文件加载到内存队列中。然后将它们全部发送到通道进行处理。当我们到达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 访问它。该文件是请求消息的payload。

编辑

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 中获得所有其他信息。