spring - 使用 Spring Kafka 添加自定义 header

Question

我计划使用 Spring Kafka 客户端在 Spring Boot 应用程序中使用 kafka 设置并生成消息。我在 Kafka 0.11 中看到了对自定义 header 的支持，详见 here .虽然它适用于本地 Kafka 生产者和消费者，但我看不到在 Spring Kafka 中添加/读取自定义 header 的支持。

我正在尝试根据我希望存储在消息 header 中的重试计数为消息实现 DLQ，而无需解析有效负载。

Answer 1

好吧，Spring Kafka 从 2.0 版开始提供 header 支持:https://docs.spring.io/spring-kafka/docs/2.1.2.RELEASE/reference/html/_reference.html#headers

你可以拥有 KafkaHeaderMapper实例并使用它来填充 Message 的标题在通过 KafkaTemplate.send(Message<?> message) 发送之前.或者你可以使用普通的 KafkaTemplate.send(ProducerRecord<K, V> record) .

当您使用 KafkaMessageListenerContainer 接收记录时, KafkaHeaderMapper可以通过 MessagingMessageConverter 提供注入(inject)RecordMessagingMessageListenerAdapter .

因此，任何自定义 header 都可以通过任何一种方式传输。