3. Flink 核心概念

本章是 Flink 实时计算的理论基石。我们将从 Flink 的定位与架构出发，逐步深入 DataStream API、时间语义、水位线机制以及状态与检查点。这些概念是构建高频行情灰度发布系统的必备知识，理解它们将帮助你掌握 Flink 处理乱序、延迟与故障恢复的核心能力。

3.1 Flink 简介

Apache Flink 是一个开源的分布式流处理框架，专为有状态计算设计，能够处理无界（实时）和有界（批处理）数据流。其核心设计理念是“流是本质，批是特例”。

Flink 的运行时架构采用主从模式：

• JobManager：负责作业调度、检查点协调与故障恢复。
• TaskManager：负责执行具体的算子任务，每个 TaskManager 包含多个 Task Slot。

3.2 DataStream API

DataStream API 是 Flink 处理无界数据流的核心编程接口。它提供了一系列转换算子，允许开发者以声明式的方式构建复杂的流处理拓扑。

一个典型的 DataStream 程序包含三个步骤：

1. 获取执行环境：StreamExecutionEnvironment.getExecutionEnvironment()
2. 接入数据源：env.addSource(...) 或 env.fromSource(...)
3. 定义转换逻辑：应用 map、filter、keyBy、window 等算子。

常用算子对比：

算子	用途	示例
map	一对一转换	解析行情 JSON 为 Java POJO
filter	条件过滤	只保留买单行情
keyBy	按 Key 分区	按股票代码分区
window	窗口聚合	计算 1 秒内最高价
process	底层处理	自定义状态与定时器

// 示例：DataStream API 基础用法
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream<String> sourceStream = env.socketTextStream("localhost", 9999);

DataStream<StockQuote> parsedStream = sourceStream
    .map(line -> JsonUtil.parse(line, StockQuote.class))
    .filter(quote -> quote.getType().equals("bid"));

parsedStream
    .keyBy(StockQuote::getSymbol)
    .window(TumblingProcessingTimeWindows.of(Time.seconds(1)))
    .reduce((q1, q2) -> q1.getPrice() > q2.getPrice() ? q1 : q2)
    .print();

env.execute("High-Freq Quote Processing");

3.3 事件时间与处理时间

Flink 支持三种时间语义，其中事件时间和处理时间是最常用的两种：

时间语义	定义	优点	缺点
事件时间	数据产生时携带的时间戳	结果准确，不受处理速度影响	需要处理乱序与延迟
处理时间	数据到达算子的系统时间	延迟最低，实现简单	结果不确定，受系统负载影响

设置事件时间需要在执行环境中显式声明：

env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

3.4 水位线（Watermark）

水位线是 Flink 衡量事件时间进度的核心机制。它本质上是一个带时间戳的标记，表示“在这个时间戳之前的数据已经全部到达”。水位线用于触发窗口计算，并处理乱序数据。

水位线的生成策略：

• 周期性水位线：每隔一定时间（如 100ms）生成一次，适用于大多数场景。
• 标点水位线：在数据流中遇到特定标记时生成，适用于数据本身携带进度信息。

常见的水位线生成器：

生成器	行为	适用场景
forMonotonousTimestamps	假设时间戳单调递增	严格有序的数据流
forBoundedOutOfOrderness	允许固定范围的乱序	高频行情（通常允许 100-500ms 乱序）
自定义 WatermarkStrategy	完全自定义逻辑	复杂业务场景

// 示例：允许 200ms 乱序的水位线
DataStream<StockQuote> withWatermarks = parsedStream
    .assignTimestampsAndWatermarks(
        WatermarkStrategy.<StockQuote>forBoundedOutOfOrderness(Duration.ofMillis(200))
            .withTimestampAssigner((quote, timestamp) -> quote.getEventTime())
    );

3.5 状态与检查点

状态是 Flink 实现有状态计算的基础。每个算子可以维护自己的状态，用于存储中间结果或历史数据。检查点（Checkpoint）则是 Flink 保证故障恢复后状态一致性的核心机制。

状态类型：

状态类型	说明	示例
ValueState	存储单个值	记录某只股票的最新价格
ListState	存储列表	缓存最近 10 笔成交
MapState	存储键值对	维护订单簿的买卖盘口
ReducingState	增量聚合	实时计算累计成交量

检查点机制：

• Barrier 对齐：Flink 通过插入 Checkpoint Barrier 到数据流中，确保所有算子在同一时刻保存状态快照。
• Exactly-Once 语义：结合 Barrier 对齐与事务性输出，保证每条数据只被处理一次。
• 状态后端：支持 MemoryStateBackend、FsStateBackend、RocksDBStateBackend，高频行情推荐使用 RocksDB 以支持大状态。

// 示例：使用 ValueState 记录股票最新价格
public static class LatestPriceFunction extends RichFlatMapFunction<StockQuote, Double> {
    private transient ValueState<Double> latestPrice;

    @Override
    public void open(Configuration parameters) {
        ValueStateDescriptor<Double> descriptor =
            new ValueStateDescriptor<>("latestPrice", Types.DOUBLE);
        latestPrice = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void flatMap(StockQuote quote, Collector<Double> out) throws Exception {
        Double current = latestPrice.value();
        if (current == null || quote.getPrice() > current) {
            latestPrice.update(quote.getPrice());
            out.collect(quote.getPrice());
        }
    }
}

检查点配置示例：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 启用检查点，间隔 1 秒
env.enableCheckpointing(1000);
// 设置语义为 Exactly-Once
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
// 设置状态后端为 RocksDB
env.setStateBackend(new RocksDBStateBackend("hdfs:///flink/checkpoints"));