我正在使用Spring Integration构建一个应用程序,该应用程序用于将文件从一个FTP服务器(源)发送到另一FTP服务器(目标)。我首先使用入站适配器将文件从源发送到本地目录,然后使用出站适配器将文件从本地目录发送到目标。
我的代码似乎运行良好,并且能够实现我的目标,但是我的问题是,在文件传输期间将连接重置为目标FTP服务器时,在连接开始工作后文件传输无法继续。
我使用了inbound和outbound适配器的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获得。