如何聚合多个Kafka键值对 批量获取所有用户最新视频观看位置

你的需求可以通过Kafka Streams轻松实现，以下是两种常用的落地方式：

方案1：Kafka Streams窗口+聚合（无额外组件依赖） #

核心思路是先通过KTable保留每个用户的最新观看位置，再通过固定30秒滚动窗口做全局聚合，仅在窗口结束时输出全量最新数据：

• 第一步：将输入topic加载为KTable，自动维护每个key的最新value

KTable<String, Integer> userWatchPosition = builder.table("你的topic名称", Consumed.with(Serdes.String(), Serdes.Integer()));

• 第二步：将KTable转成KStream，按30秒固定窗口分组，聚合所有key-value对到同一个集合中，配合suppress操作仅输出窗口关闭时的最终结果

userWatchPosition.toStream()
    // 按固定30秒滚动窗口分组，所有数据归到同一个分组（这里用固定key"global"实现全局聚合）
    .groupBy((key, value) -> "global", Grouped.with(Serdes.String(), Serdes.Integer()))
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofSeconds(30)))
    // 聚合操作：把每个用户的最新值放到Map里
    .aggregate(
        () -> new HashMap<String, Integer>(),
        (aggKey, newValue, aggMap) -> {
            aggMap.put(aggKey, newValue);
            return aggMap;
        },
        Materialized.with(Serdes.String(), Serdes.HashMap(Serdes.String(), Serdes.Integer()))
    )
    // 仅输出窗口完全关闭后的最终聚合结果，避免中间过程输出
    .suppress(Suppressed.untilWindowCloses(Suppressed.BufferConfig.unbounded()))
    .toStream()
    // 这里拿到的就是每30秒全量的用户最新观看位置Map，直接批量写数据库即可
    .foreach((windowKey, allUserPosition) -> {
        // 执行你的批量更新逻辑
        batchUpdateDB(allUserPosition);
    });

方案2：KTable状态存储+定时任务（更易调试） #

如果觉得聚合逻辑不好理解，也可以直接给KTable绑定一个可查询的状态存储，单独起一个定时任务每30秒遍历一次存储中的全量数据：

• 第一步：创建KTable时指定状态存储名称

KTable<String, Integer> userWatchPosition = builder.table(
    "你的topic名称",
    Materialized.<String, Integer, KeyValueStore<Bytes, byte[]>>as("user-position-store")
        .withKeySerde(Serdes.String())
        .withValueSerde(Serdes.Integer())
);

• 第二步：在Kafka Streams应用启动完成后，启动一个定时调度任务，每30秒执行一次全量查询

// 定时任务执行逻辑
@Scheduled(fixedDelay = 30000)
public void batchUpdate() {
    // 获取状态存储
    ReadOnlyKeyValueStore<String, Integer> store = streams.store(
        StoreQueryParameters.fromNameAndType("user-position-store", QueryableStoreTypes.keyValueStore())
    );
    // 遍历所有key获取最新值
    Map<String, Integer> allUserPosition = new HashMap<>();
    try (KeyValueIterator<String, Integer> iterator = store.all()) {
        while (iterator.hasNext()) {
            KeyValue<String, Integer> entry = iterator.next();
            allUserPosition.put(entry.key, entry.value);
        }
    }
    // 执行批量更新
    batchUpdateDB(allUserPosition);
}

常见问题说明 #

你之前用peek只能拿到单条更新记录，是因为peek是在KTable每条记录更新时触发的回调，仅能拿到当前发生变更的单条key的数据，自然无法获取全量的所有key快照。

注意：如果你的用户量超过100万级，全量聚合的Map会占用较多内存，可以考虑按Kafka分区维度分批聚合、分批更新数据库，避免OOM问题。

内容的提问来源于stack exchange，提问作者mzmuda33