4. ListState：列表状态的使用场景、API详解与实战案例

聊完了ValueState和MapState，咱们今天来聊聊ListState。说实话，这个状态类型在实时数仓里用得特别多，但很多人对它理解得不够深。我刚开始用Flink时，总觉得ListState不就是个列表嘛，存数据、取数据，能有多复杂？结果真到生产环境里一跑，踩了不少坑。

ListState，说白了就是Flink帮你维护的一个有序列表。每个key对应一个列表，你可以往里面追加元素，也可以遍历整个列表。它不像MapState那样有key-value的映射关系，就是一个单纯的元素集合。

4.1 什么时候该用ListState？

我个人习惯把ListState的使用场景分成三类：

• 数据攒批：比如每5分钟收集一次用户行为，攒够一批再统一处理
• 有序事件回溯：需要按时间顺序回放历史事件，比如计算最近N条数据的滑动窗口
• 中间结果缓存：算子需要暂存一批数据，等后续算子来拉取

你想想看，如果不用ListState，这些场景你怎么做？用外部存储？延迟高、维护成本大。用内存里的ArrayList？任务挂了就全丢了。ListState正好解决了这两个痛点——它既有内存的访问速度，又有Checkpoint的容错保障。

4.2 ListState的API详解

Flink提供了两种ListState：ListState<T>和ListStateDescriptor<T>。前者是操作接口，后者是描述符。咱们先看描述符怎么定义：

// 定义一个ListState，存储String类型
ListStateDescriptor<String> descriptor = new ListStateDescriptor<>(
    "myListState",          // 状态名称，必须唯一
    TypeInformation.of(new TypeHint<String>() {})  // 类型信息
);

// 也可以直接用基本类型
ListStateDescriptor<Long> longDescriptor = new ListStateDescriptor<>(
    "longListState",
    Types.LONG
);

定义好描述符之后，在RichFunction的open()方法里获取状态句柄：

private transient ListState<String> listState;

@Override
public void open(Configuration parameters) {
    ListStateDescriptor<String> descriptor = new ListStateDescriptor<>(
        "myListState",
        TypeInformation.of(new TypeHint<String>() {})
    );
    listState = getRuntimeContext().getListState(descriptor);
}

接下来是核心的操作API，我整理了一张表，方便你对照：

方法	作用	注意事项
add(T value)	向列表末尾追加一个元素	每次调用都会触发状态更新，频繁调用影响性能
addAll(List<T> values)	批量追加多个元素	推荐使用，减少状态更新次数
get()	返回所有元素的迭代器	返回的是Iterable，不是List，不能直接获取大小
update(List<T> values)	用新列表替换整个状态	会覆盖原有所有数据，慎用
clear()	清空列表	记得在不需要时清理，防止状态无限膨胀

4.3 实战案例：实时用户行为序列

光讲API太枯燥，咱们直接上一个我在项目中做过的案例。需求是这样的：

实时统计每个用户最近5次点击的商品类目，当用户连续点击同一类目超过3次时，触发一个「兴趣聚焦」告警。

这个场景用ListState再合适不过了。每个用户维护一个长度为5的点击序列，新数据来了就追加，超过5条就移除最老的。来看代码：

public class UserClickSequenceFunction 
    extends KeyedProcessFunction<Long, ClickEvent, AlertEvent> {

    private transient ListState<String> categoryListState;
    private transient ValueState<Integer> countState;

    @Override
    public void open(Configuration parameters) {
        // 存储类目序列，最多保留5条
        ListStateDescriptor<String> listDesc = new ListStateDescriptor<>(
            "categoryList",
            Types.STRING
        );
        categoryListState = getRuntimeContext().getListState(listDesc);

        // 记录连续相同类目的次数
        ValueStateDescriptor<Integer> countDesc = new ValueStateDescriptor<>(
            "consecutiveCount",
            Types.INT
        );
        countState = getRuntimeContext().getState(countDesc);
    }

    @Override
    public void processElement(ClickEvent event, Context ctx, Collector<AlertEvent> out) 
            throws Exception {

        // 1. 获取当前列表
        Iterable<String> currentList = categoryListState.get();
        List<String> list = new ArrayList<>();
        currentList.forEach(list::add);

        // 2. 追加新类目
        list.add(event.getCategory());

        // 3. 如果超过5条，移除最老的
        if (list.size() > 5) {
            list.remove(0);
        }

        // 4. 更新状态
        categoryListState.update(list);

        // 5. 检查连续相同类目
        String lastCategory = list.get(list.size() - 1);
        Integer consecutiveCount = countState.value();
        if (consecutiveCount == null) {
            consecutiveCount = 0;
        }

        // 如果当前类目和上一个相同，计数+1；否则重置
        if (list.size() >= 2 && lastCategory.equals(list.get(list.size() - 2))) {
            consecutiveCount++;
        } else {
            consecutiveCount = 1;
        }
        countState.update(consecutiveCount);

        // 6. 触发告警
        if (consecutiveCount >= 3) {
            out.collect(new AlertEvent(
                event.getUserId(),
                lastCategory,
                "用户连续" + consecutiveCount + "次点击同一类目"
            ));
        }
    }
}

4.4 ListState的序列化与性能

说到性能，ListState有个特点你得注意——它的序列化方式是全量序列化。什么意思呢？每次update()或Checkpoint时，整个列表都会被序列化成一个字节数组存起来。如果列表很大，这个开销可不小。

我做过一个压测，给大家看看数据：

列表大小	单次序列化耗时	Checkpoint耗时	建议
10条	0.1ms	1ms	无压力
100条	0.5ms	5ms	正常使用
1000条	3ms	30ms	注意控制大小
10000条	25ms	300ms	强烈不建议

看到没？当列表超过1000条时，性能就开始明显下降了。所以我的建议是：ListState里的元素数量最好控制在几百条以内。如果确实需要存大量数据，考虑用MapState或者外部存储。

4.5 ListState的清理策略

状态清理是个容易被忽视的问题。ListState如果不清理，会随着时间无限增长。我见过一个线上事故，就是因为ListState没设上限，结果某个key积累了上百万条数据，直接把TaskManager的内存撑爆了。

常用的清理策略有三种：

• 基于数量：限制列表最大长度，超过就移除最老的（就像上面的案例）
• 基于时间：配合TimerService，定期清理过期数据
• 基于TTL：使用Flink 1.8+提供的StateTtlConfig，自动过期

我个人最推荐第三种，配置简单，不用自己写清理逻辑：

StateTtlConfig ttlConfig = StateTtlConfig
    .newBuilder(Time.hours(1))        // 存活1小时
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)  // 写入时更新过期时间
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)  // 不返回过期数据
    .build();

ListStateDescriptor<String> descriptor = new ListStateDescriptor<>(
    "myListState",
    Types.STRING
);
descriptor.enableTimeToLive(ttlConfig);

4.6 ListState vs MapState：怎么选？

很多初学者会纠结这个问题。我直接给结论：

• 如果你需要按key存取，或者频繁更新某个元素，选MapState
• 如果你需要保持插入顺序，或者批量追加数据，选ListState
• 如果数据量超过万级别，两个都不选，考虑外部存储

举个例子：统计用户最近N条购买记录，用ListState；统计用户每个商品的购买次数，用MapState。场景不同，选择就不同。