《Java 消息中间件·事件驱动架构》7.3 窗口与聚合

为什么需要窗口

事件流是无限的，但聚合通常需要在有限的时间范围内进行。例如「统计过去 5 分钟的借阅量」就需要一个时间窗口来界定聚合的范围。没有窗口的聚合是全局聚合（如 count()），适用于「从头到尾累计」的场景，但不适合「最近 N 分钟统计」的需求。

Kafka Streams 支持三种窗口类型，分别适用于不同的时间语义场景。

滚动窗口（Tumbling Window）

滚动窗口是固定大小、不重叠、无间隔的窗口。如每 5 分钟一个窗口，00:00-00:05、00:05-00:10、00:10-00:15 等。

import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;
import java.time.Duration;

public class TumblingWindowDemo {
    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();
        builder.stream("libraryd.loan.events")
               .groupByKey()
               .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
               .count()
               .toStream()
               .to("libraryd.loan.counts.5min");
        System.out.println(builder.build().describe());
    }
}

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

javac --release 21 -cp "$CP" TumblingWindowDemo.java && \
java -cp ".:$CP" -Dorg.slf4j.simpleLogger.defaultLogLevel=warn TumblingWindowDemo

TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)) 创建一个 5 分钟滚动窗口，NoGrace 表示不允许迟到数据（迟到数据会被丢弃）。如果需要允许迟到，使用 ofSizeAndGrace()。

跳跃窗口（Hopping Window）

跳跃窗口是固定大小、可重叠、有跳跃间隔的窗口。如大小 5 分钟、跳跃 1 分钟，窗口为 00:00-00:05、00:01-00:06、00:02-00:07 等。

builder.stream("libraryd.loan.events")
       .groupByKey()
       .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5))
                                .advanceBy(Duration.ofMinutes(1)))
       .count();

跳跃窗口适合需要平滑统计的场景，如「最近 5 分钟的平均 QPS」。由于窗口重叠，同一条消息可能出现在多个窗口中，计算结果会重复计数（这是设计意图，不是 bug）。

会话窗口（Session Window）

会话窗口没有固定大小，而是由活动间隙决定。如用户 30 分钟无活动视为会话结束。

builder.stream("user.clicks")
       .groupByKey()
       .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(30)))
       .count();

这个代码表示：如果用户 30 分钟没有点击，当前会话结束，开始新会话。会话窗口会自动合并重叠的会话（如用户在两台设备上同时活动）。合并逻辑是：如果两个会话的时间范围重叠，合并为一个更大的会话。

三种窗口的对比

窗口与 topic retention 的关系

窗口聚合的结果写入输出 topic，但输入 topic 的 retention 设置决定了历史事件的可重放性。

如果输入 topic 的 retention.ms=86400000（1 天），那么超过 1 天的事件会被清理，无法重放。如果窗口大小是 7 天，但 retention 只有 1 天，那么窗口内的部分事件已经丢失，聚合结果不准确。

规则：窗口大小必须小于输入 topic 的 retention 时间，否则窗口聚合会基于不完整的数据。在生产环境中，通常设置 retention.ms = 窗口大小 × 3，确保有足够的历史数据用于重放和恢复。

迟到数据的处理

事件时间（event time）和处理时间（processing time）的差异导致迟到数据：

• Grace period：允许迟到的时间窗口。ofSizeWithNoGrace 表示不允许迟到，ofSizeAndGrace 可以配置迟到容忍度。
• 迟到数据的行为：在 grace period 内到达的数据会更新已有窗口的结果；超过 grace period 的数据被丢弃或发送到死信队列。

// 允许 1 分钟迟到
.windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(1)))

Grace period 的设置需要在业务准确性和系统资源之间权衡。Grace period 越长，窗口结果越准确，但状态存储越大（因为窗口必须保持开放更长时间），处理延迟也越高。

窗口存储的本地大小

窗口聚合使用 WindowStore 而非 KeyValueStore。WindowStore 的键包含时间戳信息：

key: (loanId, windowStartTime)
value: count

这意味着窗口数量增加时，状态存储的大小线性增长。如果窗口大小是 5 分钟，一天有 288 个窗口，每个 key 会有 288 条记录。需要配置 retention.ms 来清理过期窗口。

WindowStore 的存储大小估算：

总大小 = key 数量 × 窗口数 × (key 大小 + value 大小 + 时间戳大小)

例如 10 万个 key、每天 288 个窗口、每条记录 50 字节：

100000 × 288 × 50 = 1.44 GB/天

这个估算说明，窗口聚合的状态存储可能非常大，需要提前规划磁盘容量。

窗口的 join 操作

两个窗口化的流可以进行 join：

KStream<String, String> loans = builder.stream("loans");
KStream<String, String> returns = builder.stream("returns");

loans.join(returns,
    (loan, ret) -> loan + "->" + ret,
    JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(10)));

这个 join 的含义是：如果一笔借阅和一笔归还的时间差在 10 分钟内，认为它们匹配。Join 窗口的大小取决于业务语义：对于图书馆系统，可能是几小时（用户当天借当天还）；对于电商系统，可能是几天（用户下单后到收货确认）。

Join 窗口和聚合窗口是两个不同的概念：join 窗口控制两个事件的时间差，聚合窗口控制统计的时间范围。

窗口聚合的性能影响

窗口聚合对性能的影响主要来自状态存储的维护：

1. 状态存储大小：窗口数越多，状态存储越大，RocksDB 的读取和写入延迟增加。
2. changelog topic 写入：每次窗口更新都要写入 changelog topic，增加了 I/O 开销。
3. 内存占用：RocksDB 使用堆外内存，窗口数过多会导致内存压力。

优化建议：

• 减少窗口大小：在满足业务需求的前提下，使用尽可能小的窗口。
• 增加 retention 清理：缩短 window.retention.ms，及时清理过期窗口。
• 使用无窗口聚合：如果业务允许，使用全局聚合（如 KTable）而非窗口聚合。

本节小结

• 三种窗口类型：滚动窗口（固定不重叠）、跳跃窗口（固定可重叠）、会话窗口（动态）。
• 窗口大小必须小于输入 topic 的 retention 时间，否则聚合基于不完整数据。
• Grace period 控制迟到数据的处理，超过 grace period 的数据被丢弃。
• 会话窗口由活动间隙决定，自动合并重叠会话，适合用户行为分析。
• 窗口存储使用 WindowStore，键包含时间戳，状态大小随窗口数线性增长。
• Join 窗口控制两个事件的时间差，与聚合窗口是不同概念。
• 窗口聚合的性能影响主要来自状态存储维护，需合理规划窗口大小和 retention。

窗口聚合的输出格式

窗口聚合的输出键包含窗口信息：

KStream<Windowed<String>, Long> counts = loans
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
    .count()
    .toStream();

输出键的类型是 Windowed<String>，包含原始 key 和窗口时间范围：

Key: [loan-001@1700000000000/1700000300000], Value: 42

这个格式表示 loan-001 在窗口 [1700000000000, 1700000300000) 内的计数为 42。下游消费者需要解析 Windowed 键来提取窗口信息。

如果下游只需要原始 key，可以使用 map() 转换：

counts.map((windowedKey, value) ->
    KeyValue.pair(windowedKey.key(), value));

窗口聚合的 Watermark 机制

Watermark 是判断「窗口是否关闭」的时间标记。当 watermark 超过窗口的结束时间时，窗口被关闭，结果输出。

Watermark 的推进方式：

1. 基于处理时间：Watermark 等于当前系统时间，最简单但不准确。
2. 基于事件时间：Watermark 等于已消费消息的最大事件时间减去延迟容差。
3. 基于自定义逻辑：应用自定义 watermark 生成器。

Kafka Streams 默认使用处理时间 watermark。要使用事件时间 watermark，需要配置 TimestampExtractor：

props.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG,
          MyEventTimeExtractor.class.getName());

窗口聚合的测试策略

窗口聚合的测试比普通聚合更复杂，因为需要考虑时间因素。测试策略：

1. 固定时间测试：使用 TopologyTestDriver 的 pipeInput() 方法，手动控制时间戳，验证窗口边界。
2. 边界测试：在窗口边界发送消息（如 00:04:59 和 00:05:01），验证消息是否进入正确的窗口。
3. 迟到数据测试：发送 grace period 外的消息，验证是否被丢弃。
4. 会话合并测试：发送重叠的会话事件，验证会话是否正确合并。

示例测试代码：

@Test
void windowBoundaryTest() {
    TopologyTestDriver driver = new TopologyTestDriver(topology, props);
    TestInputTopic<String, String> input = driver.createInputTopic(
        "loans", new StringSerializer(), new StringSerializer());
    
    // 窗口 00:00-00:05
    input.pipeInput("L001", "CREATED", Instant.parse("2024-01-01T00:00:00Z"));
    input.pipeInput("L001", "RETURNED", Instant.parse("2024-01-01T00:04:59Z"));
    
    // 窗口 00:05-00:10
    input.pipeInput("L001", "CREATED", Instant.parse("2024-01-01T00:05:01Z"));
    
    // 验证输出
    TestOutputTopic<Windowed<String>, Long> output = driver.createOutputTopic(
        "counts", new WindowedDeserializer<>(), new LongDeserializer());
    
    assertEquals(2L, output.readKeyValue().value); // 00:00-00:05
    assertEquals(1L, output.readKeyValue().value); // 00:05-00:10
}

（本实验未在本机执行，仅作为预期行为说明。）

本节小结

• 三种窗口类型：滚动窗口（固定不重叠）、跳跃窗口（固定可重叠）、会话窗口（动态）。
• 窗口大小必须小于输入 topic 的 retention 时间，否则聚合基于不完整数据。
• Grace period 控制迟到数据的处理，超过 grace period 的数据被丢弃。
• 会话窗口由活动间隙决定，自动合并重叠会话，适合用户行为分析。
• 窗口存储使用 WindowStore，键包含时间戳，状态大小随窗口数线性增长。
• Join 窗口控制两个事件的时间差，与聚合窗口是不同概念。
• 窗口聚合的性能影响主要来自状态存储维护，需合理规划窗口大小和 retention。

阅读导航：上一节 7.2 状态存储 ｜ 下一节 8.1 表与流的互转