《Java 消息中间件·事件驱动架构》7.1 Streams 的拓扑与处理器

Streams 不是另一个 MapReduce

Kafka Streams 的 API 看起来像普通 Java Stream，但状态和位移管理全是隐式的。Streams 的本质是一个常驻消费者组：以 Kafka topic 为输入源、中间 topic 为状态交换、输出 topic 为结果投递。所有数据处理在 JVM 进程内完成，不需要额外集群管理器。运维 Streams 应用的心智模型和 Kafka consumer group 完全一致：分区再平衡、消费者 lag、commit offset 全部保留。

拓扑的三类节点

• Source：从 Kafka topic 读取数据，作为拓扑入口。每个 source 对应一个消费者。
• Processor：对数据转换、过滤、聚合，可有状态或纯无状态。
• Sink：把处理结果写回到 Kafka topic，每个 sink 对应一个生产者。

三类节点通过有向边连接形成 DAG。一个拓扑可以有多个 source topic 和多个 sink topic，子拓扑之间通过 topic 隐式连接。

最小拓扑代码与编译运行

构造过滤拓扑，输入 libraryd.loan.events，只保留包含 OVERDUE 的事件：

import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;

public class TopologyProbe {
    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> loans = builder.stream("libraryd.loan.events");
        loans.filter((k, v) -> v != null && v.contains("OVERDUE"))
             .to("libraryd.overdue.events");
        System.out.println(builder.build().describe());
    }
}

编译运行（单次运行，随调度波动，不作基准）：

export CP="kafka-streams-4.3.0.jar:kafka-clients-4.3.0.jar:slf4j-api-1.7.36.jar:slf4j-simple-1.7.36.jar"
javac --release 21 -cp "$CP" TopologyProbe.java && \
java -cp ".:$CP" -Dorg.slf4j.simpleLogger.defaultLogLevel=warn TopologyProbe

kafka-streams 默认打印大量 INFO 日志，用 -Dorg.slf4j.simpleLogger.defaultLogLevel=warn 降噪。

Topology.describe() 输出解析（原样）

Topologies:
   Sub-topology: 0
    Source: KSTREAM-SOURCE-0000000000 (topics: [libraryd.loan.events])
      --> KSTREAM-FILTER-0000000001
    Processor: KSTREAM-FILTER-0000000001 (stores: [])
      --> KSTREAM-SINK-0000000002
      <-- KSTREAM-SOURCE-0000000000
    Sink: KSTREAM-SINK-0000000002 (topic: libraryd.overdue.events)
      <-- KSTREAM-FILTER-0000000001

逐行解释：

• Sub-topology: 0：当前只有一个子拓扑，存在分支时会产生多个。
• Source: KSTREAM-SOURCE-0000000000：源节点消费 libraryd.loan.events。
• Processor: KSTREAM-FILTER-0000000001：过滤处理器，stores: [] 表示无状态。
• Sink: KSTREAM-SINK-0000000002：输出到 libraryd.overdue.events。
• –> 数据流出，<– 数据流入。

节点 ID 从 0000000000 递增，每次调用 filter()/map()/to() 分配新 ID。同一拓扑中 ID 是确定性的。

有状态处理器与状态存储

KTable<String, Integer> counts = loans
    .groupByKey()
    .aggregate(() -> 0, (k, v, agg) -> agg + 1,
               Materialized.as("loan-counts-store"));

对应拓扑输出：

Processor: KSTREAM-AGGREGATE-0000000004 (stores: [loan-counts-store])

stores: [loan-counts-store] 表示处理器操作本地 RocksDB 状态存储。Streams 在后台创建 changelog topic：{application.id}-loan-counts-store-changelog，用于故障恢复和状态迁移。

changelog topic 的自动创建

• replication factor：默认 1，生产环境应设为 3。
• partition 数：等于上游 source topic 的分区数。
• cleanup policy：默认 compact，只保留 key 的最新值。

自动创建需要 auto.create.topics.enable=true，否则启动报 UnknownTopicOrPartitionException。

拓扑的独立性与可测试性

Topology 可独立于 Kafka 集群创建验证：

1. 单元测试：用 TopologyTestDriver 输入测试数据检查输出，无需启动 broker。
2. 静态分析：部署前检查死胡同、source 冲突等拓扑问题。

本卷只演示 describe() 静态验证。

拓扑与消费者组

每个 Streams 应用实例属于一个消费者组，application.id 就是组名。两个实例共享相同 group 时，分区会被平均分配。

拓扑中每个 source topic 创建独立消费者。消费 3 个 topic 时有 3 个消费者线程，并发度受限于 topic 分区数。

拓扑调试常见问题

拓扑版本化

• 新增无状态 processor：直接滚动部署。
• 修改状态存储结构：需清空 state.dir 或迁移状态。
• 删除 source topic：同步修改订阅，否则报异常。
• 变更 application.id：等于创建新消费者组，旧状态丢失。

建议在 CI 中把 describe() 输出作为快照捕获。

分支与合并

KStream<String, String>[] branches = loans.branch(
    (k, v) -> v.contains("CREATED"),
    (k, v) -> v.contains("RETURNED"),
    (k, v) -> true);
branches[0].to("created"); branches[1].to("returned"); branches[2].to("other");

split() 是广播语义：每条消息进入所有分支判断，但只路由到一个 sink。多团队共享同一 topic 时非常有用。

Processor API 与外部系统

Processor API 可直接连接外部 HTTP/Redis/数据库，但增加了延迟和复杂性，重试可能导致副作用重复。除非必要，尽量通过 sink topic + Kafka Connect 对接外部系统。

监控指标

部署注意事项

1. state.dir 默认在 /tmp/kafka-streams，生产环境改为持久化目录。
2. RocksDB 占用堆外内存，实际内存使用超过 -Xmx。
3. 使用 KafkaStreams.close() 而非 kill -9，确保状态 flush。
4. 滚动更新时先停止旧实例再启动新实例。

拓扑优化的常见策略

当拓扑处理延迟过高时：

1. 减少分区再平衡：增大 session.timeout.ms 和 heartbeat.interval.ms。
2. 增加消费者实例：提升并发度，但不超过 topic 的分区数。
3. 减少 poll 间隔：降低 max.poll.records，增加消费频率。
4. 优化状态存储：增大 RocksDB 的 block cache，减少磁盘 I/O。
5. 使用无状态处理器：聚合逻辑允许时，用纯 KStream 而非 KTable。

拓扑与 Kafka Connect 的互补

Kafka Streams 和 Kafka Connect 是两个不同的抽象层：

• Streams：处理实时流数据，关注低延迟、有状态计算。
• Connect：集成外部系统，关注可靠性、数据格式转换。

两者配合：Connect 从数据库读取 CDC 事件进入 topic，Streams 消费 topic 实时计算，再通过 Connect 写回数据库。数据集成和流计算解耦，各自负责不同问题域。

拓扑设计与微服务边界

一个 Streams 应用应该多大？答案是：一个 bounded context 对应一个 Streams 应用。

例如 libraryd 的借阅业务：

• libraryd.loan.events：借阅事件的 source topic
• libraryd.overdue.events：逾期事件的 sink topic
• libraryd.loan.aggregates：借阅聚合的状态存储 changelog

财务和库存系统各自维护 Streams 应用，通过共享 topic 通信。每个团队独立部署和演进拓扑。

拓扑调试的诊断命令

# 查看拓扑描述
java -cp "..." TopologyProbe | head -30

# 检查消费者组状态
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group libraryd-streams-app

# 查看 changelog topic 内容
kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic libraryd-streams-app-loan-counts-store-changelog \
  --from-beginning --property print.key=true

本节小结

• Streams 拓扑由 Source、Processor、Sink 三类节点组成，形成 DAG。
• Topology.describe() 是静态验证拓扑的关键工具，应纳入 CI。
• 有状态处理器使用本地 RocksDB，后台创建 changelog topic 用于恢复。
• 拓扑可独立验证和测试，无需启动 Kafka broker。
• 消费并发度受限于分区数，部署时需注意状态目录持久化和 graceful 关闭。
• 分支与合并实现多团队共享 topic，Processor API 对接外部系统需谨慎。
• 监控指标涵盖处理延迟、消费速率和状态存储延迟三个维度。
• 拓扑设计应按 bounded context 划分，每个 context 对应一个 Streams 应用。

常见拓扑反模式

以下是 Kafka Streams 拓扑设计中的常见反模式：

1. 在一个拓扑中消费过多 topic：一个拓扑消费 10+ 个 topic 会导致消费者线程爆炸，debug 困难。建议按 bounded context 拆分成多个小拓扑。
2. 有状态处理器不做 changelog 备份：如果 changelog topic 被误删，状态存储无法恢复。生产环境应单独备份 changelog topic。
3. state.dir 放在 /tmp：/tmp 目录在 Linux 重启后会被清理，导致状态丢失。生产环境应使用 /var/lib 等持久化目录。
4. 忽视 Topology.describe() 的变化：拓扑代码修改后 describe() 输出改变，如果不检查就直接部署，可能导致消费者组异常。
5. 使用 kill -9 停止 Streams 应用：kill -9 不会触发 graceful shutdown，状态存储可能损坏，offset 可能未提交。应使用 kill -15 或 KafkaStreams.close()。
6. graceful shutdown 超时设置过短：如果 close() 的超时时间（默认 30 秒）不足以 flush 所有状态，可能导致数据丢失。
7. changelog topic 的 replication factor 为 1：单副本 changelog topic 一旦 broker 故障，状态无法恢复。生产环境应设为 3。
8. 在 Processor API 中做阻塞 I/O：Processor API 中的阻塞调用（如 HTTP 请求）会阻塞整个消费者线程，导致 lag 飙升和 rebalancing 频繁。

识别这些反模式需要在 CI 中增加拓扑静态分析：捕获 describe() 输出变化、检查 state.dir 配置、验证 changelog topic 的 replication 等。

拓扑演进策略

当业务需求变化时，拓扑需要演进。演进策略取决于变更的类型：

对于状态存储结构的变更，推荐的迁移流程：

1. 在新版本中创建新的 state store（如 loan-counts-v2）。
2. 双写：旧 topology 继续写 loan-counts，新 topology 同时写 loan-counts-v2。
3. 验证 loan-counts-v2 的数据正确性。
4. 切换流量到新 topology。
5. 删除旧 state store 和 changelog topic。

这种双写策略保证了迁移过程中的数据一致性，避免了「切换到新 topology 后发现数据错误」的风险。

阅读导航：上一节 6.3 消费者驱动的契约 ｜ 下一节 7.2 状态存储