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

Operator State，说白了就是算子状态。它跟 Keyed State 最大的区别在于——每个并行子任务维护一份自己的状态，而不是按 key 来划分。我刚开始接触 Flink 时，总觉得 Operator State 用得少，后来才发现，它在某些场景下简直是救命稻草。

举个例子：你有一个 Source 算子，需要记录当前读取到文件的哪个位置。这种场景下，每个并行实例只需要知道自己读到哪了，跟 key 没关系。这时候 Operator State 就派上用场了。

3.1 Operator State 的三种类型

Flink 内置了三种 Operator State，我按使用频率排个序：

状态类型	数据结构	重分配模式	典型场景
ListState	List 结构	均匀切分（Round-Robin）	Source 的偏移量记录、批量缓存
UnionListState	List 结构	广播（每个实例拿到全量）	全局配置快照、小规模元数据同步
BroadcastState	Map 结构	广播（所有实例状态一致）	动态规则更新、模型参数分发

你可能会问：ListState 和 UnionListState 都是 List，区别在哪？

区别就在故障恢复时的状态重分配策略。我画了张图，你看完就明白了。

3.2 ListState：最常用的 Operator State

ListState 的使用场景很明确：每个并行实例需要维护一个元素列表，故障恢复时状态被均匀切分到新的实例上。

我记得之前做过一个项目，从 Kafka 消费数据后需要做去重。当时用了 ListState 来缓存最近 5 分钟的 key，每个并行实例只缓存自己处理过的 key。代码大概是这样的：

public class DedupFunction extends RichFlatMapFunction<Event, Event> {
    private transient ListState<String> recentKeys;

    @Override
    public void open(Configuration parameters) throws Exception {
        ListStateDescriptor<String> descriptor = 
            new ListStateDescriptor<>("recent-keys", TypeInformation.of(String.class));
        recentKeys = getRuntimeContext().getListState(descriptor);
    }

    @Override
    public void flatMap(Event event, Collector<Event> out) throws Exception {
        if (!containsKey(event.getKey())) {
            recentKeys.add(event.getKey());
            out.collect(event);
        }
        // 清理过期 key 的逻辑省略...
    }
}

3.3 UnionListState：全量复制，慎用

UnionListState 跟 ListState 的 API 几乎一样，唯一的区别在状态重分配策略。它会把所有旧实例的状态合并成一个全量列表，然后广播给每个新实例。

什么时候用？我一般只在两种场景下用：

• 全局配置快照：每个实例都存了一份完整的配置列表，恢复时不需要重新拉取
• 小规模元数据：数据量不大，但每个实例都需要全量访问

3.4 BroadcastState：动态规则的利器

BroadcastState 是我个人最喜欢的 Operator State。它专门用来处理「一份数据流给所有并行实例共享」的场景。

典型的用法是：一个数据流是业务数据，另一个流是动态规则。规则流通过 .broadcast() 广播到所有下游实例，每个实例都维护一份完整的规则状态。

// 定义广播状态描述符
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<Output> result = dataStream
    .connect(broadcastStream)
    .process(new KeyedBroadcastProcessFunction<String, Event, Rule, Output>() {
        
        @Override
        public void processElement(Event event, ReadOnlyContext ctx, Collector<Output> out) {
            // 从广播状态中读取规则（只读）
            ReadOnlyBroadcastState<String, Rule> state = ctx.getBroadcastState(ruleDescriptor);
            Rule rule = state.get(event.getType());
            if (rule != null && rule.matches(event)) {
                out.collect(new Output(event, rule.getAction()));
            }
        }

        @Override
        public void processBroadcastElement(Rule rule, Context ctx, Collector<Output> out) {
            // 更新广播状态（可写）
            ctx.getBroadcastState(ruleDescriptor).put(rule.getId(), rule);
        }
    });

3.5 实战对比：三种状态怎么选？

我整理了一个决策表，你遇到具体场景时可以直接对照：

场景	推荐状态	原因
Source 读取文件/消息队列的偏移量	ListState	每个实例独立维护，故障恢复时均匀切分
缓存最近 N 条记录用于去重或排序	ListState	状态按实例切分，不会重复存储
全局配置快照，每个实例需要全量数据	UnionListState	数据量小（MB 级别），需要全量复制
动态规则/模型参数实时更新	BroadcastState	天然支持广播语义，读写分离保证一致性
大状态（GB 级别）的全量复制	都不推荐	考虑用外部存储（Redis/DB）替代

嗯，这里要注意一点：BroadcastState 虽然好用，但它有个限制——不能和 KeyedStream 直接 connect。你需要先 keyBy，再用 connect(broadcastStream)。我刚开始用的时候也犯过这个错，编译不通过才反应过来。