Flink一致性级别详解：从Checkpoint到端到端精确一次-开发者社区

Flink从入门到上天系列第二十五篇：Flink中的一致性级别

做流计算做了几年，我越来越觉得"一致性级别"是Flink里最容易被误解、也最容易被吹过头的一个概念。前几篇我们聊了状态、聊了Checkpoint，这篇终于轮到把一致性级别完整讲透。很多人以为把CheckpointingMode.EXACTLY_ONCE一开就万事大吉，结果线上数据一对比，发现该重复的照样重复。问题出在哪？出在大多数人只看到了Flink内部的"状态一致性"，而忽略了从数据源到中间计算再到外部存储的整条链路上，每一环都在参与一致性的最终结果。这篇不整虚的，从原理到配置，从坑到取舍，一次性讲明白。

1. 先从一次线上事故说起：一致性级别到底在解决什么问题

1.1 一个典型的流计算场景

先说个真实发生过的事。某次我给一个交易系统做实时金额统计，数据源是订单流水，逻辑其实不复杂：按用户ID分组，累加消费金额，结果写到结果表。

一开始作业跑得很正常，积压也不大，我也就没太在意恢复策略。后来有一次运维在凌晨重启了集群，作业从Checkpoint恢复了，按理说数据应该接着跑。但第二天对账的时候发现，结果表里的累计金额比真实金额多了十几万。查了半天，定位到问题：部分订单在故障恢复之后被重复处理了，累加逻辑把这些重复数据又加了一遍。

这个场景背后的问题就是一致性级别。所谓一致性级别，指的是分布式流计算系统在发生故障、恢复、重放等异常情况下，计算结果与真实计算结果相比，最多能偏差到什么程度。我当年踩这个坑的时候还没系统地理解这套概念，后来才彻底想明白。

1.2 三种语义级别的直观理解

Flink里常见的一致性级别有三种，名字不复杂，但每个词背后都是一套完整的工程考虑。

最多一次（At-Most-Once）：数据最多被处理一次。丢了也就丢了，不会重复。这是效率最高、容错最弱的方式，因为故障恢复后干脆不重放数据。适合那些丢了也无所谓的场景，比如一些噪声日志、实时展示用的中间指标。说实话，现在正经业务用这个级别的极少，除非你的吞吐压力大到了必须牺牲一切可靠性的地步。

至少一次（At-Least-Once）：数据至少被处理一次，可能重复。故障恢复后从头重放数据，保证不丢，但无法避免某些数据被处理两遍。这个级别能应对大多数非资金类业务。实时大屏的PV统计、推荐特征打分这种，多算个几次问题不大。

精确一次（Exactly-Once）：数据被处理且仅处理一次，不多不少。一听就是强迫症福音，但在分布式系统里这几乎是一种"理想状态"——因为故障恢复、网络重试、数据重放这些事天然就会导致重复。要想真正做到精确一次，需要从源头到终端全链路配合。

我把三者用一个老百姓都能懂的方式类比一下你去餐厅点餐：

语义级别	类比	后果
最多一次	服务员问你菜要几份，你说一份，他只听了一半就去下单了，你最后没吃上	可能丢数据
至少一次	服务员怕你饿着，菜给你上了两遍	可能重复数据
精确一次	服务员精准记录，一份不多一份不少	不丢不重

这里面有个关键点：Flink的精确一次，不等于端到端的精确一次。这是很多人后面会遇到的第一堵墙。你配置的EXACTLY_ONCE只是让Flink内部的状态处理和恢复逻辑保证了精确一次语义，可你的数据源能不能重放、结果Sink支不支持事务或幂等，那是另一码事。后面第三部分我会专门展开讲这条链路上的分工。

2. Checkpoint机制：Flink内部保证一致性的地基

2.1 Barrier：数据流里的分界标记

聊一致性级别，不可能绕开Checkpoint。Flink的Checkpoint本质上是一种周期性生成的分布式快照——每隔一段时间（比如60秒），把每个算子的状态给"拍个照"存起来。等故障发生了，直接从最近一次成功的快照恢复。

但问题来了：状态是分散在很多并行子任务里的，一个上游Source、中间多个算子、多个分区，它们各自的状态如何保证是"同一时刻"拍下的？如果各拍各的，恢复出来的状态在时间线上就是错乱的。

Flink采用的方案是Chandy-Lamport分布式快照算法，实现方式是在数据流里插入一种特殊标记——Barrier（屏障）。Barrier本身是一条特殊事件，携带Checkpoint编号，随着数据流动。某个Source要发起Checkpoint n时，它的每个并行实例都会往数据流里发送编号为n的Barrier。下游算子收到这个Barrier，就相当于收到一条指令："处理到这里，可以准备做快照了"。

一句话总结Barrier的作用：它把无边无际的数据流切成了一个个有明确边界的区间，数据落在哪个区间，就属于哪个Checkpoint周期。故障恢复时，从一个Checkpoint恢复，就相当于把数据处理的状态回退到"所有输入流都恰好处理完该Checkpoint对应数据"的那一刻。

2.2 为什么"对齐"决定了精确一次还是至少一次

如果要让快照真正达到"全局一致"的效果，就绕不开一个词：Barrier对齐（Alignment）。

现在假设某个下游算子有两个输入通道（比如来自两个上游分区，或者一个KeyBy之后的多路输入）。Checkpoint n的Barrier到了通道A，但通道B的Barrier还没到。这时候算子的状态应该怎么办？

如果不能处理"先到的Barrier"，通道A里Barrier后面的数据会被缓存住，算子继续处理通道B里还没到Barrier的数据。等通道B的Barrier也到了，再统一做快照。这就是对齐。对齐期间，先到的数据被暂时挡住，不进入计算，这样快照出来的状态就精确对应着"两个通道都处理到Barrier那一刻"的全局一致状态。

但如果Flink不要求对齐呢？一看到Barrier就拍快照，哪怕另一个通道还在一路狂奔处理数据。那么快照里可能已经混入了通道B中Barrier之后的数据。恢复的时候，这些数据会被重新处理一遍，于是重复就发生了。

这就是对齐与不对齐的本质区别：对齐，状态是全局一致的，配合Checkpoint就能实现内部精确一次；不对齐，快照粒度粗糙，只能做到至少一次。

所以你在配置里看到的AT_LEAST_ONCE和EXACTLY_ONCE，在Flink内部最直接的体现，就是要不要做Barrier对齐。精确一次的代价，是每次Checkpoint期间，所有并行子任务都要停下来等最慢的那一路Barrier到达，这本身就制造了一种短暂的"停顿"和潜在的反压。Efficiency vs Consistency，这里第一次出现矛盾，后面我还会详细说。

顺便提一个现在越来越常用的功能：如果你用Flink 1.11以上版本，可以开启Unaligned Checkpoint（非对齐Checkpoint）。它不等Barrier都到齐，而是把还在传输管道里没有处理完的in-flight数据，连同状态一起存进快照。恢复的时候把这段in-flight数据也恢复出来，因此它依然能保证精确一次，同时不再需要等待对齐，对反压场景特别友好。代价是快照体积变大、恢复时间变长，得权衡着用。

3. 端到端一致性：Source、内部状态、Sink三环缺一不可

3.1 三环拆解与分工

老生常谈的一句话：流处理里的端到端一致性，是Source端 + Flink内部 + Sink端三环共同作用的结果。每一环解决的是不同的问题。

Source端要解决的是"数据能不能回放"。故障恢复时Flink会把状态回滚到Checkpoint时刻，然后重新消费从Checkpoint位点往后的数据。如果Source不能重置位点、不能回放历史数据，那恢复后数据就直接断层了。好在Kafka这类消息队列都支持按offset消费，Flink的Kafka Source也会把当前消费位点作为状态的一部分存入Checkpoint。恢复时位点跟着状态一起回滚，再往后拉数据，Source端这一环就闭环了。

Flink内部要解决的是"状态和计算结果能不能精确回滚"。这一环就是我前面说的Checkpoint + Barrier对齐。它保证的是：故障恢复后，算子内部的状态（比如累加值、窗口缓存）精确对应到某个一致的时间点。

Sink端要解决的是"写出去的数据能不能不重复、不丢"。这一环最麻烦。为什么？因为外部存储系统（数据库、消息队列、文件系统）的控制权不在Flink手上。Flink不可能像回滚自己的状态一样，去把MySQL里已经写进去的数据给撤回来。所以Sink端必须另想办法。

三环是串联的关系：Source的数据能回放，Flink内部能回滚，但Sink写重复了，那整体就是重复；Sink能做到不重复但Source不可回放，那整体可能丢数据。端到端一致性 = 短板效应，三个环节里最差的那环，决定了最终的一致性级别。

3.2 Sink的两条出路：幂等写入与事务写入

既然Sink端的写入无法随Checkpoint原子回滚，主流方案就两条路：幂等写入和事务写入。

幂等写入的思路很简单：允许重复操作，但重复操作不会改变最终结果。比如往Redis里写一个值是用SET覆盖而不是INCR；往MySQL里写数据用的是INSERT ... ON DUPLICATE KEY UPDATE；往ES里按文档ID覆盖写。重复写几次都无所谓，因为最后落库的值是同一个。这类方案的特点是：性能好、实现简单、不需要外部系统支持乱七八糟的分布式事务。但前提是业务本身必须有天然幂等键或值覆盖语义。如果你写的是一个明细表，没有唯一约束，主键是自增ID，那重复写两遍就会插入两行，幂等根本无从谈起。

事务写入的思路是：把结果数据积累成一个事务，Checkpoint成功之后才真正提交给外部系统。事务提交之前，外部系统看不到这批数据（或者看到的是一个待定状态）；Checkpoint失败、作业重启，事务就被回滚（Abort）掉，未提交的数据不会出现在结果里。这类方案能真正做到"不多不少"，但对Sink的外部系统有硬性要求——它必须支持事务（或者支持类似分布式事务的协议）。Kafka 0.11以上支持事务；MySQL需要开启XA；HBase本身不支持跨多个Region的行级事务，那就没办法用这个方案。

3.3 两阶段提交在Flink里的落地方式

事务写入在Flink里的标准实现是两阶段提交（Two-Phase Commit，2PC），对应Flink提供的一个抽象类：TwoPhaseCommitSinkFunction。

两阶段提交的流程可以通俗地理解为：

正常处理阶段：算子积累了一批数据，调用preCommit()方法，通知外部系统"我要准备提交了"。此时数据处于预提交（pre-commit）状态，对外不可见或半可见。
Checkpoint成功触发：Flink的Checkpoint Coordinator（协调器）在所有算子都完成快照之后，回调Sink的commit()方法，让事务真正提交，数据对外可见。
失败或超时：如果Checkpoint失败，则调用abort()方法，事务回滚，之前预提交的数据被丢弃。

这套设计的精妙之处在于：Flink用自己Checkpoint的完成时机，作为2PC的协调节点。Checkpoint没有全部成功，事务就不会提交。这样"Flink状态回滚"和"外部数据提交"就绑定成了同一个原子操作。

但2PC也带来了新的问题：事务期间的数据对外是隔离的（没人应该读到未提交的数据）；事务提交有延迟，实时性会降低；外部系统的事务也有超时限制，Checkpoint周期太长会导致事务被外部系统强制终结。这些就是今年第五部分要讲的坑的源头。

4. Kafka进Kafka出：端到端精确一次的参数级配置

4.1 完整配置代码

理论说得再多，不如来一份能直接抄的配置。下面以最常见的"Kafka进、Kafka出"链路为例，给出一个端到端精确一次的完整配置框架。我用的DataStream API的口径相对通用，读者按自己的Flink版本微调即可。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 1. 开启Checkpoint，并显式指定精确一次模式 env.enableCheckpointing(60000, CheckpointingMode.EXACTLY_ONCE); CheckpointConfig checkpointConfig = env.getCheckpointConfig(); // 2. 两次Checkpoint之间至少间隔5秒，防止两个Checkpoint重叠 checkpointConfig.setMinPauseBetweenCheckpoints(5000); // 3. 单个Checkpoint从开始到结束必须10分钟内完成，否则判定失败 checkpointConfig.setCheckpointTimeout(600000); // 4. 同一时间只允许一个Checkpoint在进行 checkpointConfig.setMaxConcurrentCheckpoints(1); // 5. Checkpoint连续失败3次才让作业失败，避免偶发问题直接杀死作业 checkpointConfig.setTolerableCheckpointFailureNumber(3); // 6. 作业手动取消时保留Checkpoint状态文件，方便后续恢复 checkpointConfig.setExternalizedCheckpointCleanup( ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION); // 7. 如果经常因为Barrier对齐导致背压，可开启非对齐Checkpoint checkpointConfig.enableUnalignedCheckpoints(true); // 8. 设置状态后端存储路径 checkpointConfig.setCheckpointStorage("hdfs:///flink/checkpoints");

接下来是Source和Sink的关键配置。Kafka Source负责把消费位点信息写入Checkpoint：

Properties sourceProps = new Properties(); sourceProps.setProperty("bootstrap.servers", "kafka-broker:9092"); sourceProps.setProperty("group.id", "flink-consistency-demo"); FlinkKafkaConsumer<String> kafkaSource = new FlinkKafkaConsumer<>( "input_topic", new SimpleStringSchema(), sourceProps ); // 从最新位点开始消费 kafkaSource.setStartFromLatest(); // 消费位点会随Checkpoint一起保存，这是Source端可回放的关键

Kafka Sink要开启EXACTLY_ONCE语义，就涉及事务了：

Properties sinkProps = new Properties(); sinkProps.setProperty("bootstrap.servers", "kafka-broker:9092"); FlinkKafkaProducer<String> kafkaSink = new FlinkKafkaProducer<>( "output_topic", new SimpleStringSchema(), sinkProps, FlinkKafkaProducer.Semantic.EXACTLY_ONCE ); env.addSource(kafkaSource) .map(line -> parseAndProcess(line)) .keyBy(event -> event.getUserId()) .process(new StatefulProcessFunction()) .addSink(kafkaSink);

4.2 关键参数逐项说明

这份配置里每个参数都不是摆设，我按重要性一个个说。

enableCheckpointing(60000, CheckpointingMode.EXACTLY_ONCE)：前一个参数是Checkpoint间隔，后一个是语义模式。间隔的选择很讲究。间隔太短，比如10秒，每次Checkpoint的分布式快照都会产生不小开销；间隔太长，比如30分钟，一方面恢复时要重放30分钟的数据，另一方面Sink端事务挂太久容易触发外部系统的超时。我自己在大多数业务里取1~5分钟。

setMinPauseBetweenCheckpoints(5000)：这个参数常被忽略。它指的是上一次Checkpoint结束到下一次Checkpoint开始之间至少间隔多久。不加这个限制的话，如果数据量大、Checkpoint消耗时间很长，Flink可能会一个接一个地启动新的Checkpoint，导致状态后端被反复写入，严重的会拖垮整个作业。

setMaxConcurrentCheckpoints(1)：精确一次语义下，Checkpoint必须串行。因为如果同时有两个Checkpoint在跑，Sink端两个事务同时处于预提交状态，一旦先提交了老事务再提交新事务，数据顺序就乱了。这里必须设为1。

setTolerableCheckpointFailureNumber(3)：连续失败多少次才让作业失败。注意是"连续失败次数"，不是累计次数。设得太小，比如1，网络抖动一次就直接杀掉作业；设得太大，可能会掩盖配置上存在的本质问题。3是比较中庸的默认值。

enableUnalignedCheckpoints(true)：这是在1.11版本之后才有的选项，适合"数据倾斜严重导致某一条流Barrier迟迟不到位、整条作业反压爆表"的场景。开了之后Barrier不用对齐，快照会把in-flight数据一起存下来。但注意快照体积会明显增大，恢复耗时变长，前提是磁盘和网络扛得住。

4.3 这样配的代价是什么

端到端精确一次不是免费的午餐。这套配置跑起来之后，你会明显感受到几个副作用。

下游可见性变差：Kafka Sink开启EXACTLY_ONCE语义后，数据要等Checkpoint完成、事务提交，下游才能读到。也就是说下游消费者看到的实时延迟，从"毫秒级"变成"一个Checkpoint周期"的量级。如果Checkpoint是1分钟一次，那下游数据延迟就是1分钟±几秒。对账类的场景可以接受，但对实时性要求极高的面板类场景可能要犹豫一下。

事务开销：Kafka开启事务后，每次写入都涉及事务日志；外层还多了一层预提交、提交的协调工作。同样的吞吐，开销比AT_LEAST_ONCE模式高。

Backpressure放大：Barrier对齐机制天然会把慢节点的延迟传导到所有并行子任务。某个上游分区的数据量一旦不均匀，所有实例都得等着。这个问题严重的话，我建议认真考虑非对齐Checkpoint，而不是硬扛。

状态存储膨胀：Checkpoint本身要存状态，非对齐模式还要存in-flight数据。如果状态本身就不小（比如几十GB），每60秒拍一次快照，对HDFS和网络都是压力。

5. 真实踩坑记录与一致性级别选型建议

5.1 坑一：Barrier对齐把慢分区变成全场瓶颈

这个坑我印象太深了。有一次跑一个精确一次的作业，上游Kafka共有12个分区，整体流量不算大，但有一个分区里绑着十几个大卖家的高并发订单数据。运行一段时间后，作业的反压水位一路上涨，Checkpoint的完成时间从几十秒涨到十几分钟。

一开始我以为是某个算子处理逻辑慢，看了一圈，发现确实执行效率没问题。最后把Checkpoint的耗时细分数据拉出来，才发现问题全在Barrier的等待阶段：那一个慢分区的Barrier迟迟不到，其他11个分区的数据全被Block住，整个作业被一只木桶的短板拖死。

后来我做了两个改动：一是开启enableUnalignedCheckpoints(true)，让快照不等对齐，in-flight数据直接存快照，作业健康度立刻好转；二是在数据入Kafka之前，对Key做二次散列，尽量让大卖家流量打散到多个分区去。我个人的经验是：如果对账类场景要求精确一次，而数据天然倾斜，优先考虑Unaligned Checkpoint，代价是快照变大，但比整条作业卡死强太多。

5.2 坑二：Kafka事务超时上限冲突

还有一个经典问题，报错信息大概是"Transaction timeout exceeded"或者"Invalid transaction timeout"。核心矛盾在于事务超时时间的上下限不一致。

Kafka Sink在开启EXACTLY_ONCE后，会默认开启一个事务，事务的超时时间默认是1小时。但Kafka Broker端有一个配置叫transaction.max.timeout.ms，默认值是15分钟。如果事务运行时间超过了Broker允许的上限，Broker会直接强制Abort这个事务。Flink这边就会不断重试，最后作业失败。

这个问题在Checkpoint间隔大、数据量大、事务迟迟提交不了的时候特别容易触发。解决办法有两个方向：一个是调大Broker端transaction.max.timeout.ms，比如调到1小时以上；另一个是在Flink Sink端把事务超时时间缩短，让它低于Broker的上限。最怕的是两边都按默认值，然后出了错还一头雾水。我现在配置Kafka Sink时，第一件事就是检查两边的超时参数是否匹配。

5.3 坑三：下游读了还没提交的数据

这个坑更隐蔽，因为它不报错，只是数据表现诡异。

Kafka Sink开启事务后，数据在预提交阶段只写了事务日志，主日志里还看不到。正常来说下游消费者读不到这些数据。但如果下游的Kafka Consumer配置的isolation.level不是read_committed（默认是read_uncommitted），那么Consumer会直接越过事务隔离，把未提交的数据也拉出来消费。

结果就是：上游Flink作业还没有跑完这一轮Checkpoint，下游就已经读到了"半成品"数据。等事务提交后下游又读一遍，数据就对不上了。排查这类问题最直接的手段就是检查所有下游消费结果Topic的Consumer配置，把isolation.level设为read_committed，并且确认上游没有启用幂等写和事务写混用的情况。