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

我的代码似乎运行良好,并且能够实现我的目标,但是我的问题是,在文件传输期间将连接重置为目标FTP服务器时,在连接开始工作后文件传输无法继续。

我使用了inboundoutbound适配器的Java配置。有人可以告诉我在重置连接后是否可以恢复我的文件传输吗?

附注:我是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);
            }

        };
    }
}

最佳答案

首先,尚不清楚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获得。

09-10 05:42
查看更多