Flink 的状态机制

Apache Flink 中的 状态（State） 是流处理中非常重要的概念，用于在无界数据流中保存中间结果、实现容错和一致性。Flink 提供了多种类型的状态管理机制，适用于不同的业务场景。

🧩 一、Flink 状态分类

Flink 的状态可以分为以下几类：

📌 二、各类状态详解

1. Operator State（操作符状态）

✅ 特点：

• 不依赖于 key
• 所有数据共享同一个状态
• 常用于 Source、Sink 等不需要按 key 区分状态的场景

🧠 示例：

public class MyOperatorStateFunction extends RichMapFunction<Integer, Integer> {
    private transient ListState<Integer> state;

    @Override
    public void open(Configuration parameters) {
        state = getRuntimeContext().getListState(new ListStateDescriptor<>("my-state", Integer.class));
    }

    @Override
    public Integer map(Integer value) throws Exception {
        state.add(value);
        return value * 2;
    }
}

2. Keyed State（键控状态）

✅ 特点：

• 每个 key 都拥有自己的独立状态
• 支持高效的分布式状态管理
• 只能在 keyBy() 后的 keyed stream 中使用

⚙️ 支持的 Keyed State 类型：

🧠 示例：使用 ValueState 记录当前最大值

public class MaxValueStateFunction extends RichMapFunction<Integer, Integer> {
    private transient ValueState<Integer> maxValueState;

    @Override
    public void open(Configuration parameters) {
        maxValueState = getRuntimeContext().getState(
            new ValueStateDescriptor<>("max-value", Integer.class)
        );
    }

    @Override
    public Integer map(Integer value) throws Exception {
        Integer currentMax = maxValueState.value();
        if (currentMax == null || value > currentMax) {
            maxValueState.update(value);
        }
        return maxValueState.value();
    }
}

3. Broadcast State（广播状态）

✅ 特点：

• 将一个状态流广播到所有下游任务
• 允许所有任务访问相同的只读状态（可变但需谨慎）
• 常用于配置更新、黑名单、规则匹配等场景

🧠 示例：广播配置流并与主数据流连接

// 定义广播流
DataStream<Rule> ruleStream = env.fromElements(new Rule("blacklist", Arrays.asList("user1", "user2")));
BroadcastStream<Rule> broadcastRuleStream = ruleStream.broadcast();

// 主流与广播流连接
mainStream
    .connect(broadcastRuleStream)
    .process(new KeyedBroadcastProcessFunction<String, Event, Rule, FilteredEvent>() {
        // 实现 processElement 和 processBroadcastElement 方法
    });

🗃 三、状态存储方式

Flink 使用 状态后端（State Backend） 来决定状态如何被存储、检查点和恢复。

🧠 设置方法：

env.setStateBackend(new FsStateBackend("file:///tmp/checkpoints"));
// 或者使用 RocksDB
env.setStateBackend(new EmbeddedRocksDBStateBackend());

🔐 四、状态生命周期与容错机制

🧠 示例：设置状态过期时间

StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.hours(1))
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .build();

ValueStateDescriptor<Integer> descriptor = new ValueStateDescriptor<>("myState", Integer.class);
descriptor.enableTimeToLive(ttlConfig);

📊 五、状态使用建议总结