我有一个非常简单的 Spring Boot 应用程序,它提供了几个静态端点,这应该驱动将 sftp 文件上传到 sftp 服务器。我的要求是,如果有多个文件,则文件应该排队。我希望通过 sftp spring 集成工作流程的默认行为来实现这一点,因为我读到 DirectChannel 自动对文件进行排队。为了测试行为,我执行以下操作:
- 发送一个大文件,通过调用端点来阻塞 channel 一段时间。
- 通过调用端点发送较小的文件。
预期结果:较小的文件在 channel 上排队,并在较大的文件上传完成后进行处理。 实际结果:打开与 sftp 服务器的新连接,较小的文件不排队上传到那里,而较大的文件继续传输。
我的应用程序中有两个文件:
DemoApplication.java
@SpringBootApplication
@IntegrationComponentScan
@EnableAutoConfiguration(exclude={DataSourceAutoConfiguration.class})
public class DemoApplication {
public static void main(String[] args) {
SpringApplication.run(DemoApplication.class, args);
}
@Bean
public SessionFactory<LsEntry> sftpSessionFactory() {
DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
factory.setHost("localhost");
factory.setPort(22);
factory.setUser("tester");
factory.setPassword("password");
factory.setAllowUnknownKeys(true);
return factory;
}
@Bean
@ServiceActivator(inputChannel = "toSftpChannel")
public MessageHandler handler() {
SftpMessageHandler handler = new SftpMessageHandler(sftpSessionFactory());
handler.setRemoteDirectoryExpression(new LiteralExpression("/"));
return handler;
}
@MessagingGateway
public interface MyGateway {
@Gateway(requestChannel = "toSftpChannel")
void sendToSftp(File file);
}
}
DemoController.java
@RestController
public class DemoController {
@Autowired
MyGateway gateway;
@RequestMapping("/sendFile")
public void sendFile() {
File file = new File("C:/smallFile.txt");
gateway.sendToSftp(file);
}
@RequestMapping("/sendBigFile")
public void sendBigFile() {
File file = new File("D:/bigFile.zip");
gateway.sendToSftp(file);
}
}
我是 spring 的新手,我完全不确定我的 sftp channel 是否在这里正确创建,我的猜测是每次我执行 sendToSftp 调用时都会创建一个新 channel 。任何有关如何在这种情况下实现队列行为的帮助将不胜感激。
最佳答案
这里没有队列,因为每个 HTTP 请求都在其自己的线程中执行。是的,当 http 线程池耗尽时,您可能会在那里排队,但这在只有两个请求的简单用例中并不存在。
无论如何,您都可以在那里实现队列行为,但您应该将 toSftpChannel
声明为 QueueChannel
bean。
这样下游进程将始终在同一线程上执行,并且下一条消息恰好在第一条消息之后从队列中提取。
参见Reference Manual了解更多信息。
更新
由于您使用的是单向组件FtpMessageHandler
,但您仍然需要对 MVC Controller 的方法进行一些回复,唯一的方法就是拥有一个@Gateway
code> 方法具有非 void
返回,当然我们需要以某种方式发送回复。
为此,我建议使用PublishSubscribeChannel
:
@Bean
@BridgeTo
public MessageChannel toSftpChannel() {
return new PublishSubscribeChannel();
}
@Bean
@ServiceActivator(inputChannel = "toSftpChannel")
@Order(0)
public MessageHandler handler() {
SftpMessageHandler handler = new SftpMessageHandler(sftpSessionFactory());
handler.setRemoteDirectoryExpression(new LiteralExpression("/"));
return handler;
}
这样我们就有了 toSftpChannel
的两个订阅者。通过 @Order(0)
,我们确保 @ServiceActivator
是第一个订阅者,因为我们需要首先执行 SFTP 传输。通过 @BridgeTo
,我们将第二个 BridgeHandler
添加到同一个 PublishSubscribeChannel
。其目的只是获取 replyChannel
header 并在那里发送请求消息。由于我们不使用任何线程,BridgeHandler
将在完成向 SFTP 的传输后立即执行。
当然,您可以使用任何其他 @ServiceActivator
或 @Transfromer
作为回复而不是请求返回,而不是 BridgeHandler
文件
,但还有其他任何东西。例如:
@ServiceActivator(inputChannel = "toSftpChannel")
@Order(1)
public String transferComplete(File payload) {
return "The SFTP transfer complete for file: " + payload;
}
关于java - Spring Integration 中的 REST 端点使消息 channel 成为多线程,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/45362706/