4. Operator State入门：ListState、UnionListState、BroadcastState的概念与使用场景

好，咱们今天聊聊 Operator State。说实话，很多初学者一上来就扎进 Keyed State 里，觉得 Operator State 用得少，就不太重视。但我个人的经验是——你迟早会碰上它，而且一旦碰上，如果没搞懂这几个状态的区别，调试起来会非常痛苦。

Operator State，说白了就是跟算子实例绑定的状态。每个并行子任务维护自己的一份状态，彼此不共享。它不像 Keyed State 那样按 key 分区，而是按算子并行度来切分。Flink 内置了三种：ListState、UnionListState 和 BroadcastState。咱们一个一个拆开讲。

4.1 ListState：最基础的 Operator State

ListState 是 Operator State 的默认实现。每个并行实例维护一个列表结构，你可以往里面 add、get、update。我最早用 Operator State 就是在做全局计数器的场景——每个并行度统计自己处理了多少条数据，然后定期汇总。

它的重分配策略是均匀拆分。比如原来有 3 个并行度，状态分别是 [A,B]、[C,D]、[E,F]；现在缩容到 2 个并行度，Flink 会把所有元素打散，重新平均分给 2 个实例。嗯，这里要注意：顺序不保证。

4.1.1 代码示例：使用 ListState 记录每个算子的处理记录

public class MyListStateFunction extends RichFlatMapFunction<String, String> {
    
    private transient ListState<String> checkpointedState;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        ListStateDescriptor<String> descriptor = 
            new ListStateDescriptor<>("myListState", TypeInformation.of(String.class));
        checkpointedState = getRuntimeContext().getListState(descriptor);
    }
    
    @Override
    public void flatMap(String value, Collector<String> out) throws Exception {
        checkpointedState.add(value);
        Iterable<String> allElements = checkpointedState.get();
        // 处理逻辑...
    }
}

你看，代码跟 Keyed State 很像，区别在于不需要 keyBy，直接在 RichFunction 里声明就行。我在项目中经常用它来记录每个算子的启动时间、处理总量这类元信息。

4.2 UnionListState：全量广播的重分配策略

UnionListState 跟 ListState 的 API 几乎一样，唯一的区别在于故障恢复时的重分配方式。UnionListState 会把所有并行实例的状态合并成一个全量列表，然后每个新实例都拿到这个全量列表。

为什么会需要这个？我记得有一次做黑名单过滤的场景：每个算子实例维护一份黑名单列表，但黑名单是全局的，我希望每个实例都能看到完整的黑名单。如果用 ListState，恢复时只会拿到一部分，那就出问题了。

4.2.1 什么时候用 UnionListState？

• 全局配置/规则：每个算子实例都需要完整的配置列表
• 小规模参考数据：比如几百条的白名单、映射表
• 调试/审计：需要每个实例都能回溯所有历史状态

说白了，UnionListState 就是用内存换便利。你想想看，如果状态只有几十 KB，那用 UnionListState 完全没问题；但如果状态有几百 MB，还是老老实实用 ListState 吧。

4.3 BroadcastState：真正的广播状态

BroadcastState 是 Operator State 里最特殊的一个。它跟 DataStream 配合使用，通过 .broadcast() 方法把一个流广播到所有并行实例上，然后每个实例维护一份完整的状态副本。

我个人的经验是，BroadcastState 最常见的用法是动态规则更新。比如实时风控系统：规则流（来自数据库或配置中心）广播到所有算子实例，每个实例拿着最新规则去匹配数据流。规则变了，状态自动更新，不需要重启作业。

4.3.1 代码示例：动态规则匹配

// 定义广播状态描述符
MapStateDescriptor<String, Rule> ruleDescriptor = 
    new MapStateDescriptor<>("rules", String.class, Rule.class);

// 广播规则流
DataStream<Rule> ruleStream = env.fromElements(rule1, rule2);
BroadcastStream<Rule> broadcastStream = ruleStream.broadcast(ruleDescriptor);

// 连接数据流和广播流
dataStream.connect(broadcastStream)
    .process(new KeyedBroadcastProcessFunction<String, Event, Rule, Output>() {
        
        @Override
        public void processElement(Event value, ReadOnlyContext ctx, Collector<Output> out) {
            // 读取广播状态，不能修改
            ReadOnlyBroadcastState<String, Rule> state = ctx.getBroadcastState(ruleDescriptor);
            Rule rule = state.get(value.getType());
            // 应用规则...
        }
        
        @Override
        public void processBroadcastElement(Rule value, Context ctx, Collector<Output> out) {
            // 只有这里可以修改广播状态
            ctx.getBroadcastState(ruleDescriptor).put(value.getId(), value);
        }
    });

你看，processElement 里拿到的状态是 ReadOnlyBroadcastState，只能读不能写。而 processBroadcastElement 里才能修改。这个设计很巧妙——保证了所有实例的状态最终一致。

4.4 三种 Operator State 对比

4.5 知识体系总览

下面这张图帮你理清这三种状态的关系和选择路径：

4.6 实战避坑指南

最后分享几个我实际项目中遇到的坑：

• 别把大状态塞进 UnionListState：我曾经把一个 500MB 的黑名单用 UnionListState 存，结果扩并行度时直接 OOM。后来改成用外部存储 + 缓存解决了。
• BroadcastState 的更新频率别太高：如果每秒更新几千次规则，每个实例都要处理广播流，CPU 会飙升。建议在广播流前面加一个 throttle 或 batch。
• ListState 的序列化开销：每次 checkpoint 都会序列化整个列表。如果列表很大，建议定期清理过期数据，或者用 RocksDB 状态后端。