3. 核心概念：Topology、Stream、Processor、State Store

说实话，刚接触 Kafka Streams 的时候，我被一堆术语搞得有点懵。什么拓扑、流、处理器、状态存储……听起来像是网络课和数据库课混在一起了。但等你真正上手写一个流处理任务，你会发现这些概念其实就对应着你代码里的几个核心组件。

这一章，我就带你把这四个概念彻底捋清楚。搞懂了它们，Kafka Streams 的骨架你就抓住了。

3.1 Topology（拓扑）—— 你的数据处理蓝图

Topology，直译过来就是“拓扑结构”。在 Kafka Streams 里，它就是你整个数据处理逻辑的有向无环图（DAG）。说白了，就是数据从哪进、经过哪些步骤、最后出到哪去的一张蓝图。

我个人习惯把 Topology 想象成一条流水线。原始数据从 Kafka Topic 流进来，经过一个个处理节点，最终输出到新的 Topic 或者状态存储里。每个节点干一件事，节点之间用边连接，数据就沿着边流动。

下面这张图展示了一个典型的 Topology 结构：

你看，数据从 orders Topic 进来，经过“过滤”和“转换”两个处理器，最终写入 valid_orders Topic。同时，过滤处理器还会把一些中间结果写到 State Store 里。

3.2 Stream（流）—— 永不停止的数据河流

Stream 是 Kafka Streams 的核心抽象。它代表一个无界、持续不断的数据序列。每条记录都是一个键值对（Key-Value），并且带有时间戳。

你想想看，传统数据库处理的是“静止的数据”，你查一次就完了。但流处理面对的是“流动的数据”，数据源源不断地来，你得实时处理。这就是 Stream 和普通集合最大的区别。

在代码里，Stream 通常表现为 KStream 或 KTable：

• KStream：记录的全量日志，每条记录都是独立的。你可以对它做 map、filter、flatMap 等操作。
• KTable：记录的变更日志，它维护的是某个 Key 的最新状态。你可以把它理解成一个“持续更新的 Map”。

举个例子：

// 创建一个 KStream，从 "orders" Topic 消费数据
KStream<String, Order> orderStream = builder.stream("orders");

// 创建一个 KTable，从 "customers" Topic 消费数据，维护客户最新信息
KTable<String, Customer> customerTable = builder.table("customers");

这里 orderStream 会收到每一条订单记录，而 customerTable 只保留每个客户的最新状态。两者处理逻辑完全不同。

3.3 Processor（处理器）—— 真正的“干活”单元

Processor 是 Topology 里的节点，负责执行具体的数据处理逻辑。Kafka Streams 提供了两种处理器：

类型	说明	典型操作
Processor API	底层 API，需要手动实现 process() 方法	自定义状态管理、精确控制处理逻辑
DSL API	高层 API，通过函数式组合完成处理	map()、filter()、groupBy()、join()

我个人建议：80% 的场景用 DSL API，它简洁、易读、不容易出错。只有当你需要精细控制状态存储、或者实现复杂的窗口逻辑时，才考虑 Processor API。

来看一个 DSL 的例子：

// DSL 风格：过滤出金额大于 100 的订单
KStream<String, Order> largeOrders = orderStream
    .filter((key, order) -> order.getAmount() > 100);

如果用 Processor API，同样的逻辑会复杂一些：

// Processor API 风格：需要手动实现 Processor 接口
class AmountFilterProcessor implements Processor<String, Order> {
    private ProcessorContext context;

    @Override
    public void init(ProcessorContext context) {
        this.context = context;
    }

    @Override
    public void process(String key, Order value) {
        if (value.getAmount() > 100) {
            context.forward(key, value);
        }
    }

    @Override
    public void close() {
        // 清理资源
    }
}

你看，DSL 一行搞定的事，Processor API 要写一个类。所以除非必要，别给自己找麻烦。

3.4 State Store（状态存储）—— 让流处理拥有“记忆”

流处理最大的挑战是什么？是数据是无界的。你没法像批处理那样，把所有数据都放在内存里算。但很多场景下，你又需要记住一些东西——比如过去 5 分钟的订单总额、某个用户的最近一次操作。

这就是 State Store 的用武之地。它是 Kafka Streams 内置的、可容错的状态存储机制。你可以把它理解成一个嵌入在流处理应用里的本地数据库。

State Store 有两种主要类型：

• KeyValueStore：最常用的存储，按 Key 存取 Value。适合做计数、累加、去重。
• WindowStore：带时间窗口的存储，数据按时间分段。适合做滑动窗口统计。

来看一个实际例子：统计每个用户过去 1 小时的订单总额。

// 使用 DSL 的窗口聚合，底层自动使用 State Store
KStream<String, Order> orderStream = builder.stream("orders");

KTable<Windowed<String>, Double> hourlyTotal = orderStream
    .groupByKey()
    .windowedBy(TimeWindows.of(Duration.ofHours(1)))
    .aggregate(
        () -> 0.0,
        (key, order, total) -> total + order.getAmount(),
        Materialized.<String, Double, WindowStore<Bytes, byte[]>>as("hourly-total-store")
    );

这里 Materialized.as("hourly-total-store") 就是告诉 Kafka Streams：帮我创建一个名为 hourly-total-store 的 State Store，用来存每个用户每小时的订单总额。

3.5 四个概念如何协同工作？

现在我们把四个概念串起来，看看它们在实际应用中是怎么配合的：

1. Topology 定义了整个处理流程的蓝图。
2. Stream 是数据流动的载体，数据从 Source Topic 流入 Topology。
3. Processor 是 Topology 里的节点，对 Stream 中的数据进行处理。
4. State Store 为 Processor 提供持久化的“记忆”能力，让流处理可以记住历史信息。

举个例子：假设你要实现一个“实时订单告警”系统——如果某个用户 5 分钟内下了超过 10 单，就触发告警。

• Topology：定义整个流程：读取订单 → 按用户分组 → 开 5 分钟窗口 → 计数 → 判断是否超过阈值 → 输出告警。
• Stream：订单数据以 KStream 的形式流入。
• Processor：groupByKey()、windowedBy()、aggregate() 这些 DSL 操作背后都是 Processor。
• State Store：aggregate() 自动使用 WindowStore 来存储每个用户在每个窗口内的订单计数。

你看，这四个概念缺一不可。没有 Topology，你没法组织处理逻辑；没有 Stream，数据没法流动；没有 Processor，没人干活；没有 State Store，你记不住历史数据，窗口聚合根本做不了。