Spring Integration：为整个流执行配置fixedDelay轮询器的优雅方式

针对你描述的需求——确保同一时刻仅处理一组实体键、出错/延迟时暂停下一轮轮询直到当前流程完成（成功或超时），同时基于实体最后修改时间动态调整下一次查询窗口，我整理了几个Spring Integration原生支持的优雅配置方案，直接就能落地：

1. 基础：锁定单轮单任务，杜绝并发执行 #

首先要从根源上避免多轮任务同时运行，核心是给轮询器设置单线程执行器+单次轮询仅取一组数据：

@Bean
public PollerMetadata entityPoller(TaskExecutor singleThreadExecutor) {
    return Pollers.fixedDelay(Duration.ofSeconds(10)) // 基础轮询间隔，可根据业务调整
            .maxMessagesPerPoll(1) // 每次轮询仅获取一组实体键列表
            .taskExecutor(singleThreadExecutor)
            .get();
}

@Bean
public TaskExecutor singleThreadExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(1);
    executor.setMaxPoolSize(1);
    executor.setQueueCapacity(0); // 拒绝队列积压，确保同一时间只有一个任务在跑
    executor.initialize();
    return executor;
}

这段配置直接保证了轮询器不会同时触发多组任务，完全符合你"同一时刻最多处理一组键列表"的要求。

2. 错误/延迟时暂停轮询：用Advice实现动态启停 #

为了在任务处理出错或超时后暂停轮询，直到当前任务完成（成功/超时），可以给轮询器绑定错误处理通知，实现失败停、成功启的逻辑：

步骤1：配置错误处理通知 #

@Bean
public ExpressionEvaluatingRequestHandlerAdvice pollerControlAdvice() {
    ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
    // 任务失败时，触发轮询器暂停
    advice.setOnFailureExpressionString("T(org.springframework.integration.scheduling.PollerMetadata).stopPoller()");
    // 任务成功完成时，恢复轮询器
    advice.setOnSuccessExpressionString("T(org.springframework.integration.scheduling.PollerMetadata).resumePoller()");
    advice.setTrapException(false); // 允许异常继续传播，方便后续做自定义错误日志
    return advice;
}

步骤2：把通知绑定到轮询器 #

修改之前的Poller配置，加入上面的Advice：

@Bean
public PollerMetadata entityPoller(TaskExecutor singleThreadExecutor, ExpressionEvaluatingRequestHandlerAdvice pollerControlAdvice) {
    return Pollers.fixedDelay(Duration.ofSeconds(10))
            .maxMessagesPerPoll(1)
            .taskExecutor(singleThreadExecutor)
            .advice(pollerControlAdvice) // 绑定启停控制通知
            .get();
}

这样一来，只要当前任务没处理完（不管是出错还是延迟），轮询器都会暂停下一轮触发，直到任务成功完成或者超时终止。

3. 动态调整查询时间窗口：基于最后修改时间的参数化查询 #

针对"根据实体最后修改时间设置下一次SQL查询窗口"的需求，你可以通过自定义MessageSource或者给原生JdbcPollingChannelAdapter绑定动态参数来实现：

方案A：自定义MessageSource（更灵活） #

自己维护上次处理的最新时间戳，每次查询前动态生成SQL：

@Bean
public MessageSource<List<Long>> entityKeySource(JdbcTemplate jdbcTemplate) {
    return new AbstractMessageSource<List<Long>>() {
        // 初始时间设为启动前1小时，避免漏处理历史数据
        private volatile LocalDateTime lastProcessedTime = LocalDateTime.now().minusHours(1);

        @Override
        protected List<Long> doReceive() {
            // 动态拼接SQL，用上一次处理的时间作为查询起始点
            String sql = "SELECT id FROM entities WHERE last_modified > ? ORDER BY last_modified";
            List<Long> entityKeys = jdbcTemplate.queryForList(sql, Long.class, lastProcessedTime);
            
            if (!entityKeys.isEmpty()) {
                // 更新lastProcessedTime为本次查询到的最新修改时间
                LocalDateTime latestModified = jdbcTemplate.queryForObject(
                        "SELECT MAX(last_modified) FROM entities WHERE id IN (?)",
                        LocalDateTime.class,
                        new Object[]{entityKeys});
                this.lastProcessedTime = latestModified;
            }
            
            // 如果没有数据，返回null会让轮询器跳过本次处理
            return entityKeys.isEmpty() ? null : entityKeys;
        }

        @Override
        public String getComponentType() {
            return "custom-jdbc-polling-source";
        }
    };
}

然后把这个自定义源绑定到集成流：

@Bean
public IntegrationFlow entityProcessingFlow(MessageSource<List<Long>> entityKeySource, PollerMetadata entityPoller) {
    return IntegrationFlows.from(entityKeySource, spec -> spec.poller(entityPoller))
            .handle(entityProcessor()) // 这里替换成你的实体处理逻辑Bean
            .get();
}

方案B：原生JdbcPollingChannelAdapter绑定动态参数 #

如果想用原生组件，可以通过ParameterSourceProvider动态注入查询参数：

@Bean
public JdbcPollingChannelAdapter jdbcPollingChannelAdapter(DataSource dataSource) {
    String sql = "SELECT id FROM entities WHERE last_modified > :lastProcessedTime ORDER BY last_modified";
    JdbcPollingChannelAdapter adapter = new JdbcPollingChannelAdapter(dataSource, sql);
    
    adapter.setParameterSourceProvider(() -> {
        MapSqlParameterSource params = new MapSqlParameterSource();
        params.addValue("lastProcessedTime", getLastProcessedTime()); // 自定义方法获取上次处理的时间
        return params;
    });
    adapter.setRowMapper((rs, rowNum) -> rs.getLong("id"));
    return adapter;
}

这种方式需要自己维护lastProcessedTime的状态，适合不需要太复杂逻辑的场景。

4. 超时控制：避免任务无限阻塞 #

为了防止任务处理陷入无限延迟，给任务执行器加上超时控制：

@Bean
public TaskExecutor singleThreadExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(1);
    executor.setMaxPoolSize(1);
    executor.setQueueCapacity(0);
    executor.setTaskDecorator(runnable -> {
        FutureTask<?> futureTask = new FutureTask<>(runnable, null);
        new Thread(() -> {
            try {
                // 设置5分钟超时，可根据业务调整
                futureTask.get(Duration.ofMinutes(5).toMillis(), TimeUnit.MILLISECONDS);
            } catch (TimeoutException e) {
                futureTask.cancel(true);
                // 超时后恢复轮询器，避免永久停摆
                PollerMetadata.resumePoller();
                // 这里可以加超时日志告警
            } catch (Exception e) {
                // 其他异常处理逻辑
            }
        }).start();
        return futureTask;
    });
    executor.initialize();
    return executor;
}

这样即使任务处理超时，轮询器也能自动恢复，不会一直停摆。

总结 #

通过以上配置组合，你可以完美实现：

• 同一时刻仅处理一组实体键
• 任务出错/延迟时自动暂停轮询，完成后恢复
• 基于最后修改时间动态调整查询窗口
• 超时控制避免无限阻塞

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