java

2.1 Kafka Streams 拓扑

Kafka Streams Topology 核心概念：StreamsBuilder 创建 Source/Sink/Processor 节点、拓扑的可视化与验证、TopologyTestDriver 单元测试、以及 Kafka Streams 与 Kafka Consumer Group 的关系。

Leeting Yan 2026-11-01 3 分钟阅读 1196 字

Kafka Streams 是什么

Kafka Streams 是一个轻量级流处理库，用于构建 Kafka 之上的有状态数据处理应用。它不是独立的集群——你的应用本身就是流处理作业，通过 Kafka Consumer Group 协调。

关键特点：

无外部依赖：只需要 Kafka 集群（甚至可用 TopologyTestDriver 无 broker 测试）
有状态处理：内置本地状态存储（RocksDB/内存）
Exactly-Once 支持：通过 Kafka 事务实现 EOS
与 Kafka 深度集成：自动处理分区分配、Consumer Group 协调、offset 提交

Topology 基础

Kafka Streams 的核心抽象是 Topology（拓扑）：有向图，节点是处理单元，边是数据流。

Source（输入） → Processor（处理） → Sink（输出）
     ↑                                              ↓
   Kafka Topic                                Kafka Topic

用 StreamsBuilder 创建拓扑：

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;

public class SimpleTopology {
    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();

        // Source：从 Topic 读取
        KStream<String, String> source = builder.stream("input-topic");

        // Processor：转换
        KStream<String, String> uppercased = source.mapValues(value -> value.toUpperCase());

        // Sink：写入 Topic
        uppercased.to("output-topic");

        // 构建拓扑
        Topology topology = builder.build();

        // 启动应用
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "simple-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // stream()/to() 都没有显式指定 Serde，运行时会回落到默认 Serde；
        // 不配这两项，启动阶段就会抛 "Default key serde not specified"
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        KafkaStreams streams = new KafkaStreams(topology, props);
        streams.start();

        // 优雅关闭
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}

三种节点类型

类型	API	作用
Source	`builder.stream()` / `builder.table()`	从 Kafka Topic 读数据
Processor	`filter()` / `map()` / `join()` / `aggregate()`	转换、过滤、聚合数据
Sink	`.to()`	写回 Kafka Topic

拓扑可视化

Topology topology = builder.build();
System.out.println(topology.describe());

输出：

Topologies:
   Sub-topology: 0
    Source: KSTREAM-SOURCE-0000000000 (topics: [input-topic])
      --> KSTREAM-MAPVALUES-0000000001
    Processor: KSTREAM-MAPVALUES-0000000001 (stores: [])
      --> KSTREAM-SINK-0000000002
      <-- KSTREAM-SOURCE-0000000000
    Sink: KSTREAM-SINK-0000000002 (topic: output-topic)
      <-- KSTREAM-MAPVALUES-0000000001

每个算子自动编号，便于调试和监控对应。

TopologyTestDriver：无 broker 测试

Kafka Streams 最强大的测试工具是 TopologyTestDriver——不需要运行 Kafka broker：

import org.apache.kafka.streams.*;
import org.apache.kafka.streams.test.TestRecord;

@Test
void testUpperCaseTopology() {
    StreamsBuilder builder = new StreamsBuilder();
    builder.<String, String>stream("input-topic")
        .mapValues(String::toUpperCase)
        .to("output-topic");

    Topology topology = builder.build();

    Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "test-app");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:9092");
    // 拓扑里没有显式 Consumed/Produced，默认 Serde 必须配上
    props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
    props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

    try (TopologyTestDriver driver = new TopologyTestDriver(topology, props)) {
        TestInputTopic<String, String> input = driver.createInputTopic(
            "input-topic",
            new StringSerializer(),
            new StringSerializer()
        );

        TestOutputTopic<String, String> output = driver.createOutputTopic(
            "output-topic",
            new StringDeserializer(),
            new StringDeserializer()
        );

        input.pipeInput("key1", "hello");

        TestRecord<String, String> result = output.readRecord();
        assertEquals("key1", result.key());
        assertEquals("HELLO", result.value());
    }
}

TopologyTestDriver 在内存中模拟整个拓扑，输入/输出都在 JVM 内完成。测试速度快（毫秒级）、可重复、无需 Docker。

与 Consumer Group 的关系

每个 Kafka Streams 应用通过 APPLICATION_ID_CONFIG 加入 Consumer Group：

应用实例 1（application.id = "eventd-cleansing"）
    ├── 消费分区 0, 1
    └── 本地状态存储 State Store

应用实例 2（application.id = "eventd-cleansing"）
    ├── 消费分区 2, 3
    └── 本地状态存储 State Store

Kafka Coordinator 自动处理：
    - 分区再均衡（实例上下线）
    - Offset 提交（at_least_once 默认 30 秒；开了 EOS 后默认 100ms，见 3.2）
    - 消费进度追踪

扩容：启动更多实例，Kafka 自动重新分配分区。

配置速查

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "eventd-cleansing");  // Consumer Group ID
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 10000);  // 每 10 秒提交一次
// EOS 下提交间隔必须小于 producer 的 transaction.timeout.ms，
// 否则事务可能先超时、提交失败
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 4);  // 每个实例的线程数

小结

概念	说明
StreamsBuilder	创建拓扑的 DSL 构建器
Topology	有向图，描述 Source → Processor → Sink
KStream	无界数据流（每条记录独立）
KTable	changelog 流（相同 key 更新）
TopologyTestDriver	内存测试，无需 Kafka broker
APPLICATION_ID	Consumer Group ID，用于协调

下一节 2.2 KStream 与 KTable ：有界/无界、更新流、changelog Topic。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页