如何通过Control Bus启停Spring Integration输入端点

背景 #

本人参与过多个Apache Camel项目，对EIP不陌生，目前正在学习Spring Integration，编写了如下流程代码：

FileReadingMessageSource fileSource = new FileReadingMessageSource();
fileSource.setBeanName("fileMessageSource");
fileSource.setDirectory(new File("C:/DestDir"));
fileSource.setAutoCreateDirectory(true);

DefaultFtpSessionFactory ftpSessionFactory = new DefaultFtpSessionFactory();

IntegrationFlow flow = IntegrationFlows.from(fileSource, configurer -> configurer.poller(Pollers.cron("0 0 * * *")))
    .filter("*.png")    // only allow PNG files through
    .controlBus()       // add a control bus
    .handle(Ftp.outboundAdapter(ftpSessionFactory, FileExistsMode.FAIL)
        .useTemporaryFileName(false)
        .remoteDirectory("uploadDir"))
    .get();

该流程实现以下功能：

• 每天轮询指定文件夹获取PNG图片
• 仅允许PNG文件通过
• 将图片上传至FTP服务器指定目录
• 定义了用于管理和监控流程的Control Bus

疑问 #

1. Spring Integration中flow和channel的区别是什么？
2. 如何借助Control Bus启停上述代码中的fileSource输入端点？已知Control Bus通过接收SPeL消息调用对应Bean的方法，但不清楚需向哪个通道发送何种具体消息来实现该端点的启停。

解答 #

1. Flow与Channel的区别 #

• IntegrationFlow：是Spring Integration中编排消息处理逻辑的完整链路，相当于一条"生产线"，包含从输入端点、各类消息处理器（filter、handle等）到输出端点的全流程定义，负责串联组件并指定执行逻辑顺序。
• MessageChannel：是消息在组件间传递的"管道"，是Flow内部各个处理器的连接载体。每个处理器的输出会发送到对应Channel，下一个处理器从Channel接收消息。Flow会自动生成默认Channel，也可自定义Channel来控制消息的路由、缓存或并发规则。简言之，Flow是整体处理逻辑，Channel是逻辑内部的消息传输通道。

2. 使用Control Bus启停fileSource端点 #

首先要明确：FileReadingMessageSource本身不是可直接启停的对象，真正需要控制的是它对应的轮询端点（MessageSourcePollingEndpoint）。当你用IntegrationFlows.from(fileSource, configurer -> configurer.poller(...))配置时，Spring会自动创建这个轮询端点。

调整Control Bus配置

你当前代码把controlBus()嵌入业务Flow的做法有问题——Control Bus会消费掉前面的业务消息，导致后续FTP上传逻辑无法执行。正确的方式是单独定义Control Bus：

// 单独定义Control Bus，绑定到名为controlBus的通道
@Bean
public IntegrationFlow controlBusFlow() {
    return IntegrationFlows.from("controlBus")
            .controlBus()
            .get();
}

// 业务Flow单独定义
@Bean
public IntegrationFlow fileToFtpFlow() {
    return IntegrationFlows.from(fileSource, configurer -> configurer.poller(Pollers.cron("0 0 * * *")))
            .filter("*.png")
            .handle(Ftp.outboundAdapter(ftpSessionFactory, FileExistsMode.FAIL)
                    .useTemporaryFileName(false)
                    .remoteDirectory("uploadDir"))
            .get();
}

发送启停指令

Control Bus默认监听名为controlBus的通道，你可以注入该通道并发送包含SPeL表达式的消息：

@Autowired
private MessageChannel controlBus;

• 启动轮询端点：
  由于你的fileSource设置了beanName为fileMessageSource，对应的轮询端点ID为fileMessageSource.inboundAdapter，发送启动指令：

  controlBus.send(MessageBuilder.withPayload("@'fileMessageSource.inboundAdapter'.start()").build());
  

• 停止轮询端点：

  controlBus.send(MessageBuilder.withPayload("@'fileMessageSource.inboundAdapter'.stop()").build());
  

可选：显式指定端点ID

你也可以在定义Flow时直接指定端点ID，更便于识别：

IntegrationFlow flow = IntegrationFlows.from(fileSource, configurer -> configurer
        .id("filePollingEndpoint")
        .poller(Pollers.cron("0 0 * * *")))
    // ... 其他配置
    .get();

此时启停指令改为：

// 启动
controlBus.send(MessageBuilder.withPayload("@'filePollingEndpoint'.start()").build());
// 停止
controlBus.send(MessageBuilder.withPayload("@'filePollingEndpoint'.stop()").build());

内容的提问来源于stack exchange，提问作者hotmeatballsoup