为什么需要状态存储
无状态转换（filter、map）不依赖历史数据。但很多场景需要有状态处理：
- 去重：记录已见过的事件 ID
- 计数：统计每用户的订单数
- 聚合：计算每商品的累计销售额
- Join：保存一侧流的最新值等待另一侧
Kafka Streams 内置状态存储，无需外部数据库。
KeyValueStore
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.StoreBuilder;
import org.apache.kafka.streams.state.Stores;
// 创建状态存储
StoreBuilder<KeyValueStore<String, Long>> countStoreBuilder =
Stores.keyValueStoreBuilder(
Stores.persistentKeyValueStore("order-counts"), // 名称
Serdes.String(), // key 序列化
Serdes.Long() // value 序列化
);
// 添加到拓扑
builder.addStateStore(countStoreBuilder);
| 存储类型 | API | 数据位置 | 适用 |
|---|---|---|---|
| 持久化 | Stores.persistentKeyValueStore() | 本地 RocksDB | 大状态、需要恢复 |
| 内存 | Stores.inMemoryKeyValueStore() | JVM 堆 | 小状态、低延迟 |
| LRU 缓存 | Stores.lruMap(name, maxEntries) | JVM 堆，按条数淘汰 | 中等状态、允许丢旧数据 |
使用状态存储的 Processor
public class CountProcessor implements Processor<String, Order, String, Long> {
private KeyValueStore<String, Long> countStore;
// context 只在 init 里传进来一次，process 里要用就必须自己保存
private ProcessorContext<String, Long> context;
@Override
public void init(ProcessorContext<String, Long> context) {
this.context = context;
this.countStore = context.getStateStore("order-counts");
}
@Override
public void process(Record<String, Order> record) {
String userId = record.key();
Long current = countStore.get(userId);
if (current == null) current = 0L;
countStore.put(userId, current + 1);
// 转发结果
context.forward(record.withValue(current + 1));
}
}
更简洁的 DSL 方式：
KTable<String, Long> counts = orders
.groupByKey()
.count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>
as("order-counts")
.withKeySerde(Serdes.String())
.withValueSerde(Serdes.Long()));
Changelog Topic 与恢复
状态存储的变更自动写入 changelog Topic：
应用写入状态: put("alice", 10)
↓
Kafka Streams 自动发送: ("alice", 10) → <app-id>-<store-name>-changelog
↓
其他实例或恢复时: 从 changelog Topic 重建状态存储
changelog Topic 的名称格式是 <application-id>-<store-name>-changelog，没有 __ 前缀（那是 Kafka 内部 Topic 的惯例，Kafka Streams 的 changelog 不这样命名）。
KV store 的 changelog 默认是 compacted Topic，保留策略由 broker 的 Topic 配置决定，不会因为应用重启而丢历史。窗口存储再多保留一段时间：
// 只作用于窗口存储的 changelog：窗口过期后仍保留一段，便于迟到数据恢复
props.put(StreamsConfig.WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG, 24 * 60 * 60 * 1000);
交互查询（Interactive Queries）
状态存储支持外部直接查询：
// 启动时开启交互查询
props.put(StreamsConfig.APPLICATION_SERVER_CONFIG, "localhost:7070");
KafkaStreams streams = new KafkaStreams(topology, props);
streams.start();
// 查询状态
ReadOnlyKeyValueStore<String, Long> store = streams
.store(StoreQueryParameters.fromNameAndType(
"order-counts", QueryableStoreTypes.keyValueStore()));
Long aliceCount = store.get("alice");
System.out.println("Alice order count: " + aliceCount);
交互查询的限制：
- 只能查询本地状态存储中的 key（如果 key 在另一个实例上，需要自己按分区路由）
- 使用
RPC或REST API暴露查询接口 - 读到的是本地实例已处理到的进度，不是输入 Topic 的最新位点；跨实例做全局视图必须自己做扇出和合并
eventd v3 不直接使用交互查询，而是通过 Kafka Topic 输出结果到 Redis，由面板读取。
状态大小监控
状态存储过大时：
- RocksDB SST 文件增多 → 读写延迟增加
- 内存压力 → GC 频繁
- Checkpoint 变慢 → 恢复时间增加
监控指标：
// 状态存储指标挂在 stream-state-metrics 这个 group 下
for (Map.Entry<MetricName, ? extends Metric> entry : streams.metrics().entrySet()) {
if ("stream-state-metrics".equals(entry.getKey().group())) {
System.out.println(entry.getKey() + ": " + entry.getValue().metricValue());
}
}
// RocksDB 后端另外暴露 rocksdb.estimate-num-keys、
// rocksdb.estimate-live-data-size 等指标，可以直接看状态规模
状态清理
Window Store 和 Session Store 支持自动过期：
// 窗口状态保留 7 天
Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("window-store")
.withRetention(Duration.ofDays(7));
// 使用 StateTtlConfig（Flink，Kafka Streams 用 retention）
Kafka Streams 的 KeyValueStore 没有内置 TTL，需要手动清理或通过 compacted Topic 的保留策略间接控制。
小结
| 概念 | 说明 |
|---|---|
| KeyValueStore | key-value 状态存储 |
| persistentKeyValueStore | RocksDB 后端，持久化 |
| inMemoryKeyValueStore | 内存后端，低延迟 |
| changelog Topic | 状态变更日志，用于恢复 |
| Interactive Query | 直接查询本地状态存储 |
| retention | 窗口/会话状态的过期时间 |
下一节 3.2 Kafka Streams Exactly-Once 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。