Spring Batch中如何为KafkaItemReader设置永久监听的无限pollTimeout

可以实现，不需要设置真正意义上的无限超时，通过调整KafkaItemReader配置+自定义逻辑就能达到常驻监听、手动停止的效果，具体操作如下：
首先明确KafkaItemReader的运行逻辑：你调用setPollTimeout配置的是单次调用Kafka Consumer poll()方法的等待时长，单次poll超时返回空结果时，KafkaItemReader就会返回null，批处理Step会判定数据读取完成，进而触发作业结束。所以单纯调整pollTimeout参数只能延长等待时间，要实现常驻效果还要配合额外逻辑。 #

具体实现步骤 #

1. 配置KafkaItemReader的最长单次poll时长
   将pollTimeout设置为Kafka Consumer支持的最大值，基本可以等效为单次poll不会因为超时返回空：

@Bean
public KafkaItemReader<String, String> kafkaItemReader(ConsumerFactory<String, String> consumerFactory) {
    return new KafkaItemReaderBuilder<String, String>()
            .consumerFactory(consumerFactory)
            .topic("你的目标Kafka主题")
            .partitions(0, 1, 2) // 替换为你需要监听的分区列表
            .pollTimeout(Duration.ofMillis(Long.MAX_VALUE))
            .saveState(true) // 开启状态保存，作业重启后可以从上次提交的offset继续消费
            .build();
}

2. 自定义代理Reader避免作业自动结束
   自定义Reader包装原生KafkaItemReader，当原生Reader返回空结果时不直接返回null，而是循环等待新消息，同时预留手动停止的信号入口：

public class PersistentKafkaItemReader<T> implements ItemReader<T> {
    private final KafkaItemReader<?, T> delegateReader;
    // 停止信号持有类，可通过外部接口修改状态
    private final AtomicBoolean stopSignal = new AtomicBoolean(false);

    public PersistentKafkaItemReader(KafkaItemReader<?, T> delegateReader) {
        this.delegateReader = delegateReader;
    }

    @Override
    public T read() throws Exception {
        T record;
        while ((record = delegateReader.read()) == null && !stopSignal.get()) {
            // 空轮询时休眠1s降低CPU占用
            Thread.sleep(1000);
        }
        return stopSignal.get() ? null : record;
    }

    // 暴露停止方法，可对接Spring Boot Actuator端点或者自定义HTTP接口触发
    public void triggerStop() {
        stopSignal.set(true);
    }
}

3. 配置作业常驻运行
   启动作业时用CountDownLatch阻塞主线程，避免JVM自动退出，同时注册关闭钩子响应终止信号：

@Bean
public CommandLineRunner startPersistentJob(JobLauncher jobLauncher, Job kafkaConsumeJob) {
    return args -> {
        CountDownLatch shutdownLatch = new CountDownLatch(1);
        // 异步启动批处理作业
        jobLauncher.run(kafkaConsumeJob, new JobParameters())
                .addExitStatusCallback(exitStatus -> shutdownLatch.countDown());
        // 注册JVM关闭钩子，收到终止信号时触发作业停止
        Runtime.getRuntime().addShutdownHook(new Thread(shutdownLatch::countDown));
        shutdownLatch.await();
    };
}

注意事项 #

• 可以对接Spring Boot Actuator的批处理管理端点，实现可视化手动停止作业
• 生产环境建议配置合理的消费offset提交策略，避免消息丢失或者重复消费
• 需要多实例部署提高可用性时，要保证不同实例监听不同的Kafka分区，避免重复消费

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