java

5.2 Flink 状态管理：OperatorState

Flink OperatorState 机制：ListState 与 BroadcastState 的差异、CheckpointedFunction 接口实现、状态在并行度变化时的重新分配（resscale/union）、BroadcastStream 广播配置流模式，以及与 KeyedState 的联合使用场景。

Leeting Yan 2026-11-01 3 分钟阅读 1141 字

OperatorState vs KeyedState

维度	KeyedState	OperatorState
绑定对象	key	算子实例（并行实例）
访问方式	`getRuntimeContext().getState()`	`CheckpointedFunction` 接口
使用场景	keyBy 后的状态	Source/Sink 或无 key 算子
重分配	按 key hash 自动	自定义重分配策略

ListState

每个算子实例有一个 ListState，在 Checkpoint 时保存列表。

public class BufferingSink implements SinkFunction<Event>, CheckpointedFunction {
    private final List<Event> bufferedElements = new ArrayList<>();
    private ListState<Event> checkpointedState;

    @Override
    public void invoke(Event value, Context context) {
        bufferedElements.add(value);
        if (bufferedElements.size() >= 100) {
            flush();
        }
    }

    private void flush() {
        // 批量写入外部系统
        for (Event event : bufferedElements) {
            externalSystem.write(event);
        }
        bufferedElements.clear();
    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        checkpointedState.clear();
        checkpointedState.addAll(bufferedElements);
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        ListStateDescriptor<Event> descriptor =
            new ListStateDescriptor<>("buffered-events", Event.class);
        checkpointedState = context.getOperatorStateStore().getListState(descriptor);

        if (context.isRestored()) {
            checkpointedState.get().forEach(bufferedElements::add);
        }
    }
}

状态重分配

当作业并行度变化时，OperatorState 需要重新分配：

重分配模式由取状态的 API 决定，没有额外的枚举参数：

ListStateDescriptor<Event> descriptor =
    new ListStateDescriptor<>("events", Event.class);

// 均匀重分配（SPLIT）：每个新实例分得一部分
ListState<Event> split = context.getOperatorStateStore().getListState(descriptor);

// 广播重分配（UNION）：所有状态发给每个新实例
ListState<Event> union = context.getOperatorStateStore().getUnionListState(descriptor);

取法	行为	适用
`getListState(descriptor)`	状态按实例切分后重新分配	大状态、均匀分配
`getUnionListState(descriptor)`	所有状态广播给所有新实例	广播配置、小状态

不要写成 getListState(descriptor, OperatorStateRedistributionMode.UNION)——Flink 1.20 的 OperatorStateStore 没有这个重载，union 语义只能用 getUnionListState 表达。

BroadcastState

广播状态：广播流中的每条记录都会发给每个并行实例，各实例各自保存一份内容相同的副本（不是共享内存）。所以状态变更天然对所有实例可见，代价是状态内存乘以并行度。

// 本章的规则模型用 record；Event 仍是第 4 章的 POJO，读取字段要调 getter
record Rule(String type, String action, double threshold) {
    boolean matches(Event event) {
        return event.getType().equals(type) && event.getAmount() > threshold;
    }
}

// 广播流：配置更新
MapStateDescriptor<String, Rule> ruleStateDescriptor =
    new MapStateDescriptor<>("rules", String.class, Rule.class);

BroadcastStream<Rule> ruleBroadcastStream = ruleStream.broadcast(ruleStateDescriptor);

// 主数据流连接广播流
dataStream
    .connect(ruleBroadcastStream)
    .process(new BroadcastProcessFunction<Event, Rule, Result>() {
        @Override
        public void processElement(Event event, ReadOnlyContext ctx, Collector<Result> out)
                throws Exception {
            // 读取广播状态（只读）
            Rule rule = ctx.getBroadcastState(ruleStateDescriptor).get(event.getType());
            if (rule != null && rule.matches(event)) {
                out.collect(new Result(event, rule.action()));
            }
        }

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

适用场景：动态规则引擎（如风控规则实时更新）。

eventd v3 的告警规则使用 BroadcastState：规则从管理后台通过 Kafka Topic 下发，Flink 作业实时更新规则，无需重启。

KeyedState + OperatorState 联合使用

自定义 Source 经常同时使用两种状态：

public class EventSource extends RichParallelSourceFunction<Event>
        implements CheckpointedFunction {
    // OperatorState：保存 offset
    private ListState<Long> offsetState;
    private long currentOffset = 0;

    @Override
    public void run(SourceContext<Event> ctx) throws Exception {
        while (isRunning) {
            Event event = readFromExternal(currentOffset++);
            synchronized (ctx.getCheckpointLock()) {
                ctx.collect(event);
            }
        }
    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        offsetState.clear();
        offsetState.add(currentOffset);
    }

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

        // 每个子任务只存自己的一个 offset，用 getListState 做均分即可。
        // 若改用 getUnionListState，每个子任务会拿到所有子任务的 offset，
        // 必须按 subtask 索引取自己那条，简单 for 循环取最后一条是错的。
        offsetState = context.getOperatorStateStore().getListState(descriptor);

        if (context.isRestored()) {
            for (Long offset : offsetState.get()) {
                currentOffset = offset;
            }
        }
    }
}

注意：列表里只有一个元素时上面这个循环等价于赋值一次；真正的连接器（如 Kafka Source）并行度变化后需要按子任务重新分配 offset，那段逻辑不能省。

小结

状态类型	作用域	重分配	适用
KeyedState	key	按 hash	keyBy 后的状态
ListState（Operator）	算子实例	UNION/SPLIT	列表、offset
BroadcastState	每个并行实例各一份副本（内容相同，不是共享内存）	广播	动态配置

下一节 5.3 Flink Checkpoint 与故障恢复。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页