很多团队做热榜的第一版，逻辑都很直白：定时任务跑一遍 SQL，按某个维度 group by 后 order by desc 取前 100 条，写回缓存。这套方案在上线初期确实够用，但流量涨到一定程度后，三个问题会非常扎眼：榜单永远是五分钟甚至十分钟之前的、上游存储被反复全量扫描、高峰期计算任务挤在一起资源告急。做 Flink 实时计算 topN 热榜，本质上就是把这套定时批处理逻辑，改造成一条持续运行的流式管道，让榜单始终贴着最新数据走。这篇内容我会把从窗口选型、两阶段聚合、乱序处理到资源调优的完整链路拆开讲，适合已经跑过简单 Flink 作业、但想系统做榜单场景的读者。
1. 实时热榜先想清楚三个问题：数据源、口径、输出
1.1 数据源：埋点流里那些容易被忽略的脏数据
做实时热榜的第一步往往不是写窗口，而是先去盘数据源。Flink 作业从 Kafka 消费的埋点流，和离线数仓里那份"洗过的表"完全不是一回事。比如用户行为日志里经常出现字段错位、item_id 为空、事件时间严重超前或落后的数据；还有上游重试机制导致的重复日志，如果一个点击行为被生产者重复发送了两遍，窗口里 count 就会多算一次。
我在之前的实时计算作业里被这类脏数据坑过好几次，后来总结出一套比较稳的做法。过滤规则不要放在业务逻辑里散着写，而是在进入窗口之前单独用一个 FlatMap 统一清洗：item_id 为空直接丢弃，非目标事件类型直接丢弃，时间戳超出当前 watermark 上下界的按各自策略处理。重复日志的去重，如果是精确计数场景，可以在 SQL 层用 ROW_NUMBER() 按主键去重后再聚合；如果是允许近似的场景，这步可以省掉，毕竟排行榜本来就不是银行账户，几个百分点的误差通常可以接受。
1.2 口径：同一条流里跑"小时榜"和"全天榜"的语义冲突
热榜的"热度"有两种常见口径：一种只统计当前自然小时内的行为次数，比如"过去1小时最热商品 Top 100"；另一种是统计从当天零点累计到现在的总次数，比如"今日热卖榜"。这两种口径对窗口的要求完全不一样，前者用滚动窗口就能解决，后者本质是一个无界累计值，用窗口套反而别扭。
比较典型的做法是拆成两条独立的 Flink 作业或两个独立的 SQL 查询来跑。小时榜用 TUMBLE 窗口，窗口大小 10 分钟、滑动步长 30 秒，每 30 秒刷新一次榜单；日榜则用 keyby(itemId) + 状态累计的方式，每次事件到达就把对应计数加一，再每隔固定时间扫描一次状态生成榜单。两条流互不干扰，各自调整并行度和延迟目标。如果硬要把两个口径塞进同一条窗口逻辑，后期改需求的时候会非常痛苦，这一点建议在方案设计阶段就定下来。
1.3 输出目标：榜单不只是算出来，还要让人查得到
榜单算出来只是中间结果，用户端看到的是一个接口服务。Flink 作业的输出目标通常是 Redis、HBase 或 MySQL，关键是写出什么结构。我见过不少人直接把整个榜单序列化成 JSON 塞进一个 key，这样最简单，但更新粒度太粗，一个商品的名次变动就要重写整个榜单，还会引入并发覆盖问题。
更合理的方式是榜单按窗口版本号拆分存储。比如 Redis 里用 ZSet，key 是"hotboard:{窗口开始时间戳}"，score 是计数，member 是 item_id；同时维护一个"当前最新窗口版本"的 key，查询端先读版本号，再读对应 ZSet。这样每个窗口的榜单互不影响，迟到数据更新也只需要按窗口 ID 精准更新对应的 ZSet。Flink 端用批量写入还是逐条写入，要看吞吐要求，但结构上这个版本号设计能帮你避开后面遇到的很多一致性问题。
2. 窗口选型：滚动、滑动与刷新粒度之间的博弈
2.1 滚动窗口为什么是基础，滑动窗口为什么更适合热榜
滚动窗口和滑动窗口的名字很容易让人混淆，但它们在热榜场景里的角色差异非常大。滚动窗口按固定长度切分，彼此不重叠，每个数据只属于一个窗口，语义干净，适合做"某个时间段内"的确定性统计。但滚动窗口的缺点是榜单结果只在窗口结束时输出一次，如果你想要"每 30 秒看到一个接近实时的榜单"，单纯靠滚动窗口做不到。
滑动窗口则连续滑动，同一时刻会有多个窗口重叠存在。比如窗口大小 10 分钟、滑动步长 30 秒，任何一条数据都会同时落入 20 个不同的窗口里，每个窗口都在独立累计。这样一来，每过 30 秒就有一个新窗口闭合并输出结果，榜单看起来就是持续刷新的。热榜业务要的本来就是"滚动刷新"的体验，所以滑动窗口这类方案是主力，滚动窗口反而更像一个语义基础。
2.2 一组建议的窗口参数与背后的代价
参数设置上，我常用的组合是：窗口大小 10 分钟、滑动步长 30 秒、watermark 延迟 5 秒、allowedLateness 30 秒。这套参数能保证端到端榜单延迟在 40 秒以内，大多数运营场景足够用了。
要清楚滑动窗口的代价：窗口数量和滑动步长成反比，10 分钟窗口、30 秒步长意味着每个 key 同时维护 20 个窗口状态。如果业务上还要按城市、频道等多个维度拆榜，状态量会直线上升。所以把窗口开多大、滑多快，不是拍脑袋定的，而是要根据你愿意承受多少状态内存来决定。我见过有人把步长压到 5 秒，结果状态翻了 12 倍，作业内存直接吃紧，只能靠频繁触发 Checkpoint 硬扛，最后背压严重还不如回到 30 秒步长。
2.3 窗口内千万别攒明细：增量聚合的必要性
新手最容易犯的错误是在窗口函数里缓存全部明细，等窗口结束再统一 count。比如用 ProcessWindowFunction 接收一个 Iterable 然后遍历计数。数据量小的时候没事，一旦单窗口内单 key 有几十万条数据，所有明细都进状态，内存和序列化开销立刻失控。
正确做法是让窗口走增量聚合路径。Flink 的窗口可以同时指定增量聚合函数和全量窗口函数，也可以只用 AggregateFunction，这样窗口状态里只保存聚合中间结果，也就是累加器，不保存原始事件。对于 count 类统计，累加器只是一个 Long，不管来多少条明细，状态占用都恒定。这个设计思路是整个热榜作业能扛住高吞吐的关键，一定要从第一版就坚持。
3. 两阶段TopN：先局部浓缩，再全局决胜
3.1 为什么不能在明细流上直接做全局 TopN
如果直接从明细流里 keyBy(itemId) 再做全局排序取前 100，会遇到一个很尴尬的问题：Flink 的排序是分散在各个并行子任务里的，每个子任务只知道本地最大的 100 个，但你不知道这些局部 Top 100 合并起来是不是全局 Top 100。
更关键的是，明细流的体量通常非常大，每秒几万甚至几十万条。把这么多明细全部 shuffle 到一个下游算子做全局排序，下游单点压力巨大，而且状态里存的东西会非常庞大。全局 TopN 不是不能做，而是不能一上来就做。正确思路是先压缩数据规模，把高频的逐条明细变成低频的聚合结果，让进入第二阶段的输入量降低好几个数量级。
3.2 第一阶段：按 key 预聚合，把千万条变成几百条
第一阶段的核心是"窗口内预聚合"。不管用什么窗口，先按 item_id 分组，用增量聚合把同一个 key 在一个窗口内的所有事件合并成一个计数结果。这一步之后，窗口输出的不是明细，而是类似"item_123，窗口 10:00-10:10，数量 4521"这样的一条记录。
为什么说这是整套方案的骨架？因为两阶段设计把压力从"全量明细的高基数 shuffle"变成了"聚合结果的低基数广播"。举个例子，一个 10 分钟窗口内有 1000 万个行为事件，涉及 2 万个商品，预聚合后只有 2 万条计数结果。第二阶段对这 2 万条计数做全局排序取前 100，计算量完全可控。实测下来，这一步通常能把下游算子输入减少 90% 以上，作业吞吐提升比调并行度明显得多。
3.3 第二阶段：维护候选池的 TopN，并解决同分排序
第二阶段收到的输入已经是"商品 + 计数"这样的浓缩结果。这一阶段的核心是设计一个可序列化的 TopN 候选池。很多人直接用一个 TreeMap 按 count 排序，把 map 对象放在普通成员变量里，这在低并行度、小数据量的 Demo 里没问题，但一旦开 Checkpoint，非托管状态不会被持久化，重启恢复后榜单就全乱了，属于埋雷。
正确做法是把候选池放进 Flink 的托管状态，比如 ValueState 或 ListState，类型是一个自定义的可序列化 TopN 容器。容器内部用 TreeMap<Long, List > 按计数排序，同一个计数对应的多个商品放进一个 List，这样既能保证取前 100 时遍历效率，又解决了大量商品 count 相同、排序不稳定导致的榜单抖动问题。排序的时候还需要一个业务上稳定的次级 key，我通常用 item_id 升序作为 tie-breaker，保证同样计数的商品每次排序结果一致。
3.4 核心代码：预聚合与 TopN 容器实现
第一阶段的预聚合，Flink SQL 里可以直接这样写：
CREATE TABLE behavior_source ( user_id BIGINT, item_id STRING, event_time TIMESTAMP(3), WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND ) WITH ( 'connector' = 'kafka', 'topic' = 'behavior_topic', 'properties.group.id' = 'realtime_hotboard', 'scan.startup.mode' = 'latest-offset', 'format' = 'json' ); CREATE VIEW windowed_cnt AS SELECT item_id, COUNT(*) AS cnt, TUMBLE_START(event_time, INTERVAL '10' MINUTE) AS win_start, TUMBLE_END(event_time, INTERVAL '10' MINUTE) AS win_end FROM behavior_source GROUP BY item_id, TUMBLE(event_time, INTERVAL '10' MINUTE); SELECT item_id, cnt, win_start, rn FROM ( SELECT item_id, cnt, win_start, ROW_NUMBER() OVER ( PARTITION BY win_start ORDER BY cnt DESC, item_id ASC ) AS rn FROM windowed_cnt ) WHERE rn <= 100;第二阶段用 DataStream API 写一个带状态的 TopN 聚合函数，核心逻辑是这样的：
public static class TopNAggregateFunction extends AggregateFunction<ItemCount, TopNBuffer, List<ItemCount>> { private final int threshold; public TopNAggregateFunction(int threshold) { this.threshold = threshold; } @Override public TopNBuffer createAccumulator() { return new TopNBuffer(threshold); } @Override public TopNBuffer add(ItemCount value, TopNBuffer acc) { return acc.put(value); } @Override public TopNBuffer merge(TopNBuffer a, TopNBuffer b) { b.iterator().forEachRemaining(a::put); return a; } @Override public List<ItemCount> getResult(TopNBuffer acc) { return acc.sortedList(); } }TopNBuffer 内部持有 TreeMap 和当前最小阈值，插入时如果超过 threshold，就丢掉最小的一条。这个容器本身要能被 Flink 序列化，所以注意别在里面放不可序列化的依赖。实际项目中，第二阶段往往还需要再配一个 ProcessFunction 或定时器，把最终榜单按窗口版本号写入外部存储，这部分见下一章节。
4. 乱序、迟到数据与榜单"翻旧账"的博弈
4.1 watermark 和 allowedLateness 的 trade-off
事件时间处理里，watermark 代表"这个时间之前的数据已经到齐了"的假设。Flink 默认的 watermark 延迟设置得很保守，如果业务对延迟要求高，延迟可以设小一点；如果对完整性要求高，延迟就要设大一点。热榜场景通常选中间值：5 秒到 10 秒的延迟是比较常见的折中。
但 watermark 只能应对乱序，不能保证数据必定完整。网络抖动、上游重启、Kafka 分区的分区数调整，都可能导致部分数据晚到超过 watermark。为了让这些数据还能参与统计，就需要 allowedLateness。这个参数允许窗口在闭合后继续保留状态一段时间，等待迟到数据并重新触发计算。代价是状态不能立刻释放，窗口结束 30 秒后内存才会回收。对热榜来说，我倾向于把 allowedLateness 设成窗口滑动步长同量级，比如 30 秒，既保证大部分迟到数据能补救，又不至于让状态堆积太夸张。
4.2 榜单已经查出结果了，迟到数据来了怎么办
滞后数据真正麻烦的地方在于：上一版榜单可能已经写进 Redis、用户已经看到了，新数据一到，榜单又得改。这不是 Flink 层面的问题，而是结果读取侧的一致性问题。
我的建议是把"榜单查询"和"榜单计算"彻底解耦。Flink 计算端只负责生成带窗口版本号的完整榜单快照，下游通过版本号决定是否要覆盖旧结果。比如 10:10 窗口的结果先写入了 Redis，10:10 窗口又因为迟到数据重算了一次，Flink 再次写出这同一版本号的榜单，外部存储按 version 字段做覆盖。查询端始终读最新已完成的版本，不展示中间计算中的状态。这样即使存在多次重算，最终只会收敛到一份正确结果，整个过程对用户无感。
4.3 幂等写 Redis：ZSet 与窗口版本号的配合
写 Redis 的具体实践中，我比较推荐用 Pipeline 批量写入 ZSet。假设当前窗口版本号为 1710000600，先删除该版本号对应的 ZSet，再批量 ZADD 写入新的 Top 100，两步用事务或 Lua 脚本包起来，保证要么都成功要么都失败。这样即便 Flink 重算多次，每个版本最终都是完整覆盖，不会出现旧高分和新低分混在同一个 ZSet 里的情况。
如果对延迟要求更苛刻，可以不删重建，而是逐条比较 score 后更新，但那样需要在 Redis 里先读出存量列表，逻辑更绕，收益却不大。热榜这种每秒都可能刷新的场景，快照式覆盖是最不容易出错的做法。Flink 端只需保证写出顺序是按窗口版本号单调递增的，配合外部存储的原子操作，最终一致性就能得到满足。
5. 数据倾斜、背压和状态膨胀：三个真实会遇到的坑
5.1 热门 key 把单个并行子任务压满后的现象
热榜场景有个天然的不均匀性：少部分商品贡献了大部分流量。keyBy(itemId) 之后，那两三个头部商品的计数全部落到了同一两个子任务上，其他子任务闲得发慌，这两个子任务 CPU 打满、背压不断。典型现象是 Flink UI 上某些子任务的 Records In 明显偏高，Checkpoint 时长越来越长，最后整个作业延迟掉到分钟级。
这个坑几乎每个做热榜的人都会踩一次。最直接的缓解手段是提高 key 的分布粒度，也就是"加盐"。比如为每个 item_id 生成 8 个逻辑分片，key 变成 item_id + salt，让同一个商品的数据先散到 8 个子任务分别计数，最后在第二阶段再把同一商品不同盐值的计数合并。注意加盐只是缓解倾斜，不能完全消除，真正热门的 key 无论怎么加盐，第二阶段合并时仍然有集中处理，但那时输入量已经小很多，压力是可控的。
5.2 MiniBatch 与两阶段聚合：倾斜缓解的组合拳
Flink 1.16 之后 MiniBatch 是默认结合本地聚合生效的，但很多人没意识到 MiniBatch 的缓冲区大小对倾斜场景的影响。MiniBatch 的本质是把一小段时间的输入攒在本地，合并重复 key 后再往下游发，从而降低 shuffle 和下游压力。如果 MiniBatch 窗口设得太小，比如默认的 10ms 级别，热门 key 的冲击还是会直接打到底层；如果设得太大，延迟又上去了，榜单刷新会变肉。
我一般把 MiniBatch 设置在 100ms 到 200ms 之间，同时配合 5.1 的加盐方案。因为预聚合阶段已经把输入从明细变成了计数，每个 key 的输出频率本身不高，MiniBatch 在这里的作用更多是平滑毛刺。另外还要留意 group by 字段是否包含了盐值，如果 Salt 字段被加进 key，最终合并前的数据分布就会更均匀，这是两阶段聚合能生效的前提。
5.3 Checkpoint 时长与状态 TTL 的管理
热榜作业的状态量其实不大，但如果窗口太多、allowedLateness 太长，状态也可能膨胀。Checkpoint 时长一旦接近或超过窗口滑动步长，就会拖垮整体吞吐。解决思路有两类：一是调大 Checkpoint 间隔和超时时间，二是主动清理不再需要的数据。
定期清理可以用 Flink 的 State TTL，给状态里的缓存加上过期时间。但 TTL 对窗口状态不生效，窗口状态的生命周期是由窗口结束时间决定的，所以清理的关键在于减小 allowedLateness、缩短窗口最大保留时间。对于榜单这种"只看最近 10 分钟"的场景，窗口闭合后 30 秒内没有迟到数据进来，状态就可以果断释放，不需要把 TTL 拉得太长。
5.4 榜单输出的对象复用与序列化优化
最后提一个容易被忽略但实际影响明显的点：Flink 作业里高频产生的中间结果对象，尽量复用而不是频繁 new。TopN 候选池里的 ItemCount 对象如果每条输入都 new 一个，GC 压力在高峰期会非常明显。比较通用的做法是在 AggregateFunction 的 add 方法里更新已有对象，而不是新建对象；能复用 Accumulator、复用 List、复用容器就复用。
序列化层面也值得花点时间。如果 Flink 作业启用了 Kryo，而你的 POJO 类没有合理实现序列化，可能导致状态体积莫名膨胀、恢复时间加长。建议使用 POJO 类型并启用 Flink 原生的 TypeInformation 序列化器，检查日志里是否出现"Kryo fallback"的提示。某些自定义 TopN 容器如果用了复杂泛型，很容易触发 Kryo 兜底，此时手动注册 Serializer 或者调整类结构，能换来可观的性能提升。
整个方案跑下来，我最大的感受是：实时热榜的难点不在于"算出 TopN"这一步，而在于把数据源、窗口、状态、输出串成一条在流量波动下仍然稳定的链路。窗口选型决定了榜单的刷新体验，两阶段聚合决定了作业能否扛住高吞吐，状态设计和外部存储版本号决定了迟到数据场景下的最终一致性，这三层都做扎实了，热榜服务基本就能稳定运行很久。如果你也在做类似的实时榜单，可以先从 10 分钟窗口、30 秒滑动、两阶段聚合这一套组合起步跑通，再根据业务量逐步调整。
