3. 状态类型详解（下）：Operator State 之 ListState、UnionListState、BroadcastState 实战

上一章我们把 Keyed State 掰开揉碎讲了一遍。今天轮到 Operator State 了。

说实话，我刚接触 Flink 那会儿，总觉得 Operator State 不如 Keyed State 用得频繁。后来做实时数仓的维度关联、做动态规则引擎，才发现 Operator State 才是真正的「幕后功臣」。尤其是 BroadcastState，简直是流式动态配置的标配。

好，咱们直接进入正题。

3.1 Operator State 的定位与特性

Operator State，顾名思义，是绑定在算子实例上的状态。每个并行子任务维护自己的一份状态，不按 Key 划分。

你想想看，Keyed State 是按 Key 分片的，每个 Key 有自己的状态。而 Operator State 是整个算子实例共享的——同一个 subtask 处理的所有数据，看到的都是同一份状态。

Flink 提供了三种 Operator State 的原生实现：ListState、UnionListState、BroadcastState。咱们一个一个来。

3.2 ListState：最基础的 Operator State

ListState 就是把状态存成一个列表。每个 subtask 维护一个列表，状态快照时对整个列表做一次快照。

它的重分配模式是 Even-Split——并行度变化时，Flink 会把列表元素均匀地分给新的 subtask。

我举个例子。假设原来并行度是 2，subtask-0 有 [A, B, C]，subtask-1 有 [D, E]。改成并行度 3 后，Flink 会把 5 个元素尽量平均地分给 3 个 subtask。结果可能是：subtask-0 拿 [A, B]，subtask-1 拿 [C, D]，subtask-2 拿 [E]。

这种模式适合什么场景？我做过一个需求：从 Kafka 消费数据，每个 subtask 需要记录自己处理过的文件偏移量列表。用 ListState 就很自然——每个 subtask 只管自己的列表，重分配时 Flink 自动帮你均分。

来看代码：

public class MyListStateSink extends RichSinkFunction<String> {
    private transient ListState<String> checkpointedState;

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        ListStateDescriptor<String> descriptor =
                new ListStateDescriptor<>(
                        "my-list-state",
                        TypeInformation.of(new TypeHint<String>() {})
                );

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

    @Override
    public void invoke(String value, Context context) throws Exception {
        checkpointedState.add(value);
    }
}

3.3 UnionListState：全量广播的重分配模式

UnionListState 和 ListState 的 API 几乎一模一样。唯一的区别在于重分配策略。

UnionListState 的模式叫 Union——并行度变化时，每个新的 subtask 会拿到所有旧 subtask 的完整列表。

还是刚才的例子。并行度从 2 变成 3，subtask-0 有 [A, B, C]，subtask-1 有 [D, E]。用 UnionListState 的话，每个新 subtask 都会拿到 [A, B, C, D, E] 的完整拷贝。

你可能会问：这不浪费吗？确实，UnionListState 的存储和网络开销比 ListState 大。但它解决了一个关键问题——状态合并。

我在项目中遇到过这样一个场景：用 Flink 做全局去重，每个 subtask 维护一个布隆过滤器。如果使用 ListState，重分配时布隆过滤器被拆成碎片，每个新 subtask 拿到的都是不完整的，去重逻辑就废了。用 UnionListState，每个新 subtask 拿到所有旧过滤器的完整数据，可以在本地做合并。

代码和 ListState 几乎一样，只是把 getListState 换成 getUnionListState：

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

3.4 BroadcastState：流式动态配置的标配

BroadcastState 是 Operator State 里最实用的一个。它的核心思想是：一个流广播到所有并行实例，每个实例维护一份相同的状态。

典型的应用场景：

• 动态规则引擎：规则变化通过广播流下发，所有 subtask 同时更新
• 维度表关联：维度数据变化时广播到所有实例
• 全局配置更新：比如阈值、黑白名单、模型参数

BroadcastState 的使用需要配合 BroadcastStream 和 BroadcastProcessFunction。我直接给一个完整的实战例子——动态规则过滤：

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

// 2. 将配置流广播
DataStream<Tuple2<String, String>> ruleStream = env
        .addSource(new RuleSource())
        .broadcast(ruleDescriptor);

// 3. 连接主流和广播流
DataStream<String> output = mainStream
        .connect(ruleStream)
        .process(new BroadcastProcessFunction<String, Tuple2<String, String>, String>() {

            @Override
            public void processElement(
                    String value,
                    ReadOnlyContext ctx,
                    Collector<String> out) throws Exception {

                // 从广播状态读取规则
                String rule = ctx.getBroadcastState(ruleDescriptor)
                        .get("filter_rule");

                if (rule == null || value.contains(rule)) {
                    out.collect(value);
                }
            }

            @Override
            public void processBroadcastElement(
                    Tuple2<String, String> rule,
                    Context ctx,
                    Collector<String> out) throws Exception {

                // 更新广播状态
                ctx.getBroadcastState(ruleDescriptor)
                        .put(rule.f0, rule.f1);
            }
        });

3.5 三种 Operator State 对比

为了方便你选型，我整理了一张对比表：

特性	ListState	UnionListState	BroadcastState
底层存储	List	List	Map
重分配模式	Even-Split（均分）	Union（全量广播）	全量复制
适用场景	每个 subtask 独立的状态	需要合并的全局状态	动态配置、规则、维度
状态量级	中	大（全量复制）	小（广播流）
API 入口	getListState	getUnionListState	getBroadcastState

3.6 知识体系总览

下面这张图把三种 Operator State 的核心逻辑串起来了。你看一眼就能明白它们各自的位置和关系。

3.7 实战选型建议

讲了这么多，最后给点实在的建议：

1. 能用 Keyed State 就别用 Operator State。Keyed State 的隔离性和性能都更好。Operator State 只在「状态不按 Key 划分」时才用。
2. ListState 是默认选择。如果你的状态只是每个 subtask 独立维护，不需要跨实例合并，用 ListState 最省心。
3. UnionListState 慎用。只有当你明确需要「重分配时全量合并」才用它。状态量超过 100MB 时，建议评估是否能用 BroadcastState 替代。
4. BroadcastState 是动态配置的首选。但记住：广播流的数据量必须小，更新频率不能太高。我一般控制在每秒不超过 100 条更新。