@Header 和 Spring Stream 函数式编程模型

Question

有没有办法在以下 kafka 消费者代码中使用 @Header ？我正在使用 Spring Cloud Stream (Kafka Stream binder implementation)，并且在我的实现之后使用 functional model 例如。

@Bean
public Consumer<KStream<String, Pojo>> process() {
    return messages -> messages.foreach((k, v) -> process(v));
}

如果使用 Spring for apache kafka 那么这可以很简单

@KafkaListener(topics = "${mytopicname}", clientIdPrefix = "${myprefix}", errorHandler = "customEventErrorHandler")
public void processEvent(@Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key,
                         @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
                         @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                         @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long ts
                         @Valid Pojo pojo) {
    ...
    // use headers here
    ...
}

Answer 1

没有； Kafka Streams binder 不基于 Spring Messaging。

您可以在添加到您的信息流中的Transformer（通过ProcessorContext）中访问标题、主题等。

您可以将 Kafka 消息通道绑定器与

@Bean
public Consumer<Message<Pojo>> process() {
    return message -> ...
}