3. Operator State 详解：ListState、UnionListState、BroadcastState 的使用场景与实现

聊完 Keyed State，咱们得把目光转向 Operator State。说实话，很多新手容易把这两者搞混。我刚开始接触 Flink 时也踩过这个坑——明明数据是按分区处理的，为什么还要有 Operator State？

说白了，Operator State 是绑定在 算子并行子任务 上的状态。每个并行实例维护自己的一份状态，跟 key 没关系。你想想看，如果我要记录每个算子的处理进度、要广播一份配置到所有并行实例，Keyed State 就使不上劲了。

3.1 Operator State 的三种“姿势”

Flink 提供了三种 Operator State 的原生实现，我按使用频率排个序：

• ListState — 最常用，每个并行实例维护一个列表
• UnionListState — ListState 的变体，重分配策略不同
• BroadcastState — 所有实例共享同一份状态

这三种状态都实现了 ListCheckpointed 或 BroadcastState 接口。嗯，这里要注意：ListCheckpointed 是 Flink 1.x 的老接口，但很多生产代码还在用，你得认识它。

3.2 ListState：最朴素的 Operator State

ListState 的逻辑很简单——每个并行子任务维护一个 List。快照时把这个 List 存下来，恢复时再还给你。

我在项目中遇到过这样一个场景：需要记录每个 Kafka 分区已经消费到的 offset。用 ListState 就特别合适，每个并行实例只管自己消费的那几个分区。

public class MySourceFunction extends RichParallelSourceFunction<String>
        implements ListCheckpointed<Long> {

    private transient ListState<Long> checkpointedState;
    private List<Long> offsets = new ArrayList<>();

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        ListStateDescriptor<Long> descriptor =
                new ListStateDescriptor<>("offsets", Long.class);

        checkpointedState = context.getOperatorStateStore()
                .getListState(descriptor);

        // 恢复状态
        if (context.isRestored()) {
            offsets.clear();
            for (Long offset : checkpointedState.get()) {
                offsets.add(offset);
            }
        }
    }

    @Override
    public List<Long> snapshotState(long checkpointId, long timestamp) throws Exception {
        // 返回当前 offsets 的快照
        return new ArrayList<>(offsets);
    }
}

3.3 UnionListState：重分配时的“全量拷贝”

UnionListState 跟 ListState 长得几乎一样，唯一的区别在于 状态重分配策略。

为什么会这样？你想想看，当算子并行度从 4 改成 3 时，ListState 会把每个实例的列表均匀分给新实例。而 UnionListState 呢？它会把所有实例的列表 全量广播 给每个新实例。

我曾经在做一个全局去重过滤器时用过 UnionListState。每个实例需要知道所有已经处理过的 ID，重分配时不能丢失任何数据。用 UnionListState 就对了——每个新实例拿到全量数据，自己再做去重。

3.4 BroadcastState：所有实例共享的“公告板”

BroadcastState 是三种 Operator State 里最特殊的一个。它不实现 ListCheckpointed，而是走 BroadcastProcessFunction 或 KeyedBroadcastProcessFunction。

说白了，BroadcastState 就是一块“公告板”。你把配置、规则、模型参数写上去，所有并行实例都能读到一模一样的内容。

我做过一个实时风控系统，规则引擎的规则会动态更新。用 BroadcastState 就特别优雅：

1. 一个低吞吐的广播流，源源不断地推送新规则
2. 所有并行实例订阅这个广播流，更新本地的 BroadcastState
3. 数据流过来时，直接查本地 BroadcastState 做判断

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

// 广播配置流
DataStream<Rule> ruleStream = env
        .addSource(new RuleSourceFunction())
        .broadcast(ruleStateDescriptor);

// 连接数据流和广播流
DataStream<Output> output = dataStream
        .connect(ruleStream)
        .process(new BroadcastProcessFunction<Event, Rule, Output>() {

            @Override
            public void processElement(Event event,
                    ReadOnlyContext ctx,
                    Collector<Output> out) throws Exception {
                // 从 BroadcastState 读取规则
                ReadOnlyBroadcastState<String, Rule> state =
                        ctx.getBroadcastState(ruleStateDescriptor);
                Rule rule = state.get("current_rule");
                if (rule != null && rule.matches(event)) {
                    out.collect(new Output(event, rule));
                }
            }

            @Override
            public void processBroadcastElement(Rule rule,
                    Context ctx,
                    Collector<Output> out) throws Exception {
                // 更新 BroadcastState
                ctx.getBroadcastState(ruleStateDescriptor)
                        .put("current_rule", rule);
            }
        });

3.5 三种状态的选择策略

我画了一张图，帮你快速决策该用哪种 Operator State：

3.6 避坑指南

最后，我把自己踩过的坑总结一下：

• ListState 的序列化问题： 如果你用 ListCheckpointed 接口，返回的 List 元素必须可序列化。我遇到过有人把 KafkaConsumer 对象塞进去，结果快照时直接报错。
• UnionListState 的扩容陷阱： 前面说过了，状态量大时别用 UnionListState。如果非要用，记得在扩容前先做状态清理。
• BroadcastState 的更新延迟： BroadcastState 的更新不是事务性的。如果你在 processBroadcastElement 里更新了状态，但下游还没读到，数据可能会用旧规则处理。我一般会在规则里加版本号，下游做版本校验。
• 状态清理： Operator State 不会自动过期。如果你用 ListState 存 offset，记得定期清理已经提交的 offset，否则状态会无限增长。

Operator State 虽然不如 Keyed State 那么常用，但在特定场景下它是不可替代的。尤其是 BroadcastState，在实时数仓、风控、规则引擎这些需要动态配置的场景里，几乎是标配。你只要记住：按实例分区的用 ListState，需要全量的用 UnionListState，需要共享的用 BroadcastState，基本就不会选错。