【SkyWalking从入门到精通】第26篇：Buffer源码深度解析——无锁环形缓冲区的精巧设计

<–
title: “Buffer源码深度解析——无锁环形缓冲区的精巧设计”
series: “Apache SkyWalking实战全解析”
episode: 26
publish_date: “2026-07-02”
author: “技术博客作者”
tags: [“SkyWalking”, “Buffer”, “无锁编程”, “环形缓冲区”, “AtomicInteger”, “并发编程”, “源码分析”]
description: “深度剖析SkyWalking DataCarrier中Buffer的源码实现，包括环形数组数据结构、AtomicInteger CAS无锁写入、OVERWRITE与BLOCKING两种写满策略的代码细节，以及高并发场景下的线程安全保证机制。”
prev_article: “文章025_SkyWalking的内存消息队列DataCarrier.md”
next_article: “文章027_Channel源码解析.md”
–>

上一篇【第25篇】SkyWalking的内存消息队列DataCarrier——高性能数据中转站
下一篇【第27篇】Channel源码解析——多Buffer协同的分区艺术

1. 引言：数组+原子计数器=无锁队列？

上篇文章我们宏观介绍了DataCarrier的三层架构。这篇文章我们钻进Buffer的"五脏六腑"，看看它是如何用最简单的数据结构——数组和原子计数器——实现无锁、零GC、线程安全的消息缓冲。

先抛出一个灵魂拷问：用ArrayList能不能实现一个线程安全的队列？答案是不能——因为add()方法在扩容时会创建新数组并复制元素,既不线程安全也会产生GC。那怎么做到线程安全又零GC？答案是一个看似简单的组合：定长数组 + AtomicInteger。

2. Buffer的数据结构

2.1 核心字段

publicclassBuffer<T>{// 底层存储：定长Object数组privatefinalObject[]buffer;// 写入游标：AtomicInteger保证原子性privatefinalAtomicIntegerindex;// 数组容量（必须是2的幂，方便位运算取模）privatefinalintbufferSize;// 最大索引值 = bufferSize - 1privatefinalintmaxIndex;// 写满策略privatefinalintstrategy;publicstaticfinalintOVERWRITE=0;publicstaticfinalintBLOCKING=1;}

2.2 为什么bufferSize必须是2的幂？

// 构造函数中的校验代码Buffer(intbufferSize,intstrategy){this.bufferSize=bufferSize;this.maxIndex=bufferSize-1;this.buffer=newObject[bufferSize];this.index=newAtomicInteger(0);this.strategy=strategy;// 容量必须是2的幂！// 如果不是，Buffer不会报错，但性能会大幅下降// 因为取模运算 index % bufferSize 可以用更快的位运算 index & (bufferSize - 1)}

原因很简单：i & (n-1)比i % n快得多，尤其在高频调用场景下。假设bufferSize=1024，取模时：

• index % 1024需要一次除法运算（CPU的除法比加法慢10-30倍）
• index & 1023只需要一次按位与运算（1个CPU周期）

高频路径上每省一点，累积起来就是可观的性能提升。

2.3 内存布局可视化

+------------------------------------------------------------------+ | Buffer 内存布局 (bufferSize=8) | +------------------------------------------------------------------+ | | | buffer 数组: | | Index: 0 1 2 3 4 5 6 7 | | ┌──────┬──────┬──────┬──────┬──────┬──────┬──────┬──────┐| | │ T obj│ T obj│ T obj│ T obj│ T obj│ T obj│ T obj│ T obj│| | │ 预分配│ 预分配│ 预分配│ 预分配│ 预分配│ 预分配│ 预分配│ 预分配│| | └──────┴──────┴──────┴──────┴──────┴──────┴──────┴──────┘| | | | index (AtomicInteger): 当前值 = 3 → 下一个写入位置 = 3 & 7 = 3 | | | | 环形特性: | | 当index到达8时，8 & 7 = 0，自动回到数组开头 | | | | ┌───┬───┬───┬───┬───┬───┬───┬───┐ | | │ 0 │ 1 │ 2 │ 3 │ 4 │ 5 │ 6 │ 7 │ | | └───┴───┴───┴───┴───┴───┴───┴───┘ | | ↑ ↑ | | └── N+1次写入回到这里 ──┘ ← 第N次写入 | +------------------------------------------------------------------+

3. save()方法——无锁写入的核心

3.1 源码分析

publicbooleansave(Tdata){// 第1步：CAS自增获取写入位置inti=index.getAndIncrement();if(bufferSize!=0){// 第2步：环形映射（位运算取模）intslot=i&maxIndex;// 第3步：直接写入（无锁！）buffer[slot]=data;returntrue;}returnfalse;}

3.2 为什么这能保证线程安全？

关键在于**AtomicInteger.getAndIncrement()**的CAS（Compare-And-Swap）机制：

+------------------------------------------------------------------+ | 无锁写入的并发保证 | +------------------------------------------------------------------+ | | | 时间线: | | ──────────────────────────────────────────────────────────────→ | | | | Thread-1: getAndIncrement() → 获取位置 3 | | Thread-2: getAndIncrement() → 获取位置 4 (原子操作,不会冲突) | | Thread-3: getAndIncrement() → 获取位置 5 | | | | 然后各自写入: | | Thread-1: buffer[3] = data1 (位置3是Thread-1独占的) | | Thread-2: buffer[4] = data2 (位置4是Thread-2独占的) | | Thread-3: buffer[5] = data3 (位置5是Thread-3独占的) | | | | 关键: 每个线程拿到的是互不相同的index值 | | 同一个位置不可能被两个线程同时写入 | | 因此不需要锁！ | +------------------------------------------------------------------+

3.3 这个过程像什么？

有点像食堂打饭——每个窗口前面是一条队伍（不对，DataCarrier是反过来）：

想象一个旋转寿司转盘。寿司师傅（生产者线程们）每个人拿一个号码牌（getAndIncrement()返回值），然后把寿司放到对应编号的盘子里（buffer[slot] = data）。因为每个号码牌是唯一的，不同师傅不会把寿司放到同一个盘子里。不需要排队，不需要协调——号码牌系统自动帮他们做了分配。

3.4 一个潜在问题：生产-消费速度不匹配

如果生产者速度远大于消费者，index会不断增长。Buffer虽然是环形的，但覆盖写入会导致消费者还没读，数据就被覆盖了。

这就引出了Buffer的两种写满策略。

4. OVERWRITE和BLOCKING两种策略

4.1 OVERWRITE策略（覆盖模式）

// OVERWRITE策略的简化逻辑publicbooleansave(Tdata){inti=index.getAndIncrement();intslot=i&maxIndex;// 不检查是否已被消费// 直接覆盖！旧数据丢失buffer[slot]=data;returntrue;}

使用场景：你最关心的是"最新数据"，不在乎旧数据丢失。比如CPU使用率、内存使用率这类指标——丢了上一秒的数据无所谓，用户想看的是现在的状态。

类比：监控摄像头覆盖录像——磁带（环形缓冲区）转满一圈后，旧的录像自动被新的覆盖。你不会需要看7天前的监控，但你需要看到刚才发生的。

4.2 BLOCKING策略（阻塞模式）

// BLOCKING策略的简化逻辑publicbooleansave(Tdata){// 获取当前写入位置inti=index.get();// 计算与消费者位置的差距intconsumerIndex=consumer.getConsumerIndex();// 如果生产者领先消费者太多，拒绝写入if(i-consumerIndex>=bufferSize){// 这里不真的阻塞，而是返回false// 调用方自行决定重试或丢弃returnfalse;}// 正常写入i=index.getAndIncrement();intslot=i&maxIndex;buffer[slot]=data;returntrue;}

使用场景：你关心数据的"完整性"，每一条都不能丢。比如Trace数据——丢了某个Span，整个Trace链就断了。

类比：快递驿站——如果货架（缓冲区）满了，新快递就不能放进去（不能覆盖别人的快递），需要等消费者取走一部分腾出空间。

4.3 两种策略的对比图

+------------------------------------------------------------------+ | OVERWRITE vs BLOCKING | +------------------------------------------------------------------+ | | | OVERWRITE 模式: | | ┌─────────────────────────────────────────────────────┐ | | │ │ | | │ 消费者慢 → 转盘照转 → 旧寿司被覆盖 → 只有最新的 │ | | │ │ | | │ [已消费] [已消费] [待消费] [待消费] [写入中] ←覆盖 │ | | │ ↑ ↑ │ | | │ consumerCursor writeCursor │ | | └─────────────────────────────────────────────────────┘ | | | | BLOCKING 模式: | | ┌─────────────────────────────────────────────────────┐ | | │ │ | | │ 消费者慢 → 队列满 → 拒绝写入 → 调用方处理 │ | | │ │ | | │ [已消费] [待消费] [待消费] [待消费] [已满] ←阻塞！ │ | | │ ↑ ↑ │ | | │ consumerCursor writeCursor │ | | └─────────────────────────────────────────────────────┘ | +------------------------------------------------------------------+

5. drain()方法——批量消费的艺术

5.1 源码分析

// Consumer接口publicinterfaceIConsumer<T>{voidconsume(List<T>data);}// Buffer中的drain方法（简化版）publicvoiddrain(IConsumer<T>consumer){// 获取当前生产者的位置intwriteIndex=index.get();// 计算可消费的数量intavailable=writeIndex-consumerIndex;if(available<=0){return;// 没有新数据}// 批量读取List<T>consumedList=newArrayList<>(available);for(inti=consumerIndex;i<writeIndex;i++){intslot=i&maxIndex;Tdata=(T)buffer[slot];if(data!=null){consumedList.add(data);buffer[slot]=null;// 清空引用，帮助GC}}// 更新消费位置consumerIndex=writeIndex;// 批量回调消费者if(!consumedList.isEmpty()){consumer.consume(consumedList);}}

5.2 批量的价值

+------------------------------------------------------------------+ | 逐条消费 vs 批量消费 | +------------------------------------------------------------------+ | | | 逐条消费: | | ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ | | │读取1条 │→│回调consumer│→│读取1条 │→│回调consumer│ ... │ | └──────────┘ └──────────┘ └──────────┘ └──────────┘ | | 每次消费都要: 数组访问 + 方法调用 + context switch | | | | 批量消费: | | ┌──────────────────────────────────────────────┐ | | │ 读取100条 → 一次回调consumer(100条数据) │ | | └──────────────────────────────────────────────┘ | | 100倍减少: 方法调用 + gRPC网络请求 | | | | 尤其是gRPC场景，批量发送可以极大减少网络往返次数 | +------------------------------------------------------------------+

6. 线程安全保证机制深度分析

6.1 CAS的乐观锁哲学

// AtomicInteger.getAndIncrement()的底层原理// Unsafe类实现（简化版）publicfinalintgetAndIncrement(){returnunsafe.getAndAddInt(this,valueOffset,1);}// Unsafe.getAndAddInt (JDK8+)publicfinalintgetAndAddInt(Objecto,longoffset,intdelta){intv;do{v=getIntVolatile(o,offset);// CAS循环：不断尝试，直到成功}while(!compareAndSwapInt(o,offset,v,v+delta));returnv;}

CAS的核心思想是乐观的——它假设大多数时候不会有冲突，直接尝试操作，如果发现冲突了就重试。这就像在图书馆占座：

• 悲观锁（synchronized）：进门先锁门，一次只让一个人进来找座位
• CAS（乐观锁）：大家都进去找座位，如果发现座位被占了就换个座位

在低到中等并发场景下，CAS的性能远优于锁。DataCarrier就是利用了这个特性。

6.2 一个微妙的问题：指令重排序

有一个潜在的坑：Java编译器和CPU可能会对指令进行重排序。也就是说：

// 你写的顺序:inti=index.getAndIncrement();// ①intslot=i&maxIndex;// ②buffer[slot]=data;// ③// 可能被重排为:inti=index.getAndIncrement();// ① (不能重排，有内存屏障)buffer[slot]=data;// ③ (可能先执行？)intslot=i&maxIndex;// ② (可能后执行？)

AtomicInteger的getAndIncrement()本身保证了内存可见性（相当于一个volatile写+读），但buffer[slot] = data这个普通写操作没有volatile语义。

不过在实际场景中，这不会导致问题，因为：

1. 每个线程写的是不同的slot（因为拿到了不同的index）
2. 消费者只在drain()中读取，而drain()通过compare-and-swap确定读取范围
3. 读写之间的happens-before关系是通过consumerIndex的volatile语义保证的

6.3 线程安全保证全景图

+------------------------------------------------------------------+ | Buffer 线程安全保证全景 | +------------------------------------------------------------------+ | | | 生产者侧（多个线程并发写入）: | | ┌─────────────────────────────────────────────────────┐ | | │ ① getAndIncrement() → CAS原子获取位置 │ | | │ 保证：每个线程拿到唯一的写入位置 │ | | │ │ | | │ ② buffer[slot] = data → 写入各自独占的slot │ | | │ 保证：不同线程写入不同位置，天然无冲突 │ | | │ │ | | │ ③ 如果slot已被消费者读取过 → 直接覆盖（OVERWRITE） │ | | │ 或检查后拒绝 （BLOCKING） │ | | └─────────────────────────────────────────────────────┘ | | | | 消费者侧（单线程批量消费）: | | ┌─────────────────────────────────────────────────────┐ | | │ ① 读取writeIndex → 获取volatile最新值 │ | | │ 保证：看到所有生产者的最新写入 │ | | │ │ | | │ ② for循环读取 → 只读已提交的slot │ | | │ 保证：不会读到"写了一半"的数据 │ | | │ │ | | │ ③ consumerIndex = writeIndex → 原子更新 │ | | │ 保证：标记已消费，数据不会再被重复消费 │ | | └─────────────────────────────────────────────────────┘ | +------------------------------------------------------------------+

7. Buffer容量设计对性能的影响

7.1 容量选择指南

7.2 容量太小的问题

+------------------------------------------------------------------+ | Buffer容量太小 → 数据丢失 | +------------------------------------------------------------------+ | | | 时间线 (bufferSize=8, 高频写入): | | ──────────────────────────────────────────────────────────────→ | | | | 写入: [0][1][2][3][4][5][6][7] [0][1][2][3][4][5][6][7] ... | | 消费: [2] [2][3] | | | | 问题: 消费者还没来得及读[4]-[7]，生产者已经进入下一圈覆盖了 | | 结果: 数据[4]-[7]丢失！ | +------------------------------------------------------------------+

7.3 容量太大的问题

+------------------------------------------------------------------+ | Buffer容量太大 → 延迟增加 | +------------------------------------------------------------------+ | | | 时间线 (bufferSize=16384, 低频写入): | | ──────────────────────────────────────────────────────────────→ | | | | 写入: 1条/s → 需要16384秒(4.5小时)才能填满 | | 消费: 消费者一直在等"凑够一批" | | | | 问题: 消费者可能等待很久才能收到数据 | | 结果: SkyWalking UI上看到的是4.5小时前的数据！ | +------------------------------------------------------------------+

7.4 SkyWalking的默认配置

// DataCarrier 默认配置publicstaticfinalintDEFAULT_BUFFER_SIZE=512;// 生产者数量（每个CPU核心一个生产者线程）publicstaticfinalintDEFAULT_CHANNEL_SIZE=Math.min(4,Runtime.getRuntime().availableProcessors());

默认Buffer大小512，Channel数量4（或CPU核心数）。对于大多数场景，总缓冲容量 4×512=2048条，足以应对正常的流量波动。

8. 总结

Buffer用最简单的配方——数组 + AtomicInteger——实现了无锁、零GC、线程安全的高性能环形缓冲区：

1. 数组提供连续内存和O(1)随机访问
2. **AtomicInteger.getAndIncrement()**用CAS原子分配写入位置
3. 位运算取模i & maxIndex实现零开销的环形索引
4. OVERWRITE允许在消费者慢时覆盖旧数据，BLOCKING则拒绝写入
5. drain()批量消费减少方法调用和网络请求

代码不过两百行，却处处透着精心的工程考量。下一篇文章，我们往上一层——看看Channel如何把多个Buffer组织起来，实现多生产者之间的负载均衡。

上一篇【第25篇】SkyWalking的内存消息队列DataCarrier——高性能数据中转站
下一篇【第27篇】Channel源码解析——多Buffer协同的分区艺术