如何在我的 Spring Cloud Stream 项目中将传入的 headers 映射为 String 而不是 byte[]？

Question

我有一个简单的 Spring Cloud Stream 项目，使用 Spring Integration DSL 流程并使用 Kafka 活页夹。一切正常，但是来自 Kafka 的消息 header 值以 byte[].

的形式到达

这意味着我的 SI @Header 参数需要是 byte[] 类型。哪个有效，但最好将它们作为字符串（我关心的所有入站 header 都是字符串值）。

我已将 Kafka 客户端配置为使用 StringSerializer/StringDeserializer。我假设我还需要以某种方式告诉 Spring Kafka 将哪些 header 映射为字符串以及使用什么字符编码。

我显然在这里遗漏了一些东西。有什么建议吗？

Answer 1

将绑定器属性 headerMapperBeanName 设置为 DefaultKafkaHeaderMapper bean 的 bean 名称。

spring.cloud.stream.kafka.binder.headerMapperBeanName

The bean name of a KafkaHeaderMapper used for mapping spring-messaging headers to and from Kafka headers. Use this, for example, if you wish to customize the trusted packages in a DefaultKafkaHeaderMapper that uses JSON deserialization for the headers.

然后您可以指定要由映射器映射为字符串的 headers：

/**
 * Set the headers to not perform any conversion on (except {@code String} to
 * {@code byte[]} for outbound). Inbound headers that match will be mapped as
 * {@code byte[]} unless the corresponding boolean in the map value is true,
 * in which case it will be mapped as a String.
 * @param rawMappedHeaders the header names to not convert and
 * @since 2.2.5
 * @see #setCharset(Charset)
 * @see #setMapAllStringsOut(boolean)
 */
public void setRawMappedHeaders(Map<String, Boolean> rawMappedHeaders) {

如何在我的 Spring Cloud Stream 项目中将传入的 headers 映射为 String 而不是 byte[]？

How can I map incoming headers as String instead of byte[] in my Spring Cloud Stream project?

spring

spring-integration

spring-cloud-stream

spring-kafka

spring-integration-dsl