Flink 1.13背压监控三指标源码深度解析

背压这个话题，在Flink社区里属于那种“人人都知道重要，但真排查起来又很虚”的东西。1.13之前，我看背压基本靠Web UI那个红黄绿图标，再配合线程栈采样，说白了就是“猜测性运维”——能看出来哪个算子卡住了，但看不出它到底忙了多久、堵了多久、闲了多久，更别想把背压量化到毫秒级。Flink 1.13这一版把这块补齐了，新增了taskmanager.busyTimeMsPerSecond、taskmanager.backPressuredTimeMsPerSecond、taskmanager.idleTimeMsPerSecond三个任务级指标，从源码层面重新设计了背压监控的数据来源和计算逻辑。这篇文章我会直接进到1.13.x源码里，把这三个指标的定义、埋点位置、计算方式、上报链路一条条拆开，结合我实际排查背压问题时踩过的坑一起写出来。

1. 为什么说1.13的背压监控是一次升级

1.1 旧机制的局限

Flink 1.12及更早版本的背压监控，核心手段是“线程栈采样”。TaskManager上会起一个采样线程，周期性抓取各个Task线程的调用栈快照，然后通过分析栈里的方法名来判断该Task目前是不是处于背压状态。经典的做法是看栈帧里是否出现了PipelinedSubpartition或者BufferListener相关的阻塞调用，命中就认为当前处于背压。

这个方案有几个先天短板：

• 只能给出“疑似背压”的比例，不是真实时间占比。采样是离散的，50ms抓一次栈，抓到一次算一次，最后统计到的百分比完全取决于采样窗口内是否恰好命中阻塞点，会漏采，也会误报。
• 无法区分背压、忙碌、空闲。旧机制只有“背压”一个结论，忙是因为在真实计算、还是在等输入、还是因为输出堵住了，全部混在一起。
• 开销不小且难以量化优化。线程栈抓取本身有性能开销，而且背压到底是200ms还是2s，对排查方向的影响非常大，旧机制完全回答不了。

1.2 新机制的核心变化

1.13引入的三指标，本质上把背压监控从“采样猜概率”变成了“埋点算时间”。实现思路是在任务执行主循环里，按真实耗时统计每个Task在每秒内分别处于忙碌、背压、空转三种状态各占多少毫秒。这三个状态互斥，并且时间分配天然满足：

busyTimeMsPerSecond + backPressuredTimeMsPerSecond + idleTimeMsPerSecond ≈ 1000

这意味着你看到某个算子backPressuredTimeMsPerSecond: 850，不用查任何额外信息，直接就知道：这一秒里，这个算子有850毫秒在等待下游释放缓冲区，计算资源和上游数据都足够，瓶颈就在下游。而旧机制最多只能告诉你“这个算子背压概率85%”，哪个更有说服力不言而喻。

这套设计的优点还不止精度。因为指标产出在Task内部，天然能做到算子粒度，每个并行子任务都有独立数值，汇总到Job级别后，Web UI可以直接画出每个算子的“忙/堵/闲”堆叠图，一眼看出整条链路瓶颈在哪。下面我直接进源码看实现。

2. 三个新指标到底怎么读

先记住指标全名，后面会反复用到：

注意，这三个指标虽挂在taskmanager前缀下，实际是Task级别的，每个subtask一份。Web UI背压选项卡里看到的百分比，就是拿数值除以10换算过来的，比如busyTimeMsPerSecond=750就对应“忙碌75%”。

拿一个简单的source → map → sink链路举例：

• 如果map算子的backPressuredTimeMsPerSecond一直在800以上，说明map计算本身没问题，数据也源源不断，但下游sink写入外部系统太慢，数据在map到sink的通道上堵住了。这时候该加的是sink的并行度或优化外部存储写入。
• 如果map算子的idleTimeMsPerSecond超过700，说明map大多数时间在等数据，瓶颈在上游source，可能是source读取源系统太慢，也可能是source并行度不足。这时候你去加map的并行度没任何意义。
• 如果map算子的busyTimeMsPerSecond接近1000，说明map的计算逻辑本身就是瓶颈，这时候才轮到考虑优化算子内计算或增加该算子并行度。

这三个指标放一起，能直接告诉你瓶颈在哪一端，而不是像以前那样只能对着采样概率猜。

3. 源码走读：指标到底是在哪一层被算出来的

3.1 指标注册入口 TaskMetricGroup

指标定义挂在org.apache.flink.runtime.taskmanager.TaskMetricGroup下。Task启动时，会通过TaskManager.getTaskManagerMetricGroup().addTask(...)创建出该Task对应的TaskMetricGroup，其中IO相关的指标被子分组TaskIOMetricGroup承接。1.13里TaskIOMetricGroup的字段大致形如：

public class TaskIOMetricGroup extends ProxyMetricGroup<TaskMetricGroup> { private final Counter numBytesIn; private final Counter numBytesOut; // 1.13 新增的三个基于时间的Gauge private final Gauge<Double> busyTimeMsPerSecond; private final Gauge<Double> backPressuredTimeMsPerSecond; private final Gauge<Double> idleTimeMsPerSecond; // ... }

注意这三个度量是Gauge<Double>而不是Counter，说明它们不是累加器，而是类似于“快照值”。实现上会周期性刷新这个快照，数值含义是“最近一秒钟内该状态的毫秒数”。

3.2 计时器实体：Task.TimerService

真正负责计时的逻辑在org.apache.flink.runtime.taskmanager.Task内部。1.13的Task里新增了一个内部类TimerService，它约莫长这样：

private class TimerService { static final long UPDATE_INTERVAL_MS = 100L; private long lastUpdateTime; // 保存自上一次刷新以来累计的各类耗时 private long busyTime; private long backPressuredTime; private long idleTime; void initialize() { lastUpdateTime = System.nanoTime(); } void update() { final long now = System.nanoTime(); final long elapsed = (now - lastUpdateTime) / 1000_000L; if (elapsed < UPDATE_INTERVAL_MS) { return; } lastUpdateTime = now; // 将耗时换算成每秒毫秒数，刷新到Gauge中 taskMetricGroup.getIOMetricGroup() .busyTimeMsPerSecond.update(scale(busyTime, elapsed)); taskMetricGroup.getIOMetricGroup() .backPressuredTimeMsPerSecond.update(scale(backPressuredTime, elapsed)); taskMetricGroup.getIOMetricGroup() .idleTimeMsPerSecond.update(scale(idleTime, elapsed)); // 重置累计值，开始下一个统计窗口 busyTime = 0; backPressuredTime = 0; idleTime = 0; } private double scale(long timeMs, long windowMs) { return timeMs * 1000.0 / windowMs; } }

这段代码解释了一个关键点：为什么叫“PerSecond”却要每100ms刷新一次。因为指标的语义是归一化到“每秒”，但统计窗口并不需要凑满1秒。每100ms采一次样，把窗口内三个状态的耗时按比例放大到1000ms，这样既能快速反映变化，又不会因为统计窗口过长导致指标跳变太剧烈。

3.3 三个状态如何判定

读源码的时候，最让人好奇的肯定是一个问题：任务到底怎么知道自己现在是“忙碌”还是“背压”还是“空闲”？答案藏在任务主循环和输入输出通道的可用性状态里。

Flink 1.13的判定逻辑可以简化为：

• 任务在处理邮箱邮件，或者在执行算子逻辑、读写数据时，计入busyTime。
• 任务尝试向输出端发送数据，但输出端的ResultPartition缓冲区池已满，需要等待下游消费腾出空间时，计入backPressuredTime。
• 任务既没有邮件要处理，也没有输入数据可读，输出端也随时可以写，完全处于等待上游数据的状态时，计入idleTime。

这三个状态互斥，每次计时埋点只会选择其中一个累加。后文会展开每个状态具体在哪个代码路径上被记上一笔。

4. 主循环里的埋点逻辑

4.1 runMailboxLoop 的骨架

Flink的Task线程启动后，进入的核心方法是runMailboxLoop()。1.13里这个方法的骨架大致如下：

private void runMailboxLoop() throws Exception { final MailboxDefaultAction defaultAction = mailboxDefaultAction; if (defaultAction == null) { return; } // 主循环 while (isRunning) { // 尝试取一条邮件，可能是检查点、取消、生命周期控制等命令 final Optional<Mail> maybeMail = mailbox.tryTakeMail(); if (maybeMail.isPresent()) { // 处理邮件 maybeMail.get().run(); } else { // 没有邮件，执行默认操作，即处理输入数据 defaultAction.runWithActionController(actionContext); } // 周期刷新耗时指标 timerService.update(); } }

这里几个关键角色需要先对号入座：

默认操作里封装了processInput()，这是读取上游数据、执行用户算子逻辑的入口。任务大部分“忙碌”时间就是在这里累计的。

4.2 默认操作与 processInput

StreamTask中，默认操作实现里调用了StreamInputProcessor.processInput()。processInput()会从inputGate.getNext()取数据，然后交给算子处理，最后写入recordWriter。这个链路里出现的耗时，计入busyTime。

但要注意，processInput()内部并不是一条直线。当上游数据源源不断时，它确实会连续处理，忙得不可开交；可一旦下游输出通道的缓冲区池满了，recordWriter.emit()就会陷入等待，此时代码实际卡在PipelinedSubpartition的缓冲区申请逻辑上。而1.13会把这个等待的耗时单独累计到backPressuredTime，而不是笼统地算进busyTime。

这部分的实现，社区在RecordWriter的emit路径上做了可用性检查。逻辑上类似：

public void emit(Record record) throws IOException { // 核心逻辑是确保当前通道有可用buffer if (!targetChannel.isAvailable()) { // 记录背压开始时间，进入等待 backPressuredTimeStartNanos = System.nanoTime(); // 等待 buffer 或者处理邮箱中更紧急的命令 waitForAvailability(); } // 真正写入数据 }

当waitForAvailability()返回时，累计backPressuredTime += (now - backPressuredTimeStartNanos)。这样就能把“被下游堵住的时间”精准剥离出来，既不影响“忙碌”的定义，也不会和“空闲”混在一起。

4.3 背压感知的邮箱循环

1.13对背压期间的行为还做了一个配套改造：背压时不再让Task线程完全死等，而是让Task在等待缓冲区的同时，保持对邮箱命令的响应能力。这一点非常重要，直观感受就是：作业卡在背压时，取消任务、触发checkpoint不再会长时间没反应。

从实现上说，TaskMailbox在背压等待期间会被要求周期性唤醒，去检查有没有新邮件进来。ActionController也承担了一部分“背压时让出执行权”的协调职责。简单讲，主循环是这么配合的：

• 缓冲区可用 → 执行默认操作处理数据
• 缓冲区不可用 → 进入背压等待，但每间隔一段时间醒来，检查邮箱里有没有取消、checkpoint等指令
• 邮箱有指令 → 优先处理指令，处理完再回到背压等待

这一轮改造让任务在恶劣背压条件下仍能快速响应管理命令，生产环境里最大的收益就是“背压不会拖垮整个作业的cancel和checkpoint”。

5. 指标从Task到Web UI的完整链路

5.1 指标上报

TimerService每100ms刷新的是Task内部Gauge快照。但这些数值要能被看到，还得经过指标系统上报。Flink的指标系统采用Reporter模式，1.13默认开启PrometheusReporter或JMXReporter时，这三个指标会周期性地发送到外部监控系统。

这里常有人问：为什么用Prometheus查flink_taskmanager_busyTimeMsPerSecond时，数值偶尔会跳到极大或接近0？这是正常的，因为Gauge刷新周期是100ms，而Prometheus抓取周期通常是15s到30s，抓到的只是“某个100ms窗口归一化后的瞬时值”。如果恰好抓到任务空转的窗口，就会看到idleTimeMsPerSecond接近1000。这不是bug，是采样粒度不同造成的观感差异。

5.2 REST API 查询

如果只想临时看一眼，不需要接监控系统，直接用REST API即可：

# 先拿JobID curl "http://localhost:8081/jobs/overview" # 拿某个Job各subtask的背压指标 curl "http://localhost:8081/jobs/<jobId>/metrics?get=taskmanager.busyTimeMsPerSecond,taskmanager.backPressuredTimeMsPerSecond,taskmanager.idleTimeMsPerSecond"

返回的是按Task ID聚合的{"id": "...", "value": "..."}列表。注意看value单位仍是毫秒/秒，要转百分比就除以10。

5.3 UI 呈现

Web UI的背压选项卡在1.13之后改成了堆叠图。每个Task显示“忙碌、背压、空闲”三段色条，色条占比直接对应 ( \frac{value}{1000} )。这个展示方式比旧版红黄绿圆点直观太多了：瓶颈在哪一层、瓶颈的时间占比多大，一眼就能扫出来。我在生产环境排障时，基本先看这张图，再决定要不要深挖具体Task的详细指标。

6. 生产环境常见问题与排查经验

6.1 看板全空或数值为0

最常见的原因是指标收集周期和刷新周期错位。有的环境启用了PrometheusReporter，但没开默认的histogram/gauges，或者Prometheus拉取正好错过刷新窗口。排查思路很简单：先在本地用REST API打点，确认Task级Gauge确实有值；确认后再查Reporter配置是否正确暴露了这几个指标。

另外注意版本差异。1.13.0刚上线时，这三个指标在全并行度较大时偶尔会出现初始值为0的情况，后续版本有修复。如果你用的是早期1.13.x，遇到指标全0先看是不是已知问题，直接升级补丁版本比调配置更省心。

6.2 背压与空闲同时为0的怪异场景

按定义，三个指标相加应约等于1000。如果排查时看到busyTimeMsPerSecond和backPressuredTimeMsPerSecond都正常，唯独idleTimeMsPerSecond接近0，说明这个算子几乎一直在满负荷运转，但没有被下游阻塞——这种情况下可以关注是否是大量小记录造成的CPU空转高，而不是真正的数据密集型计算。

还有一类极端情况：任务处于CANCELING或FAILING状态时，Task线程不再执行默认操作，三个指标都会停住不动，表现为“全0”。如果Web UI背压选项卡看到某Task全0而其他Task正常，先去确认该Task是否正处于取消或失败的状态，别被指标误导成“这个Task没在干活”。

6.3 上游抖动导致误读背压

这三个指标分Task统计，天然会受上下游关联影响。比较典型的是：某个上游TaskidleTimeMsPerSecond高，导致下游Task输入不足，下游随之出现idleTimeMsPerSecond也偏高——但这时候下游其实不是瓶颈，只是因为没数据可吃。所以正确姿势是先看全链路的堆叠图，找到那个idleTime最低而backPressuredTime最高的点，才能真正定位瓶颈。

我习惯把指标按“数据供给方 vs 数据消费方”分组：

这张表我一直贴在工位，排查时照着看基本不会跑偏。

6.4 指标口径差异

做了迁移的朋友可能还会遇到：1.13的新指标和旧版采样得到的背压百分比对不上。比如旧版UI显示某算子背压“80%”，新指标却是backPressuredTimeMsPerSecond=400。这完全正常，因为两个口径根本不同：旧版是“采样时刻处于背压阻塞的概率”，新版是“统计窗口内背压状态的真实时长占比”。对不上说明旧版高估或低估了背压，新版反而更接近真实情况。碰到对比类报表，建议直接切换口径，不建议做新旧换算。

7. 最后分享一点使用心得

用新背压监控快一年，最大体会就是排障效率提升了一个量级。以前背压是我最不愿意排查的问题，因为“猜”的成本太高，现在直接看三个指标就能锁定是上游、本算子还是下游的问题。

实际使用中还有几个小细节值得注意：

• 这三个指标只看平均值还不够，最好按并行子任务维度拆开看。我遇到过同一个算子不同subtask的backPressuredTimeMsPerSecond差异极大，一个900、一个50，这种情况多半是数据倾斜而不是下游整体慢。
• 排查checkpoint超时问题时，这三个指标也很有用。如果某个算子长期backPressuredTimeMsPerSecond很高，checkpoint barrier被堵住的概率就很大，这时候优先解决背压比调checkpoint参数更治本。
• 指标是100ms级别刷新的，做告警阈值时建议统计窗口至少取1分钟，避免瞬时抖动误报。我常用的一套告警规则是：连续5分钟backPressuredTimeMsPerSecond均值超过700就触发告警，再配合下游指标确认瓶颈。

这三个指标背后其实是Flink对背压问题的一次重新定义：从“模糊的采样概率”进化为“精确的时间账本”。源码层面的实现说穿了并不玄妙，就是主循环里精准埋点，但带来的运维范式变化非常大。1.13之后的版本里这块还在继续演进，理解了这套埋点逻辑和权衡取舍，后面再看新版本变更会轻松很多。