/* MD / 富文本中的 .toc（含博客园搬家等嵌套结构）；.toc-box 在侧栏，不受影响 */
#content_views .toc,
/* 编辑器常在目录前后插入空 p（:empty 仍占 20px），一并去掉避免顶空隙 */
#content_views.markdown_views > p:empty:has(+ .toc),
#content_views.markdown_views > .toc + p:empty,
/* 富文本旧版目录标记 */
#content_views.htmledit_views #main-toc,
#content_views.htmledit_views #hr-toc,
#content_views.htmledit_views p[id*=\”-toc\”] {
display: none !important;
}
/* 目录去掉后，紧跟的首个标题不再多出一块上边距 */
#content_views.markdown_views > .toc + h1,
#content_views.markdown_views > .toc + h2,
#content_views.markdown_views > .toc + h3,
#content_views.markdown_views > .toc + h4,
#content_views.markdown_views > .toc + p:empty + h1,
#content_views.markdown_views > .toc + p:empty + h2,
#content_views.markdown_views > .toc + p:empty + h3,
#content_views.markdown_views > .toc + p:empty + h4 {
margin-top: 0 !important;
}
做实时计算，大家第一反应往往是 Flink。但如果你的场景没那么重，不想维护庞大的 Flink 集群，Kafka Streams 绝对是个被低估的利器。它直接嵌在 Java 进程里，跟 Kafka 亲儿子一样，部署起来就是个普通的 Spring Boot 应用。
最近带团队搞了几个流处理项目，把 Kafka Streams 从基础用法到生产级踩坑都趟了一遍。今天就把状态存储、窗口计算、Exactly-Once 这些核心玩法，还有多租户和运维的实战经验盘一盘。
1. 核心概念与拓扑设计：别被名词唬住
Kafka Streams 的核心抽象就两个：KStream 和 KTable。
说白了，KStream 就是流水账，来一条处理一条；KTable 是账本，只记最新状态（类似数据库的表）。
流处理拓扑（Topology） 就是计算逻辑，本质上是个有向无环图。设计拓扑时，Source 进数据，Processor 搞计算，Sink 吐结果。平时写 DSL 链式调用（stream().filter().map().to()）挺爽，但逻辑一复杂，还是得老老实实切回 Processor API，不然调试起来能让人怀疑人生。
2. Spring Boot 集成：自动配置虽好，参数得自己捏
Spring Boot 把 Kafka Streams 封装得很省事，加个 @EnableKafkaStreams 就能跑。但自动配置归自动配置，有些生产环境的参数必须自己配置，不能全指望默认值。
@Configuration
@EnableKafkaStreams
public class KafkaStreamsConfig {
@Bean
public StreamsBuilderFactoryBeanCustomizer streamsCustomizer() {
return factory -> {
// 自定义配置，顺便把未捕获异常处理器也配了，防止线程默默死掉
factory.setStreamsConfiguration(customStreamsConfig());
factory.setUncaughtExceptionHandler((thread, exception) -> {
log.error("Stream thread {} 崩了", thread.getName(), exception);
// 这里可以触发告警，或者直接调 KafkaStreams.close() 让应用重启
});
};
}
@Bean
public KafkaStreamsConfiguration customStreamsConfig() {
Map<String, Object> props = new HashMap<>();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-stream-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
// 生产环境无脑上 EOS v2，老版本的 exactly_once 会让 Broker 连接数爆炸
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
// 状态目录别放在系统盘，挂载独立数据盘
props.put(StreamsConfig.STATE_DIR_CONFIG, "/data/kafka-streams");
// 注意：这个异常处理器是 spring-kafka 提供的，需要确保引入了对应依赖
props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
SendToDeadLetterTopicExceptionHandler.class);
return new KafkaStreamsConfiguration(props);
}
}
Spring Boot 会自动管理 StreamsBuilder 和 KafkaStreams 的生命周期。应用启动时初始化拓扑，关闭时优雅提交 Offset 并刷盘。
3. 状态存储 RocksDB：快是真快，爆也是真爆
Kafka Streams 敢做有状态计算，底气就在 RocksDB。数据存本地磁盘，读写极快。但稍微不注意，就能把机器内存和磁盘撑爆。
状态恢复与 Standby Replicas
状态存储不是孤立的，Kafka Streams 会在后台为每个状态存储建一个内部的 Changelog Topic。数据改了，同步写 Changelog。节点宕机重启，从 Changelog 拉数据重建本地状态。
这里有个实战经验：一定要配 Standby Replicas（备用副本）。
props.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);
不然节点一挂，新节点拉起时从 Changelog 狂拉数据，那恢复时间能让你等到怀疑人生。配了备用副本，其他节点会异步维护一份只读副本，主节点挂了直接热切换，恢复时间从分钟级降到毫秒级。
4. 窗口聚合：关掉 Grace Period 保平安
流数据无限长，必须切块算。Kafka 给了三种窗口：滚动、滑动、会话。
// 1. 滚动窗口：固定大小不重叠，比如每小时统计一次
KTable<Windowed<String>, Long> hourlyClicks = clickStream
.groupByKey()
.windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofHours(1)))
.count(Materialized.as("hourly-clicks-store"));
// 2. 滑动窗口：固定大小但重叠，算移动平均
KTable<Windowed<String>, Long> movingAvg = eventStream
.groupByKey()
.windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)).advanceBy(Duration.ofMinutes(1)))
.count();
// 3. 会话窗口：动态大小，基于活动间隔，算用户在线时长
KTable<Windowed<String>, Long> sessionDuration = userActivityStream
.groupByKey()
.windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(30)))
.count();
重点提一句 WithNoGrace。老版本默认有 24 小时的 Grace Period，用来处理迟到数据。但在很多业务里，这 24 小时的宽限期会导致状态存储里堆积大量过期数据，直接把 RocksDB 撑爆。只要业务能容忍极少量的迟到数据丢失，果断加上 WithNoGrace 关掉它。
5. 表流 Join：注意单向触发和序列化大坑
把 KStream 和 KTable 做 Join 是常态，比如订单流关联用户信息表。
// 1. 构建用户信息 KTable
KTable<String, UserDetail> userTable = builder.table(
"user-details-topic",
Consumed.with(Serdes.String(), userDetailSerde)
);
// 2. 订单流 Join 用户表
KStream<String, EnrichedOrder> enrichedOrders = orderStream
.leftJoin(
userTable,
(order, user) -> new EnrichedOrder(order, user),
// 坑点：这里的 Serde 别瞎写 null，左右两边的 Value Serde 都得老老实实传进去
Joined.with(Serdes.String(), orderSerde, userDetailSerde)
);
两个大坑：
6. Exactly-Once 语义：事务打包，拒绝重复
EOS (Exactly-Once) 听起来很玄乎，其实底层就是靠事务。
以前用 exactly_once，每个 Task 一个事务生产者，Broker 连接数直接爆炸。现在无脑上 exactly_once_v2，所有 Task 共享一个事务生产者，资源消耗断崖式下降。
底层逻辑很简单：读数据、改状态、写结果、提交 Offset，这四步打包成一个 Kafka 事务。要么全成功，要么全回滚。宕机重启？没关系，从上次提交的 Offset 接着跑，状态靠 Changelog 恢复，数据绝对不会重复处理。
7. 交互式查询 (IQ)：缓存带来的一致性陷阱
IQ 是个好东西，能把 Streams 应用变成分布式 KV 数据库，直接通过 REST 接口查状态。
@RestController
@RequestMapping("/api/state")
public class StateQueryController {
// 注意：注入的是 KafkaStreams，不是 StreamsBuilder
private final KafkaStreams kafkaStreams;
public StateQueryController(KafkaStreams kafkaStreams) {
this.kafkaStreams = kafkaStreams;
}
@GetMapping("/user/{userId}/score")
public ResponseEntity<Long> getUserScore(@PathVariable String userId) {
ReadOnlyKeyValueStore<String, Long> store = kafkaStreams.store(
StoreQueryParameters.fromNameAndType("user-score-store", QueryableStoreTypes.keyValueStore())
);
Long score = store.get(userId);
return score != null ? ResponseEntity.ok(score) : ResponseEntity.notFound().build();
}
}
巨坑预警：Kafka Streams 默认开了内存缓存！你刚写进去的数据，可能还在缓存里没刷到 RocksDB，这时候去查，根本查不到。
怎么解？
8. 拓扑优化：DTO 不可变与自定义序列化
DTO 尽量用 Lombok 的 @Value 搞成不可变对象，流处理里最怕状态被意外篡改。
@Value
@Builder
public class OrderAggregate {
String orderId;
BigDecimal totalAmount;
int itemCount;
}
DSL 搞不定的复杂逻辑（比如定时任务、多状态存储联动），别硬憋，直接上 Processor API。
序列化方面，JSON 开发快，但性能和体积不如 Protobuf。生产环境数据量大的话，老老实实切 Protobuf，能省下不少带宽和磁盘。
9. 多租户隔离：防住“吵闹的邻居”
SaaS 场景下搞多租户，最简单的就是 Topic 加前缀（如 tenantA.orders）。但如果某个租户数据量特别大，把资源吃光了，其他租户就得跟着遭殃。
这时候就得物理隔离，给大租户单独分配 Application ID：
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "stream-app-" + tenantId);
这样每个租户拥有独立的 Consumer Group 和状态存储。资源配额方面，Streams 自己不管这事，得靠 K8s 的 Limit 和 Request 来卡脖子，限制每个 Pod 的 CPU 和内存。
10. 生产运维：重置、监控与异常兜底
应用重置
跑飞了或者要重跑历史数据，用重置脚本。记住，跑这脚本前必须先把应用停了！
kafka-streams-application-reset.sh –application-id order-stream-app \\
–bootstrap-server localhost:9092 –input-topics orders –to-earliest
监控指标
Streams 自带的 JMX 指标很全，接个 Micrometer 打到 Prometheus 就行。重点盯 task-closed-rate，如果这指标忽高忽低，说明 Rebalance 风暴来了，赶紧去查是不是有节点在频繁挂掉或者处理太慢。
异常兜底
生产环境脏数据是常态，必须做兜底：
// 1. 反序列化遇到脏数据，扔到死信队列，别让应用直接崩了
props.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG,
SendToDeadLetterTopicExceptionHandler.class);
// 2. 生产异常（如 Broker 拒绝写入），配置继续运行（Kafka 2.8+ 支持）
props.put(StreamsConfig.DEFAULT_PRODUCTION_EXCEPTION_HANDLER_CLASS_CONFIG,
ContinueOnProductionExceptionHandler.class);
写在最后
Kafka Streams 在 Java 生态里是个很实在的流处理框架，没有 Flink 那么重，但能解决 80% 的实时计算需求。
当然，它也有自己的脾气，比如 Rebalance 慢、状态存储调优麻烦、IQ 查询有缓存延迟。用的时候得多留个心眼，别把它当成无所不能的银弹。
今天就先聊到这，大家有遇到奇葩问题的，或者对 Flink 和 Streams 选型有纠结的，欢迎留言探讨。
🎁 福利时间
如果你正在备战面试或者想要学习其他知识，给大家推荐一个宝藏知识库，作者整理了一些列 Java 程序员需要掌握的核心知识，有需要的自取不谢。
知识库地址：https://farerboy.com/
