Flink 内存模型与网络栈：TaskManager 的内存是如何分配的？

Flink 内存模型与网络栈：TaskManager 的内存是如何分配的？

调大 taskmanager.memory.process.size 之后作业反而被 OOMKilled，堆内存明明还有富余却报了 Direct buffer memory——这类问题的根源在于一个容易被忽略的事实：你配置的内存大小，不等于 JVM 实际会使用的内存大小。Flink 的内存模型是一套分层预算 + 动态水位系统：Managed Memory 的用量由 RocksDB 的 Block Cache 命中率和 Write Buffer 冲刷频率间接决定，Network Memory 的消耗取决于“通道数 × 每通道缓冲数”，而 JVM 自身的 native 开销根本不在 Flink 的参数表里。这篇文章将拆解这套预算机制的真实行为，用 Little’s Law 和 Amdahl 定律把“调内存”从玄学变成算术——本文所有算例的数字都给出推导过程，读者可以逐行复算——并用一个真正消耗 Managed Memory 的 Scala 作业观察内存与网络指标的实际水位。

引言：配置的“值”与运行的“账”

一个真实的排障场景：TaskManager 配置了 process.size: 8g，容器内存限制 8g，运行一段时间后被 K8s OOMKilled。jmap -heap 显示堆使用率只有 40%——看起来堆内存充足，但进程总 RSS 已经逼近限制。
问题出在堆外。Flink 的堆外内存由三个互不隶属的部分组成：Managed Memory（Flink 统一管理，交给 RocksDB 或批算子）、Network Memory（网络缓冲池）、以及 JVM 自身的 native 开销（线程栈、Metaspace、DirectByteBuffer、JNI、GC 内部结构）。前两者在 Flink 的参数体系里有明确的上界；第三者只在 taskmanager.memory.process.size 里通过 JVM Overhead 预留了一个比例（默认 fraction 0.1，min 192MB，max 1GB）。一旦 JNI 调用、Netty 直接缓冲或第三方 SDK 的 native 分配超出这个预留，进程 RSS 就会突破 Flink 的预算边界，触发容器级 OOMKill——而这时堆内存可能还很空闲。
本文的路线：先把“进程的两本账”和两个理论模型立起来（第 1 章）；再逐区拆解预算机制、RocksDB 水位与网络缓冲算术（第 2 章）；然后用带 RocksDB 状态的 Scala 作业让内存与网络水位真正可见（第 3 章）；最后给出 OOM 分层诊断流程与网络预算下的并行度上限推导（第 4 章）。

一、前置知识：进程的两本账与两个理论模型

1.1 两本账：JVM 视角 vs 进程视角

排查内存问题的第一步，是承认 JVM 的账本和操作系统的账本对不上：

• JVM 的账：堆（-Xmx）、Metaspace、直接内存（-XX:MaxDirectMemorySize）、线程栈。jmap、jstat 看到的是这本账。
• 进程的账（RSS）：以上全部，加上 JVM 运行时的 native 结构——GC 内部数据（G1 的记忆集/卡表）、JIT 编译后的代码缓存、glibc 的 malloc 碎片、JNI 库（RocksDB 在关闭托管内存时）分配的 C++ 堆。pmap -x <pid>、K8s cgroup 计量的是这本账。
  容器 OOMKill 判定用的是第二本账，而 Flink 的参数体系只能精确控制第一本账的大部分——两者的差额就是 JVM Overhead 预留存在的原因。理解“两本账”是后面一切诊断的地基：堆空闲 + OOMKilled，几乎必然意味着差额出在 native 侧。

1.2 两个贯穿全文的理论模型

后面所有“该调哪个参数”的判断，都可以归结为两个模型：
模型一：Little’s Law（L = λW）。系统中的平均滞留量 = 到达速率 × 平均滞留时间。它是排队论最基本的恒等式，在这里有两个化身：

• 网络化身（BDP，带宽时延积）：通道上需要维持的 in-flight 数据量 = 吞吐 × 往返延迟。要让单通道跑 100MB/s、credit 通告 RTT 为 20ms，通道上必须常驻 ≥ 100MB/s × 20ms = 2MB 的 in-flight 数据——这就是“管道要装满”的定量表达。
• 状态化身：RocksDB 的 Write Buffer 总量 = 写入速率 × Flush 间隔的容忍值；MemTable 太小意味着 Flush 太频繁（写放大），太大意味着内存被无效占用。
  模型二：Amdahl 定律。加速比受串行部分限制；推广到资源维度——当瓶颈是资源总量（网络内存、序列化 CPU 的每字节成本）而非计算并行度时，加并行度不提升吞吐，只会稀释资源：同样的网络预算被摊到更多通道上，每通道分到的缓冲更少，单通道 BDP 需求更难满足。这解释了本文反复出现的一个反直觉现象：加并行度有时让吞吐不升反降——第 4 章会把这个“有时”量化成具体的数字。

本节小结：JVM 账与进程账（RSS）天然对不齐，差额是 native 结构——这是 JVM Overhead 预留存在的原因，也是容器 OOMKill 的主战场。Little’s Law 给出“吞吐 × 延迟 = in-flight 需求”的预算公式（网络上是 BDP，状态上是 Write Buffer），Amdahl 定律解释资源型瓶颈下加并行度为何无效。记住这两个模型，后面的所有结论都是算出来的。

二、核心剖析：分区预算、RocksDB 水位与网络缓冲算术

2.1 统一内存模型：先纠正一棵常见的错树

Flink 1.10 引入的统一内存模型（FLIP-49）将 TaskManager 进程内存划分如下：

Total Process Memory (taskmanager.memory.process.size)
├── Total Flink Memory
│ ├── JVM Heap
│ │ ├── Framework Heap (默认 128MB)
│ │ └── Task Heap (用户代码、HashMapStateBackend 的状态)
│ └── Off-Heap (受 -XX:MaxDirectMemorySize 约束)
│ ├── Managed Memory (默认 fraction 0.4)
│ ├── Network Memory (默认 fraction 0.1，min 64MB，max 1GB)
│ └── Framework Off-Heap (128MB) + Task Off-Heap (默认 0)
├── JVM Metaspace (taskmanager.memory.jvm-metaspace.size，默认 256MB) ← 与 Overhead 并列
└── JVM Overhead (fraction 0.1，min 192MB，max 1GB) ← 线程栈/GC/JIT/native

一个高频错误是把 Metaspace 画进 JVM Overhead 里——正确的关系是并列：Metaspace 有自己独立的配置项和默认值（256MB），Overhead 覆盖的是线程栈、GC 结构、JIT 代码缓存、JNI 等更难预测的部分。
各区域的职责与消耗方：

2.2 比例计算的生效规则：min/max 会“吃掉”比例

一个容易踩的配置陷阱：Network Memory 的实际值不是简单的“Flink Memory × 0.1”，而是先按比例计算，再与 min/max 取交集；若落到界外则被拉回，而拉回的差额从其他按比例计算的区域里扣。
以 Total Flink Memory = 4GB 为例：按 0.1 计算 Network = 409.6MB，落在 [64MB, 1GB] 内，取 409.6MB。如果显式把 network.min 调到 512MB，比例计算被覆盖，Network 取 512MB——多出的约 100MB 从 Managed Memory 或 Task Heap 中扣除，因为总预算不变。
这意味着：在网络密集作业中调高 network.min，Managed Memory 的份额会自动缩小，RocksDB 的 Block Cache 随之变小、读命中率下降——网络调优的代价以状态读性能的形式转移到了另一个区域。预算系统的每一次局部调整，都会在别处产生对冲。

2.3 Managed Memory：RocksDB 的水位再分配

Managed Memory 最容易误解的一点：它是一块配额，但 RocksDB 不会启动时就占满它。开启 state.backend.rocksdb.memory.managed: true 后，Flink 把配额交给 RocksDB 的共享内存管理器，再由两个比例在三个消费者之间切分：

• state.backend.rocksdb.memory.write-buffer-ratio（默认 0.5）：50% 的配额给 Write Buffer 管理器（MemTable）；
• state.backend.rocksdb.memory.high-prio-pool-ratio（默认 0.1）：Block Cache 中 10% 作为高优先池，专供索引和 Bloom Filter；
• 其余约 40% 作为数据块的 Block Cache。
  于是 Managed Memory 内部的调优就是一张再分配决策表：
  | 症状 | 指向的失衡 | 调整方向 |
  |—|—|—|
  | 频繁 Flush、Write Stall、写放大 | Write Buffer 不足 | writebuffer.size 64→128/256MB、write-buffer-ratio 上调 |
  | Block Cache 命中率 < 90%、点查变慢 | 数据缓存不足 | write-buffer-ratio 下调（向 Cache 倾斜）、必要时上调 managed.fraction |
  | 索引/过滤器未命中 | 高优先池不足 | high-prio-pool-ratio 上调（默认 0.1） |
  一个典型的调优路径（推演示例，假设 Checkpoint 间隔为分钟级；实际收益取决于间隔设置与状态后端配置）：
  | 指标 | 调整前（ratio 0.5, wb.size 64MB） | 调整后（ratio 0.3, wb.size 256MB） |
  |—|—|—|
  | Flush 频率（每列族） | 每分钟多次，偶发 Write Stall | 数分钟一次，Stall 消失 |
  | Block Cache 命中率 | 85% 左右，点查延迟抖动 | 回到 92–95% |
  | Checkpoint 时长 | 120s 量级，偶发超时 | 回落至 40s 量级 |
  关键动作不是“调大内存”，而是在配额内部把水位从写路径挪到读路径——总量不变，结构变了。这个再分配视角比“加内存”便宜得多，也常被跳过。

2.4 JVM Overhead：把“账外账”变成“估算账”

与其列举 native 开销的种类，不如给一个可执行的估算法：

Overhead 需求 ≈ 线程数 × 默认栈大小(约1MB/线程, Linux x64 JVM 默认)
+ Metaspace 典型值(100–300MB, 与类加载数量相关)
+ GC 结构(G1 常见 5%–10% 的堆)
+ JIT 代码缓存(默认上限 240MB, 实际常见 64–128MB)
+ 用户代码的 Direct/Native(Netty 池、JNI SDK、未托管的 RocksDB)

算例：4 slot 的 TM，每 subtask 约 5 线程（业务 + 定时器 + 网络）→ 20 线程；框架线程约 30 → 合计约 50 线程 × 1MB = 50MB；Metaspace 按 256MB；GC/JIT 约 150MB；合计约 456MB，加上 Netty Direct 与杂项约 500MB。对照预算：process.size = 2g 时 Overhead = 0.1 × 2g = 204MB < 500MB → 必然超支，RSS 突破容器限制只是时间问题；process.size = 8g 时 Overhead = 819MB > 500MB，安全。同样的作业，小 TM 被 OOMKill 而大 TM 相安无事，根源常常不是内存不够，而是 Overhead 的绝对值下限被小比例压缩了——这也是 jvm-overhead.min 存在的意义。
对账与诊断命令：

kubectl top pod <pod> # 容器 RSS（第二本账）
jcmd <pid> VM.native_memory summary # NMT 视角（需 -XX:NativeMemoryTracking=summary）
pmap -x <pid> | tail -1 # 进程 RSS 与 anon 段汇总

2.5 网络栈：Credit 流控、缓冲生命周期与通道数算术

Credit 流控的本质是把反压粒度从“连接级”下沉到“通道级”：接收方按通道通告可用缓冲数（1 buffer = 1 credit），发送方每发一个 buffer 消耗一个 credit，credit 归零时只停该通道，不影响同一 TCP 连接上的其他通道。支撑 credit 的两类缓冲：

• Exclusive buffers（独占）：每个远程输入通道初始化即持有，由 taskmanager.network.memory.buffers-per-channel 控制，默认 2。作用是“保底”——在 credit 通告往返期间维持基本吞吐；
• Floating buffers（浮动）：每个输入门（gate）共享，由 floating-buffers-per-gate 控制，默认 8，用于吸收突发。
  缓冲生命周期：从发送方的输出池申请 → 经网络（或本地直通）到达接收方 → 被消费完最后一个记录后释放 → credit 通告回流。任何一步变慢，缓冲就在那一步堆积——这就是 outPoolUsage / inPoolUsage 两个指标能定位问题的原理。
  Little’s Law 落到网络上（BDP 算例）：RTT 20ms、独占缓冲 2 × 32KB = 64KB，则单通道 credit 周期内的吞吐上限 ≈ 64KB / 20ms = 3.2MB/s。要让单通道跑 100MB/s，需要 in-flight ≥ 100MB/s × 20ms = 2MB，即约 64 个 buffer/通道——默认的 2 个差了 32 倍。这就是“管道没装满”的定量含义。
  通道数算术：all-to-all 连接（KeyBy/Rebalance）下，每个下游 subtask 的输入通道数 = 上游并行度。设 TM 上有 S 个 subtask（slot 数）、每个 subtask 一个输入 gate、上游并行度 p，则：

每 TM 的 Network 需求 ≈ S × (p × buffers-per-channel + floating-buffers-per-gate) × 32KB

注意公式结构：bpc 那一项乘 p，floating 那一项不乘——这个区别在 4.2 节解不等式时至关重要。
算例（默认缓冲，p = 200，S = 8）：每 gate (200 × 2 + 8) = 408 buffers × 32KB ≈ 13MB；8 个 gate 合计 ≈ 104MB——在 1GB 上限内很宽裕。但若为了满足 BDP 把 buffers-per-channel 提到 64：每 gate (200 × 64 + 8) = 12,808 buffers × 32KB ≈ 400MB，8 个 gate 合计 约 3.2GB，直接爆掉 1GB 上限——缓冲池被压缩，每个通道反而分不到足够 credit，吞吐不升反降。BDP 调优与内存预算是一对需要联立求解的约束：调大每通道缓冲的同时，必须核算总量并同步上调 network.fraction/max。
指标判读矩阵（注意：inPoolUsage 指该任务自身输入侧、outPoolUsage 指其输出侧；远程输入通道才计入 inPoolUsage，local 通道被忽略）：

2.6 隐形乘数：序列化与数据交换模式

序列化的乘数效应：每记录序列化字节数 × 吞吐 = 网络带宽占用；序列化体积同时放大 Network 的 in-flight 字节与 RocksDB 的 SST 体积。一个序列化选择错误，会同时表现为 CPU 打满（序列化开销）、网络缓冲更快耗尽、RocksDB 写放大上升——三处症状，一个根因。
性能阶梯（相对关系来自官方基准，绝对值取决于对象结构，勿直接引用为通用数字）：

Class … cannot be used as POJO type because not all fields are valid POJO fields,
and must be processed as GenericType.

也可以用 env.getConfig().disableGenericTypes() 让回退直接报错，把静默降级变成显式失败。
数据交换模式的连接数算术：四种模式对 Network 的压力由物理连接数决定——

本节小结：内存模型是分层预算，Metaspace 与 JVM Overhead 并列而非嵌套；min/max 会覆盖比例并挤压其他区域；Managed Memory 的 0.4 是流批通用的平衡默认值，分场景应下探 0.1–0.2 或上探 0.6–0.7。Managed 内部调优的本质是“再分配决策表”——命中率低向 Cache 倾斜、Flush 频繁增大 Write Buffer，而非一味加总量。Overhead 可以用“线程数 × 栈 + Metaspace + GC/JIT”估算，小 TM 的 OOMKill 常源于比例下限被压缩。网络侧，Little’s Law 给出 BDP 需求（吞吐 × RTT），它与 Network 预算（bpc 乘 p、floating 不乘）需要联立求解；序列化与交换模式是两个隐形乘数，Scala case class 走专用序列化器而非 Kryo——真正的回退风险在普通 class 与无类型信息的 Row。

三、手把手实操：让内存与网络水位真正可见

上一篇讲反压时我们学过“指标组合判读”；本节把同一套方法论落到内存上——前提是作业必须真正消耗 Managed Memory：下面的代码启用了 RocksDB 后端，且用恒等 key 让状态无限增长、每键写入 4KB，驱动 MemTable 与 Block Cache 真实爬升。

3.1 环境与依赖

JDK 11，Flink 1.18.1，本地带 WebUI 的 mini cluster，Scala 2.12：

<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-scala_2.12</artifactId>
<version>1.18.1</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients</artifactId>
<version>1.18.1</version>
</dependency>
<!– RocksDB 状态后端：Managed Memory 的真正消费者 –>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-statebackend-rocksdb</artifactId>
<version>1.18.1</version>
</dependency>
</dependencies>

Java API 等价写法：将 flink-streaming-scala_2.12 换成 flink-streaming-java，object/def main 换成 public class/static void main，其余 API 调用同名。

3.2 作业：RocksDB 状态 + KeyBy Shuffle + 可调反压

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.{Configuration, RestOptions, TaskManagerOptions}
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.api.functions.sink.SinkFunction
import org.apache.flink.streaming.api.functions.source.SourceFunction
object MemoryNetworkLab {
def main(args: Array[String]): Unit = {
// map 每条记录的 sleep 毫秒数：>0 制造反压观察网络指标；=0 放开吞吐观察内存爬升
val mapSleepMs = if (args.length > 0) args(0).toLong else 5L
val conf = new Configuration()
conf.set(RestOptions.BIND_PORT, "8081")
conf.setInteger(TaskManagerOptions.NUM_TASK_SLOTS, 4)
val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf)
env.setParallelism(4)
env.setStateBackend(new EmbeddedRocksDBStateBackend()) // 关键：启用 RocksDB
env.enableCheckpointing(10_000)
env.getCheckpointConfig.setCheckpointStorage("file:///tmp/mem-net-lab-ckpt")
env.addSource(new RateSource(1L))
.name("source").setParallelism(1)
.keyBy(v => v) // 恒等 key：每条记录一个新键，状态无限增长
.map(new StatefulMap(mapSleepMs)).name("stateful-map").setParallelism(4)
.addSink(new NoopSink).name("sink").setParallelism(4)
// 注意：keyBy 已天然断链（chaining 仅对 forward 成立），无需也不应在 sink 上再 disableChaining
env.execute("memory-network-lab")
}
}
class RateSource(intervalMs: Long) extends SourceFunction[Long] {
@volatile private var running = true
override def run(ctx: SourceFunction.SourceContext[Long]): Unit = {
var i = 0L
while (running) { ctx.collect(i); i += 1; Thread.sleep(intervalMs) }
}
override def cancel(): Unit = running = false
}
/** 每键写入 4KB，以可控速率驱动 RocksDB 的 MemTable 与 Block Cache 增长 */
class StatefulMap(sleepMs: Long) extends RichMapFunction[Long, Long] {
@transient private var state: ValueState[Array[Byte]] = _
override def open(parameters: Configuration): Unit = {
state = getRuntimeContext.getState(
new ValueStateDescriptor[Array[Byte]]("payload", classOf[Array[Byte]]))
}
override def map(value: Long): Long = {
if (sleepMs > 0) Thread.sleep(sleepMs)
state.update(new Array[Byte](4 * 1024))
value
}
}
class NoopSink extends SinkFunction[Long] {
override def invoke(value: Long, context: SinkFunction.Context): Unit = ()
}

设计意图：sleepMs=5 时 map 每 subtask 容量约 200 条/s、四个 subtask 合计 800 条/s < Source 供给 1000 条/s → Source 被反压，outPoolUsage 飙高、map 的 busyTime 接近 1000（sleep 计入 busy，正如反压篇所学）；同时状态仍以约 800 × 4KB ≈ 3.2MB/s 增长，两三分钟后 RocksDB 状态达到数百 MB，Managed.Used 明显爬升。sleepMs=0 时吞吐放开，内存爬升更快。

3.3 观测命令与预期数值

任务级指标（反压与输出池）：

JOB=$(curl -s http://localhost:8081/jobs/overview | jq -r '.jobs[0].jid')
VID_SRC=$(curl -s http://localhost:8081/jobs/$JOB | jq -r '.vertices[] | select(.name|startswith("source")) | .id')
# Source 的输出池与吞吐
curl -s "http://localhost:8081/jobs/$JOB/vertices/$VID_SRC/metrics?\\
get=outPoolUsage,numBytesOutPerSecond,numBuffersOutPerSecond"
VID_MAP=$(curl -s http://localhost:8081/jobs/$JOB | jq -r '.vertices[] | select(.name|startswith("stateful-map")) | .id')
curl -s "http://localhost:8081/jobs/$JOB/vertices/$VID_MAP/metrics?\\
get=busyTimeMsPerSecond,idleTimeMsPerSecond"

TM 级内存指标（本节的主角）：

TM=$(curl -s http://localhost:8081/taskmanagers | jq -r '.taskmanagers[0].id')
curl -s "http://localhost:8081/taskmanagers/$TM/metrics?\\
get=Status.JVM.Memory.Managed.Used,Status.JVM.Memory.Heap.Used,Status.JVM.Memory.Metaspace.Used"

预期数值（sleepMs=5，运行 2–3 分钟后）：

# 变体：双 TM 观察 inPoolUsage
# 0) 先确认本机可用内存 ≥ 2 × process.size——双 TM 会按配置各占一份完整预算
# 1) 在 $FLINK_HOME/conf/workers 中写两行 localhost
# （TM 的 rpc.port 与 data.port 默认为 0 = 随机，同机双启不冲突）
# 2) $FLINK_HOME/bin/start-cluster.sh
# 3) $FLINK_HOME/bin/flink run -c MemoryNetworkLab your-jar.jar 5
# keyBy 通道此时含远程成分，接收方的 inPoolUsage 才有判读价值

3.4 常见错误与调试

本节小结：让水位可见的前提是让资源真正被消耗——启用 RocksDB、用增长型 key 驱动状态写入，Managed.Used 才会爬升；用可调 sleep 制造反压，outPoolUsage 才有读数。观测要同时覆盖任务级（outPoolUsage/busyTime）与 TM 级（Managed/Heap/Metaspace）两层；输出速率的估算要区分“按记录计费的本体”与“按缓冲计费的协议封装”——反压下分摊最差，取区间而非单点。最大的陷阱是 local 通道使 inPoolUsage 失去信息量——判读时永远以发送方的 outPoolUsage 为准，或换双 TM 环境（并先核算本机内存）。

四、进阶思考：OOM 分层诊断与并行度上限

4.1 OOM 的分层诊断流程

拿到一个 OOM，先分类再动手，不要先加总内存——异常类型不同，加内存可能完全无效：

4.2 网络预算下的并行度上限：把公式解出来

把 2.5 节的需求公式倒过来，可以推导给定网络预算下的并行度上限。先把约定写清楚：Budget = Network 配额，S = 每 TM 的 slot 数，bpc = buffers-per-channel（随 p 线性放大），fpg = floating-buffers-per-gate（不随 p 放大），并假设每个 subtask 只有一个输入 gate：

约束： S × (p × bpc + fpg) × 32KB ≤ Budget
求解 p： p × bpc ≤ Budget / (S × 32KB) − fpg
p ≤ (Budget / (S × 32KB) − fpg) / bpc

注意求解步骤：fpg 在括号内但不乘 p，所以它只能从“除以 S × 32KB 之后”的项里减掉，不能与 bpc 合并成一个 (bpc + fpg) 整体做分母——合并会导致 fpg 被错误地乘上 p，结果差出数倍。
算例（Budget = 512MB，S = 8，bpc = 2，fpg = 8）：

512MB / (8 × 32KB) = 536,870,912 / 262,144 = 2048 buffers
(2048 − 8) / 2 = 1020 → p_max ≈ 1020

看起来宽裕，但两个因素会快速击穿它：

• 因素一：为满足 BDP 调大 bpc。把 bpc 提到 64（2.5 节的 BDP 推导）后：(2048 − 8) / 64 ≈ 32——单 TM 可支撑的上游并行度从 1020 骤降到 32。这正是“BDP 调优与内存预算是联立约束”的定量表达：你要么扩 Network 预算，要么接受低并行度，两者不可兼得。
• 因素二：并行度本身超限。p = 2000 已经超过 p_max = 1020，这里算它正是为了展示超限的后果：需求 = 8 × (2000 × 2 + 8) × 32KB ≈ 1.05GB > 512MB，缓冲池被压缩，每通道分不到足够 credit——加并行度不提升吞吐，反而拉低单通道的 BDP 满足度。
• 因素三：gate 数比并行度更隐蔽。如果作业含 join——每个 subtask 有两个输入 gate，等效于 S 翻倍：p_max = (512MB / (8 × 2 × 32KB) − 8) / 2 = (1024 − 8) / 2 = 508，直接减半。加一个 join 不会报错、不会提示，但你的网络预算上限已经悄悄砍半——这是三条因素中最容易被忽略的一条。
  这就是 Amdahl 定律资源版的预言：资源型瓶颈下，并行度收益递减为负。解法不是继续加并行度，而是联立求解：扩 Network 预算、降每通道缓冲、或用 Rescale 换掉 Rebalance 削减连接数（100→200 时连接数增长 2 倍而非 4 倍）。

4.3 一次只改一个变量

内存调优的验证纪律：一次只改一个参数，用第 3 章的 REST 命令记录改前改后的水位（Managed.Used 的稳态、outPoolUsage 的分布、Checkpoint 时长）。如果调整生效，指标应按 2.3/2.5 节的预测方向移动；如果不动，说明瓶颈判断错了——回到分层诊断重新分类，而不是继续叠参数。

本节小结：OOM 诊断的顺序是“异常类型 → 参数 → native 泄漏”，容器级 Kill 属于第二本账，加堆无用；网络缓冲不足按代价递增处置——先扩预算、再降 bpc、最后才降并行度。网络预算给出并行度硬上限，求解时 bpc 与 fpg 的乘法结构不可混淆——默认配置下 p_max ≈ 1020，但把 bpc 提到 64 满足 BDP 后骤降至 32，含 join 的双 gate 作业再减半至 508。加并行度在资源瓶颈下收益为负，所有调整都要以指标水位的前后对比来验证。

五、总结：内存调优是预算分配，不是容量堆叠

回到最初的问题：TaskManager 的内存是如何分配的？
内存层面：Flink 把进程切成分层预算——Metaspace 与 JVM Overhead 位于 Flink 内存体系之外、相互并列；Managed Memory 是交给 RocksDB 的动态配额，内部再按 write-buffer-ratio 与 high-prio-pool-ratio 在写路径与读路径之间再分配；Network Memory 由“通道数 × 每通道缓冲数”消耗；min/max 会覆盖比例并挤压相邻区域；Overhead 可以用“线程数 × 栈 + Metaspace + GC/JIT”提前估算，是容器 OOMKill 的主战场。
网络与序列化层面：Credit 流控把反压粒度下沉到通道级，代价是 Network 需求随通道数（笛卡尔积）增长；Little’s Law 给出单通道的 BDP 需求（吞吐 × RTT），它与 Network 预算需要联立求解；序列化与交换模式是两个隐形乘数，Scala case class 走专用序列化器而非 Kryo，真正的回退风险在普通 class 与无类型信息的 Row——用日志和 disableGenericTypes() 让回退显式化。
如果只带走三句话：

本节小结：内存问题的本质是预算分配问题——五区预算各有消费者，RocksDB 在配额内自行再分配水位，网络预算与 BDP 需求联立决定并行度上限。带着“两本账 + 两个模型 + 一张再分配决策表”去排障，内存调优就从经验玄学变成了可推导的算术——而算术要可信，唯一的办法是把每一步推导摊开供读者复算，本文如此，你的排障记录也应如此。