跳到主要内容
版本：4.1.1

消息头

Doubao 中英对照 Message Headers

0.11.0.0 客户端引入了对消息中头（header）的支持。从 2.0 版本开始，Spring for Apache Kafka 支持在这些头与 spring-messaging MessageHeaders 之间进行双向映射。

备注

早期版本将 ConsumerRecord 和 ProducerRecord 映射为 spring-messaging 的 Message<?>，其中 value 属性与 payload 互映射，其他属性（topic、partition 等）映射到头。现在仍然如此，但额外的（任意）头现在也可以被映射。

Apache Kafka 的头有一个简单的 API，如以下接口定义所示：

public interface Header {

    String key();

    byte[] value();

}

提供了 KafkaHeaderMapper 策略，用于在 Kafka Headers 与 MessageHeaders 之间映射头条目。其接口定义如下：

public interface KafkaHeaderMapper {

    void fromHeaders(MessageHeaders headers, Headers target);

    void toHeaders(Headers source, Map<String, Object> target);

}

SimpleKafkaHeaderMapper 将原始头映射为 byte[]，并提供了转换为 String 值的配置选项。

JsonKafkaHeaderMapper 将键映射为 MessageHeaders 头名称，并且为了支持出站消息的富头类型，会执行 JSON 转换。一个 "special" 头（其键为 spring_json_header_types）包含一个 <key>:<type> 的 JSON 映射。该头在入站侧用于将每个头值适当地转换回原始类型。

在入站侧，所有 Kafka Header 实例都被映射为 MessageHeaders。在出站侧，默认情况下所有 MessageHeaders 都会被映射，但 id、timestamp 以及映射到 ConsumerRecord 属性的那些头除外。

你可以通过向映射器提供模式（pattern）来指定哪些头要为出站消息进行映射。以下清单展示了若干示例映射：

public JsonKafkaHeaderMapper() { 1
    ...
}

public JsonKafkaHeaderMapper(ObjectMapper objectMapper) { 2
    ...
}

public JsonKafkaHeaderMapper(String... patterns) { 3
    ...
}

public JsonKafkaHeaderMapper(ObjectMapper objectMapper, String... patterns) { 4
    ...
}

  • 使用默认的 Jackson ObjectMapper，并映射大多数头，如示例之前所讨论的。

  • 使用所提供的 Jackson ObjectMapper，并映射大多数头，如示例之前所讨论的。

  • 使用默认的 Jackson ObjectMapper，并根据所提供的模式映射头。

  • 使用所提供的 Jackson ObjectMapper，并根据所提供的模式映射头。

模式相当简单，可以包含前导通配符（*）、后导通配符，或两者兼有（例如 *.cat.*）。你可以用前导 ! 对模式取反。第一个匹配到头名称的模式（无论肯定还是否定）胜出。

当你提供自己的模式时，我们建议包含 !id 和 !timestamp，因为这些头在入站侧是只读的。

important

默认情况下，映射器仅反序列化那些包名恰好为 java.lang、java.net、java.util 或 org.springframework.util 的类。你可以通过 addTrustedPackages方法信任更多包。信任 仅按精确包名：只有当类的声明包逐字出现在信任列表中时，该类才被信任。子包 不会 被传递信任；每个包都必须显式注册。如果你从不信任的来源接收消息，请只注册你需要的具体包。要无限制地信任所有包，可使用 mapper.addTrustedPackages("*")——但对于来自不受信任来源的消息不建议这样做。位于非信任包中的类会作为 NonTrustedHeaderType 值返回给应用程序，而不进行反序列化。

// Trusts com.example but NOT com.example.events
mapper.addTrustedPackages("com.example");

// Both must be listed explicitly
mapper.addTrustedPackages("com.example", "com.example.events");
备注

在与不了解映射器 JSON 格式的系统通信时，以原始形式映射 String 头值会很有用。

从 2.2.5 版本开始，你可以指定某些字符串值的头不应使用 JSON 映射，而应与原始 byte[] 互映射。AbstractKafkaHeaderMapper 新增了一些属性；mapAllStringsOut 设置为 true 时，所有字符串值的头都将使用 charset 属性（默认 UTF-8）转换为 byte[]。此外，还有一个 rawMappedHeaders 属性，它是一个 头名称 : boolean 的映射；如果该映射包含某个头名称，且该头包含一个 String 值，则它将使用该字符集作为原始 byte[] 进行映射。该映射还用于仅当映射值中的 boolean 为 true 时，才使用该字符集将传入的原始 byte[] 头映射为 String。如果 boolean 为 false，或者该头名称不在映射中且值为 true，则传入的头仅作为未映射的原始头进行映射。

以下测试用例说明了这一机制。

@Test
public void testSpecificStringConvert() {
    JsonKafkaHeaderMapper mapper = new JsonKafkaHeaderMapper();
    Map<String, Boolean> rawMappedHeaders = new HashMap<>();
    rawMappedHeaders.put("thisOnesAString", true);
    rawMappedHeaders.put("thisOnesBytes", false);
    mapper.setRawMappedHeaders(rawMappedHeaders);
    Map<String, Object> headersMap = new HashMap<>();
    headersMap.put("thisOnesAString", "thing1");
    headersMap.put("thisOnesBytes", "thing2");
    headersMap.put("alwaysRaw", "thing3".getBytes());
    MessageHeaders headers = new MessageHeaders(headersMap);
    Headers target = new RecordHeaders();
    mapper.fromHeaders(headers, target);
    assertThat(target).containsExactlyInAnyOrder(
            new RecordHeader("thisOnesAString", "thing1".getBytes()),
            new RecordHeader("thisOnesBytes", "thing2".getBytes()),
            new RecordHeader("alwaysRaw", "thing3".getBytes()));
    headersMap.clear();
    mapper.toHeaders(target, headersMap);
    assertThat(headersMap).contains(
            entry("thisOnesAString", "thing1"),
            entry("thisOnesBytes", "thing2".getBytes()),
            entry("alwaysRaw", "thing3".getBytes()));
}

默认情况下，两个头映射器都会映射所有入站头。从 2.8.8 版本开始，这些模式也可以应用于入站映射。要为入站映射创建映射器，请使用相应映射器上的静态方法之一：

public static JsonKafkaHeaderMapper forInboundOnlyWithMatchers(String... patterns) {
}

public static JsonKafkaHeaderMapper forInboundOnlyWithMatchers(ObjectMapper objectMapper, String... patterns) {
}

public static SimpleKafkaHeaderMapper forInboundOnlyWithMatchers(String... patterns) {
}

例如：

JsonKafkaHeaderMapper inboundMapper = JsonKafkaHeaderMapper.forInboundOnlyWithMatchers("!abc*", "*");

这将排除所有以 abc 开头的头，并包含所有其他头。

默认情况下，只要 Jackson 在 classpath 上，JsonKafkaHeaderMapper 就会用于 MessagingMessageConverter 和 BatchMessagingMessageConverter。

对于批量转换器，转换后的头可在 KafkaHeaders.BATCH_CONVERTED_HEADERS 中以 List<Map<String, Object>> 的形式获取，其中列表中某个位置的 map 对应于载荷中的数据位置。

如果没有转换器（要么因为 Jackson 不存在，要么被显式设置为 null），来自消费者记录的头会以未转换的形式提供在 KafkaHeaders.NATIVE_HEADERS 头中。该头是一个 Headers 对象（在批量转换器的情况下为 List<Headers>），其中列表中的位置对应于载荷中的数据位置。

important

某些类型不适合 JSON 序列化，对这些类型可能更倾向于使用简单的 toString() 序列化。JsonKafkaHeaderMapper 有一个名为 addToStringClasses() 的方法，可让你提供应在此出站映射方式下处理的类名。在入站映射期间，它们被映射为 String。默认情况下，只有 org.springframework.util.MimeType 和 org.springframework.http.MediaType 以这种方式映射。

备注

从 2.3 版本开始，字符串值头的处理得到简化。这类头默认不再进行 JSON 编码（即不再添加包裹的 "..."）。其类型仍会被添加到 JSON_TYPES 头中，以便接收系统能将其转换回 String（从 byte[]）。映射器能够处理（解码）由旧版本产生的头（它会检查前导 "）；这样，使用 2.3 的应用程序可以消费来自旧版本的记录。

important

为了与早期版本兼容，如果由使用 2.3 的版本产生的记录可能被使用早期版本的应用程序消费，请将 encodeStrings 设置为 true。当所有应用程序都使用 2.3 或更高版本时，你可以将该属性保留为其默认值 false。

@Bean
MessagingMessageConverter converter() {
    MessagingMessageConverter converter = new MessagingMessageConverter();
    JsonKafkaHeaderMapper mapper = new JsonKafkaHeaderMapper();
    mapper.setEncodeStrings(true);
    converter.setHeaderMapper(mapper);
    return converter;
}

如果使用 Spring Boot，它会自动将此转换器 bean 配置到自动配置的 KafkaTemplate 中；否则你应将此转换器添加到模板。

支持多值头映射​

从 4.0 开始，支持多值头映射，即同一个逻辑头键在一条 Kafka 记录中出现多次。

默认情况下，HeaderMapper 不会 创建多个同名的 Kafka 头。相反，当遇到集合值（例如 List<byte[]>）时，它会将整个集合序列化为 一个 Kafka 头，其值为一个 JSON 数组。

  • 生产者端： JsonKafkaHeaderMapper 写入 JSON 字节，而 SimpleKafkaHeaderMapper 会忽略它。
  • 消费者端： 映射器将该头暴露为单个值——最后一次出现者胜出；先前的重复项被静默丢弃。

要保留每个独立的头，需要显式注册将该头指定为多值的模式。

JsonKafkaHeaderMapper#setMultiValueHeaderPatterns(String…​ patterns) 接受一个模式列表，这些模式可以是通配符表达式或精确头名称。

JsonKafkaHeaderMapper mapper = new JsonKafkaHeaderMapper();

// Explicit header names
mapper.setMultiValueHeaderPatterns("test-multi-value1", "test-multi-value2");

// Wildcard patterns for test-multi-value1, test-multi-value2
mapper.setMultiValueHeaderPatterns("test-multi-*");

任何名称与所提供模式之一匹配的头都

  • 生产者端： 写为独立的 Kafka 头，每个元素一个。
  • 消费者端： 收集到一个 List<?> 中，其中包含各个头值；每个元素在经过所配置的 HeaderMapper 执行常规反序列化或类型转换 之后 返回给应用程序。
备注

不支持正则表达式；简单模式中只允许使用 * 通配符——支持直接相等以及诸如 xxx*、*xxx、*xxx*、xxx*yyy 之类的形式。

important

在 生产者端，当 JsonKafkaHeaderMapper 序列化多值头时，该集合中的每个元素都必须是单一 Java 类型——例如，在单个头键下混用 String 和 byte[] 值会导致转换错误。