Sftp 出站网关多个删除请求在第一个 "File not found" 问题后卡住
Sftp Outbound Gateway multiple delete request getting stuck after first "File not found" issue
我正在尝试通过 Sftp 删除多个文件并使用 Spring Integration SftpOutboundGateway 方法。
我正在使用 QueueChannel 进行请求和响应。此外，我将采用异步方式，以便我可以提交多个请求。 我最近也加了错误频道
现在的问题是我的第一个请求是针对一个不存在的文件，所以我可以收到“2：没有这样的文件”异常消息。但是一旦发生这种情况，其他请求就会卡住。
以下是文件未找到案例的错误信息：
DEBUG o.s.integration.util.SimplePool - Obtained new org.springframework.integration.sftp.session.SftpSession@4ec427c0.
DEBUG o.s.i.f.r.s.CachingSessionFactory - Releasing Session org.springframework.integration.sftp.session.SftpSession@4ec427c0 back to the pool.
INFO com.jcraft.jsch - Disconnecting from xxxx
DEBUG o.s.integration.util.SimplePool - Releasing org.springframework.integration.sftp.session.SftpSession@4ec427c0 back to the pool
INFO com.jcraft.jsch - Caught an exception, leaving main loop due to Socket closed
Caused by: org.springframework.messaging.MessageHandlingException: error occurred in message handler [sftpDeleteFileHandler]; nested exception is org.springframework.messaging.MessagingException: Failed to execute on session; nested exception is org.springframework.core.NestedIOException: Failed to remove file.; nested exception is 2: No such file, failedMessage=GenericMessage [xxx]
at org.springframework.integration.support.utils.IntegrationUtils.wrapInHandlingExceptionIfNecessary(IntegrationUtils.java:189)
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:186)
at org.springframework.integration.endpoint.PollingConsumer.handleMessage(PollingConsumer.java:143)
at org.springframework.integration.endpoint.AbstractPollingEndpoint.doPoll(AbstractPollingEndpoint.java:390)
at org.springframework.integration.endpoint.AbstractPollingEndpoint.pollForMessage(AbstractPollingEndpoint.java:329)
at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$null(AbstractPollingEndpoint.java:277)
at org.springframework.integration.util.ErrorHandlingTaskExecutor.lambda$execute[=11=](ErrorHandlingTaskExecutor.java:57)
at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50)
at org.springframework.integration.util.ErrorHandlingTaskExecutor.execute(ErrorHandlingTaskExecutor.java:55)
at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$createPoller(AbstractPollingEndpoint.java:274)
at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:93)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access1(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
... 1 more
Caused by: org.springframework.messaging.MessagingException: Failed to execute on session; nested exception is org.springframework.core.NestedIOException: Failed to remove file.; nested exception is 2: No such file
at org.springframework.integration.file.remote.RemoteFileTemplate.execute(RemoteFileTemplate.java:446)
at org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway.doRm(AbstractRemoteFileOutboundGateway.java:566)
at org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway.handleRequestMessage(AbstractRemoteFileOutboundGateway.java:459)
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:123)
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:169)
... 17 more
Caused by: org.springframework.core.NestedIOException: Failed to remove file.; nested exception is 2: No such file
at org.springframework.integration.sftp.session.SftpSession.remove(SftpSession.java:83)
at org.springframework.integration.file.remote.session.CachingSessionFactory$CachedSession.remove(CachingSessionFactory.java:225)
at org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway.rm(AbstractRemoteFileOutboundGateway.java:586)
at org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway.lambda$doRm(AbstractRemoteFileOutboundGateway.java:566)
at org.springframework.integration.file.remote.RemoteFileTemplate.execute(RemoteFileTemplate.java:437)
... 21 more
Caused by: 2: No such file
at com.jcraft.jsch.ChannelSftp.throwStatusError(ChannelSftp.java:2873)
at com.jcraft.jsch.ChannelSftp.rm(ChannelSftp.java:1985)
at org.springframework.integration.sftp.session.SftpSession.remove(SftpSession.java:79)
... 25 more
更新 1：
我正在使用 Spring Boot 2.1.8 -> Spring Integration 5.1.7
配置：
@Bean
public SessionFactory<LsEntry> sftpSessionFactory() {
DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
factory.setHost(host);
factory.setPort(port);
factory.setUser(user);
factory.setPassword(password);
factory.setAllowUnknownKeys(true);
return new CachingSessionFactory<LsEntry>(factory);
}
@Bean(name = PollerMetadata.DEFAULT_POLLER)
public PollerMetadata defaultPoller() {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(new PeriodicTrigger(10));
return pollerMetadata;
}
@Bean(name = "sftp.file.delete.request.channel")
public MessageChannel sftpFileDeleteRequestChannel() {
return new QueueChannel();
}
@Bean(name = "sftp.file.delete.response.channel")
public MessageChannel sftpFileDeleteResponseChannel() {
return new QueueChannel();
}
@Bean(name = "sftp.error.channel")
public MessageChannel sftpErrorChannel() {
return MessageChannels.queue("sftp.error.channel").get();
}
@Bean
@ServiceActivator(inputChannel = "sftp.file.delete.request.channel", async = "true")
public MessageHandler sftpDeleteFileHandler() {
SftpOutboundGateway sftpOutboundGateway = new SftpOutboundGateway(sftpSessionFactory(), Command.RM.getCommand(),
"headers['file_remoteDirectory'] + headers['file_remoteFile']");
sftpOutboundGateway.setRequiresReply(true);
return sftpOutboundGateway;
}
@ServiceActivator(inputChannel = "sftp.error.channel")
public void sftpErrorHandler(final Message<MessageHandlingException> excpMessage) {
log.error(excpMessage.getPayload().getCause());
}
@MessagingGateway(errorChannel = "sftp.error.channel")
public interface SftpDeleteMessagingGateway {
@Gateway(requestChannel = "sftp.file.delete.request.channel", replyChannel = "sftp.file.delete.response.channel")
CompletableFuture<Message<Boolean>> deleteFile(final Message<Boolean> message);
}
代码：
List<CompletableFuture<Message<Boolean>>> fileDeleteResults = new ArrayList<>();
foreach(...) {
Message<Boolean> fileDeleteRequest = MessageBuilder.withPayload(true)
.setHeader(FileHeaders.REMOTE_DIRECTORY, directory)
.setHeader(FileHeaders.REMOTE_FILE, name).build();
fileDeleteResults.add(sftpDeleteMessagingGateway.deleteFile(fileDeleteRequest));
}
try {
CompletableFuture.allOf(fileDeleteResults.toArray(new CompletableFuture[fileDeleteResults.size()])).join();
for (CompletableFuture<Message<Boolean>> fileDeleteResult : fileDeleteResults){
Message<Boolean> message = fileDeleteResult.get();
log.debug((String) message.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)
+ (String) message.getHeaders().get(FileHeaders.REMOTE_FILE)
+ ": " + message.getPayload());
}
} catch (CompletionException | InterruptedException | ExecutionException excp) {
log.error(excp);
}
更新 2：
我按照建议修改了配置，还是遇到同样的问题。下面是修改后的配置 -
@Bean
@ServiceActivator(inputChannel = "sftp.file.delete.request.channel")
public MessageHandler sftpDeleteFileHandler() {
SftpOutboundGateway sftpOutboundGateway = new SftpOutboundGateway(sftpSessionFactory(), Command.RM.getCommand(),
"headers['file_remoteDirectory'] + headers['file_remoteFile']");
sftpOutboundGateway.setRequiresReply(true);
return sftpOutboundGateway;
}
@ServiceActivator(inputChannel = "sftp.error.channel")
public boolean sftpErrorHandler(final Message<MessageHandlingException> excpMessage) {
log.error(excpMessage.getPayload().getCause());
return false;
}
好的。看起来你的问题是你没有捕获异常来继续。
请参阅 ServiceActivatingHandler 上的 async 选项：
/**
* Allow async replies. If the handler reply is a {@link ListenableFuture}, send
* the output when it is satisfied rather than sending the future as the result.
* Ignored for return types other than {@link ListenableFuture}.
* @param async true to allow.
* @since 4.3
*/
public final void setAsync(boolean async) {
因此，仅当 ListenableFuture（或 Reactive Publsiher）是从目标 handleRequestMessage() 实施。这不是 SftpOutboundGateway 的事实。因此，您在网关定义上的 errorChannel = "sftp.error.channel" 是正确的方法。尽管您需要 return 来自 sftpErrorHandler 的内容，这将是网关调用的 return。否则我们会一直等待回复或错误。
我正在尝试通过 Sftp 删除多个文件并使用 Spring Integration SftpOutboundGateway 方法。
我正在使用 QueueChannel 进行请求和响应。此外，我将采用异步方式，以便我可以提交多个请求。 我最近也加了错误频道
现在的问题是我的第一个请求是针对一个不存在的文件，所以我可以收到“2：没有这样的文件”异常消息。但是一旦发生这种情况，其他请求就会卡住。
以下是文件未找到案例的错误信息：
DEBUG o.s.integration.util.SimplePool - Obtained new org.springframework.integration.sftp.session.SftpSession@4ec427c0.
DEBUG o.s.i.f.r.s.CachingSessionFactory - Releasing Session org.springframework.integration.sftp.session.SftpSession@4ec427c0 back to the pool.
INFO com.jcraft.jsch - Disconnecting from xxxx
DEBUG o.s.integration.util.SimplePool - Releasing org.springframework.integration.sftp.session.SftpSession@4ec427c0 back to the pool
INFO com.jcraft.jsch - Caught an exception, leaving main loop due to Socket closed
Caused by: org.springframework.messaging.MessageHandlingException: error occurred in message handler [sftpDeleteFileHandler]; nested exception is org.springframework.messaging.MessagingException: Failed to execute on session; nested exception is org.springframework.core.NestedIOException: Failed to remove file.; nested exception is 2: No such file, failedMessage=GenericMessage [xxx]
at org.springframework.integration.support.utils.IntegrationUtils.wrapInHandlingExceptionIfNecessary(IntegrationUtils.java:189)
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:186)
at org.springframework.integration.endpoint.PollingConsumer.handleMessage(PollingConsumer.java:143)
at org.springframework.integration.endpoint.AbstractPollingEndpoint.doPoll(AbstractPollingEndpoint.java:390)
at org.springframework.integration.endpoint.AbstractPollingEndpoint.pollForMessage(AbstractPollingEndpoint.java:329)
at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$null(AbstractPollingEndpoint.java:277)
at org.springframework.integration.util.ErrorHandlingTaskExecutor.lambda$execute[=11=](ErrorHandlingTaskExecutor.java:57)
at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50)
at org.springframework.integration.util.ErrorHandlingTaskExecutor.execute(ErrorHandlingTaskExecutor.java:55)
at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$createPoller(AbstractPollingEndpoint.java:274)
at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:93)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access1(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
... 1 more
Caused by: org.springframework.messaging.MessagingException: Failed to execute on session; nested exception is org.springframework.core.NestedIOException: Failed to remove file.; nested exception is 2: No such file
at org.springframework.integration.file.remote.RemoteFileTemplate.execute(RemoteFileTemplate.java:446)
at org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway.doRm(AbstractRemoteFileOutboundGateway.java:566)
at org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway.handleRequestMessage(AbstractRemoteFileOutboundGateway.java:459)
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:123)
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:169)
... 17 more
Caused by: org.springframework.core.NestedIOException: Failed to remove file.; nested exception is 2: No such file
at org.springframework.integration.sftp.session.SftpSession.remove(SftpSession.java:83)
at org.springframework.integration.file.remote.session.CachingSessionFactory$CachedSession.remove(CachingSessionFactory.java:225)
at org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway.rm(AbstractRemoteFileOutboundGateway.java:586)
at org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway.lambda$doRm(AbstractRemoteFileOutboundGateway.java:566)
at org.springframework.integration.file.remote.RemoteFileTemplate.execute(RemoteFileTemplate.java:437)
... 21 more
Caused by: 2: No such file
at com.jcraft.jsch.ChannelSftp.throwStatusError(ChannelSftp.java:2873)
at com.jcraft.jsch.ChannelSftp.rm(ChannelSftp.java:1985)
at org.springframework.integration.sftp.session.SftpSession.remove(SftpSession.java:79)
... 25 more
更新 1：
我正在使用 Spring Boot 2.1.8 -> Spring Integration 5.1.7
配置：
@Bean
public SessionFactory<LsEntry> sftpSessionFactory() {
DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
factory.setHost(host);
factory.setPort(port);
factory.setUser(user);
factory.setPassword(password);
factory.setAllowUnknownKeys(true);
return new CachingSessionFactory<LsEntry>(factory);
}
@Bean(name = PollerMetadata.DEFAULT_POLLER)
public PollerMetadata defaultPoller() {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(new PeriodicTrigger(10));
return pollerMetadata;
}
@Bean(name = "sftp.file.delete.request.channel")
public MessageChannel sftpFileDeleteRequestChannel() {
return new QueueChannel();
}
@Bean(name = "sftp.file.delete.response.channel")
public MessageChannel sftpFileDeleteResponseChannel() {
return new QueueChannel();
}
@Bean(name = "sftp.error.channel")
public MessageChannel sftpErrorChannel() {
return MessageChannels.queue("sftp.error.channel").get();
}
@Bean
@ServiceActivator(inputChannel = "sftp.file.delete.request.channel", async = "true")
public MessageHandler sftpDeleteFileHandler() {
SftpOutboundGateway sftpOutboundGateway = new SftpOutboundGateway(sftpSessionFactory(), Command.RM.getCommand(),
"headers['file_remoteDirectory'] + headers['file_remoteFile']");
sftpOutboundGateway.setRequiresReply(true);
return sftpOutboundGateway;
}
@ServiceActivator(inputChannel = "sftp.error.channel")
public void sftpErrorHandler(final Message<MessageHandlingException> excpMessage) {
log.error(excpMessage.getPayload().getCause());
}
@MessagingGateway(errorChannel = "sftp.error.channel")
public interface SftpDeleteMessagingGateway {
@Gateway(requestChannel = "sftp.file.delete.request.channel", replyChannel = "sftp.file.delete.response.channel")
CompletableFuture<Message<Boolean>> deleteFile(final Message<Boolean> message);
}
代码：
List<CompletableFuture<Message<Boolean>>> fileDeleteResults = new ArrayList<>();
foreach(...) {
Message<Boolean> fileDeleteRequest = MessageBuilder.withPayload(true)
.setHeader(FileHeaders.REMOTE_DIRECTORY, directory)
.setHeader(FileHeaders.REMOTE_FILE, name).build();
fileDeleteResults.add(sftpDeleteMessagingGateway.deleteFile(fileDeleteRequest));
}
try {
CompletableFuture.allOf(fileDeleteResults.toArray(new CompletableFuture[fileDeleteResults.size()])).join();
for (CompletableFuture<Message<Boolean>> fileDeleteResult : fileDeleteResults){
Message<Boolean> message = fileDeleteResult.get();
log.debug((String) message.getHeaders().get(FileHeaders.REMOTE_DIRECTORY)
+ (String) message.getHeaders().get(FileHeaders.REMOTE_FILE)
+ ": " + message.getPayload());
}
} catch (CompletionException | InterruptedException | ExecutionException excp) {
log.error(excp);
}
更新 2：
我按照建议修改了配置，还是遇到同样的问题。下面是修改后的配置 -
@Bean
@ServiceActivator(inputChannel = "sftp.file.delete.request.channel")
public MessageHandler sftpDeleteFileHandler() {
SftpOutboundGateway sftpOutboundGateway = new SftpOutboundGateway(sftpSessionFactory(), Command.RM.getCommand(),
"headers['file_remoteDirectory'] + headers['file_remoteFile']");
sftpOutboundGateway.setRequiresReply(true);
return sftpOutboundGateway;
}
@ServiceActivator(inputChannel = "sftp.error.channel")
public boolean sftpErrorHandler(final Message<MessageHandlingException> excpMessage) {
log.error(excpMessage.getPayload().getCause());
return false;
}
好的。看起来你的问题是你没有捕获异常来继续。
请参阅 ServiceActivatingHandler 上的 async 选项：
/**
* Allow async replies. If the handler reply is a {@link ListenableFuture}, send
* the output when it is satisfied rather than sending the future as the result.
* Ignored for return types other than {@link ListenableFuture}.
* @param async true to allow.
* @since 4.3
*/
public final void setAsync(boolean async) {
因此，仅当 ListenableFuture（或 Reactive Publsiher）是从目标 handleRequestMessage() 实施。这不是 SftpOutboundGateway 的事实。因此，您在网关定义上的 errorChannel = "sftp.error.channel" 是正确的方法。尽管您需要 return 来自 sftpErrorHandler 的内容，这将是网关调用的 return。否则我们会一直等待回复或错误。