Spring Kafka Streams函数式消费多主题仅能收到首个主题消息问题

配置调整 #

你当前的多主题配置写法本身符合规范，无法消费其余主题的问题一般由配置解析规则、偏移量配置两个原因导致，按以下步骤调整即可：

• 修正多主题解析配置
  部分版本的Spring Boot会将带逗号的配置值识别为单个字符串，而非数组，导致绑定器将三个主题拼接的整串识别为单个主题名，修改配置如下：

# 用中括号包裹主题列表，强制解析为数组类型
spring.cloud.stream.bindings.routeRequests-in-0.destination=[kafk.pds.orch.be-uda.complete,kafk.pds.orch.prov-uda.complete,kafk.pds.orch.location-uda.complete]
spring.cloud.stream.bindings.routeRequests-in-0.consumer.use-native-decoding=false
spring.cloud.stream.kafka.streams.binder.functions.routeRequests.applicationId=kafk.pds.orch.stream.routeRequests
# 新增偏移量配置，测试阶段从最早位置消费，避免看不到启动前写入的消息
spring.cloud.stream.kafka.streams.binder.configuration.auto.offset.reset=earliest

• 验证绑定结果
  服务启动后，在日志中搜索关键字Subscribed to topic(s):，确认打印的主题列表包含三个目标主题，而非带逗号的整串值。
• 消费者组校验
  如果调整配置后仍无法消费剩余主题的消息，可修改applicationId为全新值重启测试，避免原有消费者组已提交过偏移量，导致历史消息被跳过。

代码优化（可选） #

如果需要区分消息的来源主题，可以使用ProcessorContext获取元数据：

@Bean
public Consumer<KStream<String, String>> routeRequests() {
    return uda -> uda
        .process(() -> new FixedKeyProcessor<>() {
            private ProcessorContext context;
            @Override
            public void init(ProcessorContext context) {
                this.context = context;
            }
            @Override
            public void process(FixedKeyRecord<String, String> record) {
                System.out.printf("主题：%s, Key：%s, 内容：%s%n", 
                    context.topic(), record.key(), record.value());
            }
            @Override
            public void close() {}
        });
}

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