一文对最新版本 Flink 反压机制全景深度解析(附源码)

1. 反压形成的根本原因

1.1 反压的本质

数据生产速度 > 数据消费速度 ↓
缓冲区填满↓
生产者被阻塞↓
反压形成并向上游传播

1.2 反压产生的典型场景

场景1: 算子处理能力不足
Source (1000 records/s) → Map (1000 r/s) → Window (100 r/s) ← 瓶颈！↓反压产生场景2: 外部系统慢
Kafka Source → Transform → **Sink to DB** ← DB响应慢↓反压传播到Source场景3: 数据倾斜
Key "A": 90% 数据 ← 过载！
Key "B": 5% 数据
Key "C": 5% 数据场景4: GC 压力
大状态 + 频繁GC → Task线程暂停 → 反压

2. Buffer 管理体系（反压的物理基础）

2.1 三层 Buffer 架构

┌─────────────────────────────────────────────────────────────┐
│              Layer 1: NetworkBufferPool                      │
│              (全局共享 Buffer 池)                             │
├─────────────────────────────────────────────────────────────┤
│  • 固定大小的 MemorySegment 集合                              │
│  • 由 TaskManager 启动时分配                                  │
│  • 默认 32KB per segment                                     │
│  • 动态在 LocalBufferPool 之间重分配                          │
└─────────────────────────────────────────────────────────────┘↓ 分配给                   ↓ 分配给
┌──────────────────────────┐  ┌──────────────────────────┐
│  Layer 2: LocalBufferPool │  │  LocalBufferPool         │
│  (InputGate 专属)         │  │  (ResultPartition 专属)   │
├──────────────────────────┤  ├──────────────────────────┤
│  • 每个 InputGate 一个    │  │  • 每个 ResultPartition  │
│  • 可动态调整大小          │  │    一个                   │
│  • 支持 overdr专用
│  • Exclusive Buffers     │  │                           │
│    (专用于某个 channel)   │  │                           │
│  • Floating Buffers      │  │                           │
│    (共享于所有 channels)  │  │                           │
└──────────────────────────┘  └──────────────────────────┘

2.2 NetworkBufferPool 核心源码

// flink-runtime/.../network/buffer/NetworkBufferPool.javapublic class NetworkBufferPool implements BufferPoolFactory {// 全局 Buffer 总数private final int totalNumberOfMemorySegments;// 单个 Segment 大小（默认 32KB）private final int memorySegmentSize;// 可用的内存段队列private final ArrayDeque<MemorySegment> availableMemorySegments;// 所有 LocalBufferPoolprivate final Set<LocalBufferPool> allBufferPools = new HashSet<>();/*** 请求 Buffer（阻塞式）* 关键：如果没有可用 Buffer，线程会在这里阻塞！*/private List<MemorySegment> internalRequestMemorySegments(int numberOfSegmentsToRequest) throws IOException {final List<MemorySegment> segments = new ArrayList<>(numberOfSegmentsToRequest);final Deadline deadline = Deadline.fromNow(requestSegmentsTimeout);while (true) {if (isDestroyed) {throw new IllegalStateException("Buffer pool is destroyed.");}MemorySegment segment;synchronized (availableMemorySegments) {// 🔥 尝试获取 Segmentsegment = internalRequestMemorySegment();if (segment == null) {// ⏳ 没有可用 Buffer，等待 2 秒availableMemorySegments.wait(2000);}}if (segment != null) {segments.add(segment);}if (segments.size() >= numberOfSegmentsToRequest) {break;  // 请求满足}if (!deadline.hasTimeLeft()) {// ❌ 超时抛异常throw new IOException("Timeout requesting buffers");}}return segments;}
}

2.3 LocalBufferPool - 动态内存管理

// flink-runtime/.../network/buffer/LocalBufferPool.javapublic class LocalBufferPool implements BufferPool {// 当前池子可用的 Bufferprivate final ArrayDeque<MemorySegment> availableMemorySegments;// 每个 subpartition 的 Buffer 计数private final int[] subpartitionBuffersCount;// 最大 Buffer 数（上限）private int maxUsedBuffers;/*** 🔥 阻塞式请求 Buffer* Task 线程会在这里被阻塞，形成反压！*/private MemorySegment requestMemorySegmentBlocking(int targetChannel)throws InterruptedException {MemorySegment segment;// 循环等待直到有 Buffer 可用while ((segment = requestMemorySegment(targetChannel)) == null) {try {// ⏳ 等待 Buffer 可用的 FuturegetAvailableFuture().get();  // 阻塞点！} catch (ExecutionException e) {LOG.error("The available future is completed exceptionally.", e);ExceptionUtils.rethrow(e);}}return segment;}/*** 非阻塞式请求*/@Nullableprivate MemorySegment requestMemorySegment(int targetChannel) {MemorySegment segment = null;synchronized (availableMemorySegments) {checkDestroyed();// 1. 优先从本地池子取if (!availableMemorySegments.isEmpty()) {segment = availableMemorySegments.poll();}// 2. 如果达到上限，尝试获取 Overdraft Bufferelse if (isRequestedSizeReached()) {segment = requestOverdraftMemorySegmentFromGlobal();}if (segment == null) {return null;  // 🚫 没有可用 Buffer}// 更新 per-channel 计数if (targetChannel != UNKNOWN_CHANNEL) {subpartitionBuffersCount[targetChannel]++;// 达到per-channel上限if (subpartitionBuffersCount[targetChannel] == maxBuffersPerChannel) {unavailableSubpartitionsCount++;}}checkAndUpdateAvailability();}return segment;}
}