Spring Batch中如何使用KafkaItemReader读取多个Kafka主题？

当然可以用KafkaItemReader读取多个主题，只需要将原代码中的topic()方法替换为topics()方法即可——这个方法支持传入多个主题名称，或者一个包含多个主题的集合，无需额外创建多个Bean。

修改后的示例代码 #

return new KafkaItemReaderBuilder<String, Activity>()
    .consumerProperties(toProperties(of(
            ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress,
            ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class,
            ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class
    )))
    .pollTimeout(ofMillis(600))
    // 传入多个主题名称，替代单个主题的topic()方法
    .topics(checkInTopicName, checkOutTopicName)
    .build();

区分不同主题的消息 #

如果需要在处理阶段区分消息来自哪个主题，可通过ConsumerRecord的元数据获取主题信息——KafkaItemReader读取的每个Item本质是ConsumerRecord<String, Activity>，你可以在ItemProcessor中做逻辑分支：

public class ActivityProcessor implements ItemProcessor<ConsumerRecord<String, Activity>, Activity> {
    private final String checkInTopicName;
    private final String checkOutTopicName;

    public ActivityProcessor(String checkInTopicName, String checkOutTopicName) {
        this.checkInTopicName = checkInTopicName;
        this.checkOutTopicName = checkOutTopicName;
    }

    @Override
    public Activity process(ConsumerRecord<String, Activity> record) throws Exception {
        String sourceTopic = record.topic();
        if (checkInTopicName.equals(sourceTopic)) {
            // 执行签到相关的处理逻辑
        } else if (checkOutTopicName.equals(sourceTopic)) {
            // 执行签出相关的处理逻辑
        }
        return record.value();
    }
}

关于你的长期优化方案 #

你提到后续会改用单主题+多分区的方案来提升任务一致性，这个思路非常合理：单主题下的分区可以让消费者统一管理消费逻辑，同时通过分区策略隔离签到/签出消息，既能保持逻辑一致性，又能避免多主题带来的额外配置成本，是更优的长期架构选择。

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