如何通过Spring Integration结合WebFlux批量轮询多文件？

你的问题核心在于默认轮询配置每次只读取1个文件，加上10秒的轮询间隔，导致100个文件需要1000秒才能处理完。以下是具体的修复方案：

关键问题分析 #

1. 原Poller配置未设置maxMessagesPerPoll，Spring Integration默认每次轮询仅获取1条消息（文件）
2. 响应式处理流程存在冗余配置，可能影响批量消息的流转

修复步骤 #

1. 修改Poller配置，允许批量读取文件 #

在Pollers.fixedDelay中添加maxMessagesPerPoll配置，设置为单次轮询最大文件数（或设为-1表示读取所有可用文件）：

e -> e.poller(Pollers.fixedDelay(pollingInSeconds, TimeUnit.SECONDS)
        .maxMessagesPerPoll(-1)) // -1表示单次轮询读取所有可用文件，也可指定具体数值如100

2. 修复响应式处理的冗余配置 #

原myReactiveSource直接返回myMessagePublisher()会导致重复创建发布者，应改为直接传递传入的Flux；同时调整@ServiceActivator配置，移除不必要的async属性，改用Reactive原生调度器优化异步处理：

@Bean
Function<Flux<Message<Object>>, Publisher<Message<Object>>> myReactiveSource() {
    return flux -> flux; // 直接传递原Flux，无需重新创建发布者
}

@Bean
@ServiceActivator(
    inputChannel = "myChannel",
    reactive = @Reactive("myReactiveSource"))
ReactiveMessageHandler myMessageHandler() {
    return message -> Mono.fromRunnable(() -> {
                logger.info("Received a notification of new file: {}", message.getPayload());
                File file = (File) message.getPayload();
                // 这里添加你的文件处理逻辑
            })
            .subscribeOn(Schedulers.boundedElastic())
            .then();
}

3. 可选：启用WatchService提升文件检测效率 #

如果使用Java 7+，可以开启useWatchService(true)，利用操作系统原生文件监听机制，无需依赖固定间隔轮询，能更快感知新文件：

Files.inboundAdapter(new File(monitoringFolder))
    .useWatchService(true)
    .watchEvents(FileReadingMessageSource.WatchEventType.CREATE)

完整修改后的配置代码 #

@Configuration
@EnableIntegration
public class FileMonitoringConfig {

  private static final Logger logger =
      LoggerFactory.getLogger(FileMonitoringConfig.class.getName());

  @Value("${monitoring.folder}")
  private String monitoringFolder;

  @Value("${monitoring.polling-in-seconds:10}")
  private int pollingInSeconds;

  @Bean
  Publisher<Message<Object>> myMessagePublisher() {
    return IntegrationFlows.from(
            Files.inboundAdapter(new File(monitoringFolder))
                .useWatchService(false), // 如需实时监听可改为true
            e -> e.poller(Pollers.fixedDelay(pollingInSeconds, TimeUnit.SECONDS)
                    .maxMessagesPerPoll(-1))) // 单次轮询读取所有可用文件
        .channel(myChannel())
        .toReactivePublisher();
  }

  @Bean
  Function<Flux<Message<Object>>, Publisher<Message<Object>>> myReactiveSource() {
    return flux -> flux;
  }

  @Bean
  FluxMessageChannel myChannel() {
    return new FluxMessageChannel();
  }

  @Bean
  @ServiceActivator(
      inputChannel = "myChannel",
      reactive = @Reactive("myReactiveSource"))
  ReactiveMessageHandler myMessageHandler() {
    return message -> Mono.fromRunnable(() -> {
                logger.info("Received a notification of new file: {}", message.getPayload());
                File file = (File) message.getPayload();
                // 执行你的文件处理逻辑
            })
            .subscribeOn(Schedulers.boundedElastic())
            .then();
  }
}

效果说明 #

修改后，每次轮询会一次性读取所有符合条件的文件（或你设置的最大数量），消息会批量进入处理流程，无需等待10秒间隔逐个识别文件。如果启用WatchService，新文件创建时会立即被检测到，进一步提升处理效率。

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