java - Spring Cloud Stream 确定主题 Kafka 消息来自何处

Question

我的@StreamListener监听spring.cloud.stream.bindings.input.destination属性中指定的多个Kafka主题(总共10个主题)。

有什么方法可以确定主题消息来自 @StreamListener 注解方法内部吗？我不想单独为所有主题配置绑定(bind)并编写 10 个几乎相同的 StreamListener。

Answer 1

参见KafkaHeaders. RECEIVED_TOPIC您可以在 @StreamListener 上指定单独的参数方法。 Spring Kafka 文档中是这样的:

@KafkaListener(id = "qux", topicPattern = "myTopic1")
public void listen(@Payload String foo,
    @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) Integer key,
    @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
    @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
    @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long ts
        ) {
    ...
}

https://docs.spring.io/spring-kafka/reference/htmlsingle/#kafka-listener-annotation