版本：4.1.1

本站内所有文档均为中英对照文档，点击中文可以显示英文。此提醒连续关闭5次后，将不再显示。

文档内容来源于 spring.io，由 springdoc.tech 翻译，版权归属于 SPRING.IO (Broadcom. Inc)。可供个人学习、研究，未经许可，不得进行转载或用于商业行为。

Apache Kafka Streams 支持

Doubao 中英对照 Apache Kafka Streams Support

从 1.1.4 版本开始，Spring for Apache Kafka 为 Kafka Streams 提供一等支持。要在 Spring 应用中使用它，类路径上必须存在 kafka-streams jar。它是 Spring for Apache Kafka 项目的可选依赖，不会被传递性下载。

基础知识

参考性的 Apache Kafka Streams 文档建议按以下方式使用该 API：

// Use the builders to define the actual processing topology, e.g. to specify
// from which input topics to read, which stream operations (filter, map, etc.)
// should be called, and so on.

StreamsBuilder builder = ...;  // when using the Kafka Streams DSL

// Use the configuration to tell your application where the Kafka cluster is,
// which serializers/deserializers to use by default, to specify security settings,
// and so on.
StreamsConfig config = ...;

KafkaStreams streams = new KafkaStreams(builder, config);

// Start the Kafka Streams instance
streams.start();

// Stop the Kafka Streams instance
streams.close();

因此，我们有两个主要组件：

StreamsBuilder：拥有用于构建 KStream（或 KTable）实例的 API。
KafkaStreams：用于管理这些实例的生命周期。

备注

由同一个 StreamsBuilder 暴露给某个 KafkaStreams 实例的所有 KStream 实例会同时启动和停止，即使它们具有不同的逻辑。换句话说，由一个 StreamsBuilder 定义的所有流都受同一个生命周期控制。一旦 KafkaStreams 实例被 streams.close() 关闭，它就不能再重启。相反，必须创建一个新的 KafkaStreams 实例来重启流处理。

Spring 管理

为了从 Spring 应用上下文的角度简化 Kafka Streams 的使用，并通过容器使用生命周期管理，Spring for Apache Kafka 引入了 StreamsBuilderFactoryBean。这是一个 AbstractFactoryBean 实现，用于将 StreamsBuilder 单例实例暴露为一个 bean。下面的示例创建了这样一个 bean：

@Bean
public FactoryBean<StreamsBuilder> myKStreamBuilder(KafkaStreamsConfiguration streamsConfig) {
    return new StreamsBuilderFactoryBean(streamsConfig);
}

important

从 2.2 版本开始，流配置现在以 KafkaStreamsConfiguration 对象而非 StreamsConfig 提供。

StreamsBuilderFactoryBean 还实现了 SmartLifecycle，以管理内部 KafkaStreams 实例的生命周期。与 Kafka Streams API 类似，你必须在启动 KafkaStreams 之前定义 KStream 实例。这同样适用于 Kafka Streams 的 Spring API。因此，当在 StreamsBuilderFactoryBean 上使用默认的 autoStartup = true 时，你必须在应用上下文刷新之前在 StreamsBuilder 上声明 KStream 实例。例如，KStream 可以是一个常规的 bean 定义，而 Kafka Streams API 的使用不受任何影响。下面的示例展示了如何做到这一点：

@Bean
public KStream<?, ?> kStream(StreamsBuilder kStreamBuilder) {
    KStream<Integer, String> stream = kStreamBuilder.stream(STREAMING_TOPIC1);
    // Fluent KStream API
    return stream;
}

如果你希望手动控制生命周期（例如，按某种条件停止和启动），可以直接使用工厂 bean（&）前缀引用 StreamsBuilderFactoryBean bean。由于 StreamsBuilderFactoryBean 使用其内部的 KafkaStreams 实例，停止并再次重启它是安全的。每次 start() 时都会创建一个新的 KafkaStreams。如果你希望分别控制 KStream 实例的生命周期，也可以考虑使用不同的 StreamsBuilderFactoryBean 实例。

你还可以在 StreamsBuilderFactoryBean 上指定 KafkaStreams.StateListener、Thread.UncaughtExceptionHandler 和 StateRestoreListener 选项，它们会被委托给内部的 KafkaStreams 实例。

此外，除了在 StreamsBuilderFactoryBean 上间接设置这些选项之外，你还可以使用 KafkaStreamsCustomizer 回调接口来：

（自 _2.1.5 版本起）使用 customize(KafkaStreams) 配置内部 KafkaStreams 实例
（自 _3.3.0 版本起）使用 initKafkaStreams(Topology, Properties, KafkaClientSupplier) 实例化一个自定义的 KafkaStreams 实现

注意，KafkaStreamsCustomizer 会覆盖 StreamsBuilderFactoryBean 提供的选项。

如果你需要直接执行某些 KafkaStreams 操作，可以使用 StreamsBuilderFactoryBean.getKafkaStreams() 访问该内部 KafkaStreams 实例。

你可以按类型自动装配 StreamsBuilderFactoryBean bean，但应确保在 bean 定义中使用完整类型，如下例所示：

@Bean
public StreamsBuilderFactoryBean myKStreamBuilder(KafkaStreamsConfiguration streamsConfig) {
    return new StreamsBuilderFactoryBean(streamsConfig);
}
...
@Autowired
private StreamsBuilderFactoryBean myKStreamBuilderFactoryBean;

或者，如果你使用接口 bean 定义，可以添加 @Qualifier 按名称注入。下面的示例展示了如何做到这一点：

@Bean
public FactoryBean<StreamsBuilder> myKStreamBuilder(KafkaStreamsConfiguration streamsConfig) {
    return new StreamsBuilderFactoryBean(streamsConfig);
}
...
@Autowired
@Qualifier("&myKStreamBuilder")
private StreamsBuilderFactoryBean myKStreamBuilderFactoryBean;

从 2.4.1 版本开始，工厂 bean 新增了一个类型为 KafkaStreamsInfrastructureCustomizer 的属性 infrastructureCustomizer；这允许在流创建之前定制 StreamsBuilder（例如添加一个状态存储）和/或 Topology。

public interface KafkaStreamsInfrastructureCustomizer {

    void configureBuilder(StreamsBuilder builder);

    void configureTopology(Topology topology);

}

提供了默认的空操作实现，以避免在不需要其中一个方法时必须实现两个方法。

当你需要应用多个定制器时，提供了一个 CompositeKafkaStreamsInfrastructureCustomizer。

KafkaStreams Micrometer 支持

在 2.5.3 版本中引入，你可以配置一个 KafkaStreamsMicrometerListener，为工厂 bean 管理的 KafkaStreams 对象自动注册 micrometer 计量器：

streamsBuilderFactoryBean.addListener(new KafkaStreamsMicrometerListener(meterRegistry,
        Collections.singletonList(new ImmutableTag("customTag", "customTagValue"))));

Streams JSON 序列化与反序列化

对于以 JSON 格式读写主题或状态存储时的数据序列化与反序列化，Spring for Apache Kafka 提供了一个 JacksonJsonSerde 实现，它使用 JSON，委托给序列化、反序列化与消息转换中描述的 JacksonJsonSerializer 和 JacksonJsonDeserializer。JacksonJsonSerde 实现通过其构造器提供相同的配置选项（目标类型或 ObjectMapper）。在下面的示例中，我们使用 JacksonJsonSerde 来序列化和反序列化某个 Kafka 流的 Cat 负载（JacksonJsonSerde 可以在任何需要该实例的地方以类似方式使用）：

stream.through(Serdes.Integer(), new JacksonJsonSerde<>(Cat.class), "cats");

当为在生产者/消费者工厂中使用而以编程方式构建序列化器/反序列化器时，自 2.3 版本起，你可以使用流式 API，它简化了配置。

stream.through(
    new JacksonJsonSerde<>(MyKeyType.class)
        .forKeys()
        .noTypeInfo(),
    new JacksonJsonSerde<>(MyValueType.class)
        .noTypeInfo(),
    "myTypes");

使用 `KafkaStreamBrancher`

KafkaStreamBrancher 类引入了一种在 KStream 之上构建条件分支的更便捷方式。

考虑下面这个不使用 KafkaStreamBrancher 的示例：

KStream<String, String>[] branches = builder.stream("source").branch(
        (key, value) -> value.contains("A"),
        (key, value) -> value.contains("B"),
        (key, value) -> true
);
branches[0].to("A");
branches[1].to("B");
branches[2].to("C");

下面的示例使用 KafkaStreamBrancher：

new KafkaStreamBrancher<String, String>()
        .branch((key, value) -> value.contains("A"), ks -> ks.to("A"))
        .branch((key, value) -> value.contains("B"), ks -> ks.to("B"))
        //default branch should not necessarily be defined in the end of the chain!
        .defaultBranch(ks -> ks.to("C"))
        .onTopOf(builder.stream("source"));
        //onTopOf method returns the provided stream so we can continue with method chaining

配置

要配置 Kafka Streams 环境，StreamsBuilderFactoryBean 需要一个 KafkaStreamsConfiguration 实例。所有可用选项请参阅 Apache Kafka 文档。

important

从 2.2 版本开始，流配置现在以 KafkaStreamsConfiguration 对象而非 StreamsConfig 提供。

为了避免大多数情况下的样板代码（尤其是在开发微服务时），Spring for Apache Kafka 提供了 @EnableKafkaStreams 注解，你应将其放在一个 @Configuration 类上。你只需要声明一个名为 defaultKafkaStreamsConfig 的 KafkaStreamsConfiguration bean。一个名为 defaultKafkaStreamsBuilder 的 StreamsBuilderFactoryBean bean 会自动在应用上下文中声明。你也可以声明和使用任何额外的 StreamsBuilderFactoryBean bean。你可以通过提供一个实现 StreamsBuilderFactoryBeanConfigurer 的 bean 来对该 bean 进行额外定制。如果有多个这样的 bean，它们将按其 Ordered.order 属性应用。

清理与停止配置

当工厂停止时，会以两个参数调用 KafkaStreams.close()：

closeTimeout：等待线程关闭的时长（默认为 DEFAULT_CLOSE_TIMEOUT，设为 10 秒）。可以使用 StreamsBuilderFactoryBean.setCloseTimeout() 配置。
leaveGroupOnClose：是否在关闭时触发消费者离开组（默认为 false）。可以使用 StreamsBuilderFactoryBean.setLeaveGroupOnClose() 配置。

默认情况下，当工厂 bean 停止时，会调用 KafkaStreams.cleanUp() 方法。从 2.1.2 版本开始，工厂 bean 新增了一些构造器，接受一个 CleanupConfig 对象，其属性让你控制是否在 start() 或 stop() 期间调用 cleanUp() 方法，或两者都不调用。从 2.7 版本开始，默认是从不清理本地状态。

组协议配置

从 4.1 版本开始，你可以通过设置 groupProtocol 属性来配置底层 Kafka Streams 消费者所使用的消费者组协议。这让你可以显式管理消费者是依赖 classic 协议，还是选择 KIP-1071 中引入的较新的 streams 组协议（Streams 的服务端重平衡，在 Kafka 4.2 中 GA）。虽然新集群上默认启用此协议，但现有集群可以通过此属性选择启用。可以使用 StreamsBuilderFactoryBean.setGroupProtocol() 配置。

下面的示例展示了如何使用 StreamsBuilderFactoryBeanConfigurer 配置组协议：

@Bean
public StreamsBuilderFactoryBeanConfigurer groupProtocolConfigurer() {
    return fb -> fb.setGroupProtocol(GroupProtocol.STREAMS);
}

Header Enricher

3.0 版本新增了 ContextualProcessor 的扩展 HeaderEnricherProcessor；它提供与已弃用的、实现了已弃用 Transformer 接口的 HeaderEnricher 相同的功能。这可用于在流处理中添加头；头值是 SpEL 表达式；表达式求值的根对象有 3 个属性：

record — org.apache.kafka.streams.processor.api.Record（key、value、timestamp、headers）
key — 当前记录的键
value — 当前记录的值
context — ProcessorContext，可访问当前记录元数据

表达式必须返回 byte[] 或 String（后者将使用 UTF-8 转换为 byte[]）。

要在流中使用 enricher：

.process(() -> new HeaderEnricherProcessor(expressions))

该处理器不更改 key 或 value；它只是添加头。

important

每条记录都需要一个新实例。

.process(() -> new HeaderEnricherProcessor<..., ...>(expressionMap))

下面是一个简单示例，添加一个字面量头和一个变量头：

Map<String, Expression> headers = new HashMap<>();
headers.put("header1", new LiteralExpression("value1"));
SpelExpressionParser parser = new SpelExpressionParser();
headers.put("header2", parser.parseExpression("record.timestamp() + ' @' + record.offset()"));
ProcessorSupplier supplier = () -> new HeaderEnricher<String, String>(headers);
KStream<String, String> stream = builder.stream(INPUT);
stream
        .process(() -> supplier)
        .to(OUTPUT);

`MessagingProcessor`

3.0 版本新增了 ContextualProcessor 的扩展 MessagingProcessor，提供与已弃用的、实现了已弃用 Transformer 接口的 MessagingTransformer 相同的功能。这允许 Kafka Streams 拓扑与 Spring Messaging 组件（例如 Spring Integration 流）交互。该转换器需要一个 MessagingFunction 的实现。

@FunctionalInterface
public interface MessagingFunction {

    Message<?> exchange(Message<?> message);

}

Spring Integration 会使用其 GatewayProxyFactoryBean 自动提供一个实现。它还需要一个 MessagingMessageConverter 来将键、值和元数据（包括头）与 Spring Messaging Message<?> 相互转换。更多信息请参阅从 KStream 调用 Spring Integration 流。

恢复策略

框架提供以下异常处理器，它们遵循相同的恢复策略：

RecoveringDeserializationExceptionHandler
RecoveringProcessingExceptionHandler
RecoveringProductionExceptionHandler

恢复策略按优先级顺序如下：

如果定义了 KafkaStreamsDeadLetterDestinationResolver，恢复流并使用原生 Kafka Streams DLQ 将失败记录转发到解析出的主题分区。
如果定义了 errors.dead.letter.queue.topic.name 并设为某个主题名，恢复流并使用原生 Kafka Streams DLQ 将失败记录转发到该主题。
如果定义了 ConsumerRecordRecoverer 实现，调用它并恢复流，不产生死信记录，因为处理被委托给了 ConsumerRecordRecoverer。例如，可以使用所提供的 DeadLetterPublishingRecoverer 实现。
使流失败，不产生死信记录。

当死信记录被发布到死信主题时（无论是通过原生 Kafka Streams DLQ 还是通过 ConsumerRecordRecoverer 实现）：

使用源原始键、源原始值和源头来构建记录。
该记录会被添加Spring for Apache Kafka DLT 头。

从反序列化异常中恢复

2.3 版本引入了 RecoveringDeserializationExceptionHandler，当发生反序列化异常时它可以采取一些措施。它实现了 DeserializationExceptionHandler 接口（详情请参阅 Kafka 文档），并遵循 Spring for Apache Kafka 恢复策略。

要配置 RecoveringDeserializationExceptionHandler，请在你的流配置中添加以下属性：

@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
public KafkaStreamsConfiguration kStreamsConfigs() {
    Map<String, Object> props = new HashMap<>();
    ...
    props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
            RecoveringDeserializationExceptionHandler.class);
    ...
    return new KafkaStreamsConfiguration(props);
}

从处理异常中恢复

4.1 版本引入了 RecoveringProcessingExceptionHandler，当流处理期间发生异常时它可以采取一些措施。它实现了由 KIP-1033 引入的 ProcessingExceptionHandler 接口（详情请参阅 Kafka 文档），并遵循 Spring for Apache Kafka 恢复策略。

要启用 RecoveringProcessingExceptionHandler，请在你的流配置中添加以下属性：

@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
public KafkaStreamsConfiguration kStreamsConfigs() {
    Map<String, Object> props = new HashMap<>();
    ...
    props.put(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG,
            RecoveringProcessingExceptionHandler.class);
    ...
}

从生产异常中恢复

4.1 版本引入了 RecoveringProductionExceptionHandler，当记录生产或序列化期间发生异常时它可以采取一些措施。它实现了 ProductionExceptionHandler 接口（详情请参阅 Kafka 文档），并遵循 Spring for Apache Kafka 恢复策略。

要配置 RecoveringProductionExceptionHandler，请在你的流配置中添加以下属性：

@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
public KafkaStreamsConfiguration kStreamsConfigs() {
    Map<String, Object> props = new HashMap<>();
    ...
    props.put(StreamsConfig.PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG,
            RecoveringProductionExceptionHandler.class);
    ...
}

死信发布 Recoverer

框架提供了 DeadLetterPublishingRecoverer，它将失败记录发送到死信主题。关于该 recoverer 的更多信息请参阅发布死信记录。

要为异常处理器配置 recoverer，请在你的流配置中添加以下属性：

@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
public KafkaStreamsConfiguration kStreamsConfigs() {
    Map<String, Object> props = new HashMap<>();
    ...
    props.put(StreamsConfig.DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
            RecoveringDeserializationExceptionHandler.class);
    props.put(RecoveringDeserializationExceptionHandler.RECOVERER, recoverer());
    ...
    return new KafkaStreamsConfiguration(props);
}

@Bean
public DeadLetterPublishingRecoverer recoverer() {
    return new DeadLetterPublishingRecoverer(kafkaTemplate(),
            (record, ex) -> new TopicPartition("recovererDLQ", -1));
}

当然，recoverer() bean 可以是你自己的 ConsumerRecordRecoverer 实现。

死信目的地解析器

可以定义一个 KafkaStreamsDeadLetterDestinationResolver 类型的 bean，以在异常处理器中启用原生 DLQ 路由。

它基于错误处理器上下文、失败处理器的输入记录以及抛出的异常，来决定要使用的 DLQ 主题名和分区解析逻辑：

@Bean
public KafkaStreamsDeadLetterDestinationResolver resolver() {
    return (context, record, ex) -> {
        if (ex instanceof FooException) return new TopicPartition("dlqTopic1", -1);
        if (record instanceof Foo) return new TopicPartition("dlqTopic2", -1);
        if (context.processorNodeId().equals("processor-1")) return new TopicPartition("dlqTopic3", -1);
        return new TopicPartition("defaultDlqTopic", -1);
    };
}

负数分区号表示该分区将由默认分区器决定。

然后可以按如下方式将 KafkaStreamsDeadLetterDestinationResolver 注入到异常处理器中：

@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
public KafkaStreamsConfiguration kStreamsConfigs() {
    Map<String, Object> props = new HashMap<>();
    ...
    props.put(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG,
            RecoveringProcessingExceptionHandler.class);
    props.put(RecoveringProcessingExceptionHandler.DLQ_DESTINATION_RESOLVER, resolver());
    ...
    return new KafkaStreamsConfiguration(props);
}

死信队列主题名属性

Kafka Streams 属性 errors.dead.letter.queue.topic.name 可以被定义并设为某个主题名，以在异常处理器中启用原生 DLQ 路由。这直接指定了启用的异常处理器将把所有失败记录路由到的 DLQ 主题。

或者，也可以通过 StreamsBuilderFactoryBean 以编程方式设置：

@Bean
public StreamsBuilderFactoryBeanConfigurer streamsBuilderFactoryBeanConfigurer() {
    return sfb -> sfb.setDeadLetterTopicName("deadLetterQueueTopic");
}

交互式查询支持

从 3.2 版本开始，Spring for Apache Kafka 提供了 Kafka Streams 中交互式查询所需的基本设施。交互式查询在有状态的 Kafka Streams 应用中很有用，因为它们提供了一种持续查询应用中有状态存储的方式。因此，如果一个应用希望将所考虑系统的当前视图物化，交互式查询提供了一种实现方式。要了解更多关于交互式查询的信息，请参阅这篇文章。Spring for Apache Kafka 中的支持围绕一个名为 KafkaStreamsInteractiveQueryService 的 API 展开，它是 Kafka Streams 库中交互式查询 API 的一个门面。应用程序可以创建该服务的实例作为 bean，然后按名称用它检索状态存储。

下面的代码片段展示了一个示例。

@Bean
public KafkaStreamsInteractiveQueryService kafkaStreamsInteractiveQueryService(StreamsBuilderFactoryBean streamsBuilderFactoryBean) {
    final KafkaStreamsInteractiveQueryService kafkaStreamsInteractiveQueryService =
            new KafkaStreamsInteractiveQueryService(streamsBuilderFactoryBean);
    return kafkaStreamsInteractiveQueryService;
}

假设一个 Kafka Streams 应用有一个名为 app-store 的状态存储，那么可以通过 KafkaStreamsInteractiveQuery API 检索该存储，如下所示。

@Autowired
private KafkaStreamsInteractiveQueryService interactiveQueryService;

ReadOnlyKeyValueStore<Object, Object>  appStore = interactiveQueryService.retrieveQueryableStore("app-store", QueryableStoreTypes.keyValueStore());

一旦应用获得了对状态存储的访问权，就可以从中查询键值信息。

在这种情况下，应用所使用的状态存储是一个只读键值存储。Kafka Streams 应用还可以使用其他类型的状态存储。例如，如果应用希望查询一个基于窗口的存储，它可以在 Kafka Streams 应用业务逻辑中构建该存储，然后再检索它。正因如此，KafkaStreamsInteractiveQueryService 中检索可查询存储的 API 具有一个泛型存储类型签名，以便最终用户可以指定正确的类型。

下面是该 API 的类型签名。

public <T> T retrieveQueryableStore(String storeName, QueryableStoreType<T> storeType)

调用此方法时，用户可以像上面的示例中那样，显式请求正确的状态存储类型。

重试状态存储检索

当使用 KafkaStreamsInteractiveQueryService 尝试检索状态存储时，有可能由于各种原因找不到该状态存储。如果这些原因是暂时的，KafkaStreamsInteractiveQueryService 通过允许注入自定义的 RetryTemplate 提供了重试状态存储检索的选项。默认情况下，KafkaStreamsInteractiveQueryService 中使用的 RetryTemplate 最大尝试次数为三次，固定退避为一秒。

下面是如何将最大尝试次数为十次的自定义 RetryTemplate 注入到 KafkaStreamsInteractiveQueryService 中。

@Bean
public KafkaStreamsInteractiveQueryService kafkaStreamsInteractiveQueryService(StreamsBuilderFactoryBean streamsBuilderFactoryBean) {
    final KafkaStreamsInteractiveQueryService kafkaStreamsInteractiveQueryService =
            new KafkaStreamsInteractiveQueryService(streamsBuilderFactoryBean);
    RetryTemplate retryTemplate = new RetryTemplate();
    retryTemplate.setBackOffPolicy(new FixedBackOffPolicy());
    RetryPolicy retryPolicy = new SimpleRetryPolicy(10);
    retryTemplate.setRetryPolicy(retryPolicy);
    kafkaStreamsInteractiveQueryService.setRetryTemplate(retryTemplate);
    return kafkaStreamsInteractiveQueryService;
}

查询远程状态存储

上面展示的用于检索状态存储的 API —— retrieveQueryableStore 是为本地可用的键值状态存储设计的。在生产环境中，Kafka Streams 应用很可能基于分区数进行分布式部署。如果一个主题有四个分区，并且运行着同一个 Kafka Streams 处理器的四个实例，那么每个实例可能负责处理该主题的一个分区。在这种场景下，调用 retrieveQueryableStore 可能无法给出某个实例正在寻找的正确结果，尽管它可能返回一个有效的存储。假设这个有四个分区的主题包含关于各种键的数据，并且单个分区始终负责某个特定键。如果正在调用 retrieveQueryableStore 的实例所寻找的键并不由该实例托管，那么它将收不到任何数据。这是因为当前的 Kafka Streams 实例对这个键一无所知。为解决这个问题，调用实例首先需要获取托管该特定键的 Kafka Streams 处理器实例的主机信息。可以从同一 application.id 下的任何 Kafka Streams 实例中检索到该信息，如下所示。

@Autowired
private KafkaStreamsInteractiveQueryService interactiveQueryService;

HostInfo kafkaStreamsApplicationHostInfo = this.interactiveQueryService.getKafkaStreamsApplicationHostInfo("app-store", 12345, new IntegerSerializer());

在上面的示例代码中，调用实例正在从名为 app-store 的状态存储中查询某个特定键 12345。该 API 还需要一个相应的键序列化器，在本例中是 IntegerSerializer。Kafka Streams 会遍历同一 application.id 下的所有实例，尝试找出哪个实例托管了这个特定键。一旦找到，它会以 HostInfo 对象的形式返回该主机信息。

该 API 的形式如下：

public <K> HostInfo getKafkaStreamsApplicationHostInfo(String store, K key, Serializer<K> serializer)

当以这种分布式方式使用同一 application.id 的多个 Kafka Streams 处理器实例时，应用应提供一个 RPC 层，以便通过 RPC 端点（例如 REST 端点）查询状态存储。关于这一点的更多细节请参阅这篇文章。使用 Spring for Apache Kafka 时，利用 spring-web 技术添加一个基于 Spring 的 REST 端点非常容易。一旦有了 REST 端点，就可以用它从任何 Kafka Streams 实例查询状态存储，前提是该实例已知托管该键的 HostInfo。

如果托管该键的实例就是当前实例，那么应用不需要调用 RPC 机制，而是进行一次 JVM 内调用。然而问题在于，应用可能不知道发起调用的实例就是托管该键的实例，因为某个特定服务器可能因消费者重平衡而丢失了一个分区。为解决这个问题，KafkaStreamsInteractiveQueryService 提供了一个便捷 API，通过方法 getCurrentKafkaStreamsApplicationHostInfo() 查询当前主机信息，该方法返回当前的 HostInfo。其思路是：应用可以先获取该键托管位置的信息，然后将 HostInfo 与关于当前实例的 HostInfo 比较。如果 HostInfo 数据匹配，就可以通过 retrieveQueryableStore 进行简单的 JVM 调用；否则走 RPC 选项。

Kafka Streams 示例

下面的示例结合了本章涵盖的各个主题：

@Configuration
@EnableKafka
@EnableKafkaStreams
public class KafkaStreamsConfig {

    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
    public KafkaStreamsConfiguration kStreamsConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "testStreams");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass().getName());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
        props.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, WallclockTimestampExtractor.class.getName());
        return new KafkaStreamsConfiguration(props);
    }

    @Bean
    public StreamsBuilderFactoryBeanConfigurer configurer() {
        return fb -> fb.setStateListener((newState, oldState) -> {
            System.out.println("State transition from " + oldState + " to " + newState);
        });
    }

    @Bean
    public KStream<Integer, String> kStream(StreamsBuilder kStreamBuilder) {
        KStream<Integer, String> stream = kStreamBuilder.stream("streamingTopic1");
        stream
                .mapValues((ValueMapper<String, String>) String::toUpperCase)
                .groupByKey()
                .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMillis(1_000)))
                .reduce((String value1, String value2) -> value1 + value2,
                		Named.as("windowStore"))
                .toStream()
                .map((windowedId, value) -> new KeyValue<>(windowedId.key(), value))
                .filter((i, s) -> s.length() > 40)
                .to("streamingTopic2");

        stream.print(Printed.toSysOut());

        return stream;
    }

}

基础知识​

Spring 管理​

KafkaStreams Micrometer 支持​

Streams JSON 序列化与反序列化​

使用 KafkaStreamBrancher​

配置​

清理与停止配置​

组协议配置​

Header Enricher​

MessagingProcessor​

恢复策略​

从反序列化异常中恢复​

从处理异常中恢复​

从生产异常中恢复​

死信发布 Recoverer​

死信目的地解析器​

死信队列主题名属性​

交互式查询支持​

重试状态存储检索​

查询远程状态存储​

Kafka Streams 示例​

基础知识

Spring 管理

KafkaStreams Micrometer 支持

Streams JSON 序列化与反序列化

使用 `KafkaStreamBrancher`

配置

清理与停止配置

组协议配置

Header Enricher

`MessagingProcessor`

恢复策略

从反序列化异常中恢复

从处理异常中恢复

从生产异常中恢复

死信发布 Recoverer

死信目的地解析器

死信队列主题名属性

交互式查询支持

重试状态存储检索

查询远程状态存储

Kafka Streams 示例