|
此版本仍在开发中，尚未被视为稳定版本。对于最新的稳定版本，请使用 Spring for Apache Kafka 3.3.0！ |
消息报头
0.11.0.0 客户端引入了对消息中标头的支持。
从版本 2.0 开始， Spring for Apache Kafka 现在支持将这些 Headers 映射到和从spring-messaging MessageHeaders.
映射的先前版本ConsumerRecord和ProducerRecord到 spring-messagingMessage<?>，其中 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值。
这DefaultKafkaHeaderMapper将 key 映射到MessageHeaders标头名称，为了支持出站消息的丰富标头类型，将执行 JSON 转换。
A "special“ 标头（键为spring_json_header_types） 包含<key>:<type>.
此标头在入站端使用，以提供每个标头值到原始类型的适当转换。
在入站端，所有 KafkaHeader实例映射到MessageHeaders.
在出站端，默认情况下，所有MessageHeaders映射，但id,timestamp以及映射到ConsumerRecord性能。
您可以通过向映射器提供模式来指定要为出站消息映射的标头。 下面的清单显示了一些示例映射：
public DefaultKafkaHeaderMapper() { (1)
...
}
public DefaultKafkaHeaderMapper(ObjectMapper objectMapper) { (2)
...
}
public DefaultKafkaHeaderMapper(String... patterns) { (3)
...
}
public DefaultKafkaHeaderMapper(ObjectMapper objectMapper, String... patterns) { (4)
...
}| 1 | 使用默认的 JacksonObjectMapper并映射大多数 headers，如示例前面所讨论的那样。 |
| 2 | 使用提供的 JacksonObjectMapper并映射大多数 headers，如示例前面所讨论的那样。 |
| 3 | 使用默认的 JacksonObjectMapper并根据提供的模式映射标头。 |
| 4 | 使用提供的 JacksonObjectMapper并根据提供的模式映射标头。 |
模式相当简单，可以包含前导通配符 （、尾随通配符或两者兼而有之（例如**.cat.*).
您可以使用前导!.
与标头名称匹配的第一个模式（无论是正的还是负的）获胜。
当您提供自己的模式时，我们建议包括!id和!timestamp，因为这些标头在入站端是只读的。
默认情况下，映射器仅反序列化java.lang和java.util.
您可以通过添加受信任的软件包来信任其他（或所有）软件包addTrustedPackages方法。
如果您收到来自不受信任的来源的消息，您可能希望只添加您信任的那些包。
要信任所有包，您可以使用mapper.addTrustedPackages("*"). |
映射String当与不知道映射器的 JSON 格式的系统通信时，原始形式的标头值非常有用。 |
从版本 2.2.5 开始，您可以指定某些字符串值 Headers 不应使用 JSON 进行 Map，而是应映射到原始 Headers 或从原始 Mapbyte[].
这AbstractKafkaHeaderMapper具有新属性;mapAllStringsOut当设置为 true 时，所有字符串值标头都将转换为byte[]使用charset属性（默认UTF-8).
此外，还有一个属性rawMappedHeaders，它是header name : boolean;如果映射包含标头名称，并且标头包含String值，它将被映射为 RAWbyte[]使用 charset.
此映射还用于映射原始传入byte[]headers 设置为String当且仅当映射值中的布尔值为true.
如果布尔值为false，或者标头名称不在映射中，并且带有true值，则传入的 Headers 会简单地映射为原始的未映射 Header。
以下测试用例说明了此机制。
@Test
public void testSpecificStringConvert() {
DefaultKafkaHeaderMapper mapper = new DefaultKafkaHeaderMapper();
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()));
}默认情况下，两个 Headers 映射器都会映射所有入站 Headers。 从版本 2.8.8 开始，模式也可以应用于入站映射。 要创建用于入站映射的映射器，请在相应的映射器上使用静态方法之一：
public static DefaultKafkaHeaderMapper forInboundOnlyWithMatchers(String... patterns) {
}
public static DefaultKafkaHeaderMapper forInboundOnlyWithMatchers(ObjectMapper objectMapper, String... patterns) {
}
public static SimpleKafkaHeaderMapper forInboundOnlyWithMatchers(String... patterns) {
}例如：
DefaultKafkaHeaderMapper inboundMapper = DefaultKafkaHeaderMapper.forInboundOnlyWithMatchers("!abc*", "*");这将排除所有以abc并包括所有其他
默认情况下，DefaultKafkaHeaderMapper用于MessagingMessageConverter和BatchMessagingMessageConverter，只要 Jackson 在 Classpath 上。
使用批处理转换器，转换后的标头在KafkaHeaders.BATCH_CONVERTED_HEADERS作为List<Map<String, Object>>其中，列表位置的 map 对应于 payload 中的数据位置。
如果没有转换器（因为 Jackson 不存在，或者它被显式设置为null），则使用者记录中的标头在KafkaHeaders.NATIVE_HEADERS页眉。
此标头是一个Headersobject（或List<Headers>如果是 Batch Converter），其中列表中的位置对应于有效载荷中的数据位置。
某些类型不适合 JSON 序列化，并且简单的toString()序列化可能是这些类型的首选。
这DefaultKafkaHeaderMapper有一个名为addToStringClasses()这允许您提供应以这种方式处理出站映射的类的名称。
在入站映射期间，它们被映射为String.
默认情况下，只有org.springframework.util.MimeType和org.springframework.http.MediaType以这种方式映射。 |
从版本 2.3 开始，简化了 String 值 Headers 的处理。
默认情况下，此类标头不再是 JSON 编码的（即它们没有封闭"..."添加）。
该类型仍会添加到 JSON_TYPES 标头中，以便接收系统可以转换回 String（从byte[]).
mapper 可以处理（解码）旧版本生成的 headers（它会检查前导）;这样，使用 2.3 的应用程序可以使用旧版本中的记录。" |
为了与早期版本兼容，请将encodeStrings自true，如果使用 2.3 的版本生成的记录可能被使用早期版本的应用程序使用。
当所有应用程序都使用 2.3 或更高版本时，可以将该属性保留为默认值false. |
@Bean
MessagingMessageConverter converter() {
MessagingMessageConverter converter = new MessagingMessageConverter();
DefaultKafkaHeaderMapper mapper = new DefaultKafkaHeaderMapper();
mapper.setEncodeStrings(true);
converter.setHeaderMapper(mapper);
return converter;
}如果使用 Spring Boot，它会自动将此转换器 bean 配置到自动配置的KafkaTemplate;否则，您应该将此转换器添加到模板中。