【SkyWalking从入门到精通】第25篇：SkyWalking的内存消息队列DataCarrier——高性能数据中转站

上一篇【第24篇】SkyWalking Kubernetes部署实战——让监控与云原生共舞
下一篇【第26篇】Buffer源码深度解析——无锁环形缓冲区的精巧设计

1. 引言：为什么不用现成的队列？

想象一个场景：你的Java应用在欢快地为用户服务时，SkyWalking Agent正在默默采集每一个请求的Trace数据。高峰期每秒可能有上千个Span需要从应用程序发送到OAP Server。这些数据去哪里了？直接网络发送？不行，太慢，会拖累业务线程。扔进一个队列异步发送？用什么队列呢？

java.util.concurrent.BlockingQueue？太慢了，有锁竞争。

Disruptor？确实快，但它是外部依赖，SkyWalking Agent要保持零依赖。

Kafka？Agent端跑个Kafka Client太"重"了。

于是SkyWalking团队自己造了一个"轮子"——DataCarrier。它够轻，专为Agent场景设计；够快，无锁数据结构；够"抠"，几乎不产生GC压力。

2. 为什么需要内存队列？

2.1 数据生产与消费的速度差

+------------------------------------------------------------------+ | 数据生产消费速度差 | +------------------------------------------------------------------+ | | | 生产速度 (业务线程) 消费速度 (gRPC发送) | | ┌──────────────────────┐ ┌──────────────────────┐ | | │ 每个请求产生多个Span │ │ gRPC批量发送到OAP │ | | │ 高峰期: 10000 Span/s │ │ 网络吞吐: ~2000条/s │ | | │ 瞬时速度极快 │ │ 速度受网络限制 │ | | └──────────────────────┘ └──────────────────────┘ | | │ │ | | │ 速度差5000条/s │ | | └──────────────┬──────────────┘ | | ▼ | | ┌──────────────────────────┐ | | │ DataCarrier │ | | │ 缓冲消峰，平滑输出 │ | | └──────────────────────────┘ | +------------------------------------------------------------------+

2.2 DataCarrier在整体架构中的位置

+------------------------------------------------------------------+ | SkyWalking Agent 数据流架构 | +------------------------------------------------------------------+ | | | 业务线程池 DataCarrier层 gRPC发送层 | | ┌──────────┐ produce ┌──────────────┐ | | │ Thread-1 │ ───────────────→ │ │ consume | | ├──────────┤ │ DataCarrier │ ──────────────→ | | │ Thread-2 │ ───────────────→ │ (内存队列) │ | | ├──────────┤ │ │ ┌───────────┐ | | │ Thread-3 │ ───────────────→ │ │ ──→ │ gRPC Stub │ | | ├──────────┤ │ │ └─────┬─────┘ | | │ ... │ └──────────────┘ │ | | ├──────────┤ ▼ | | │ Thread-N │ ┌─────────────────┐ | | └──────────┘ │ OAP Server │ | | │ (11800端口) │ | | └─────────────────┘ | +------------------------------------------------------------------+

3. DataCarrier的三大核心概念

DataCarrier由三个层层嵌套的组件构成，就像俄罗斯套娃：

+------------------------------------------------------------------+ | DataCarrier 三层架构 | +------------------------------------------------------------------+ | | | ┌──────────────────────────────────────────────────────────┐ | | │ DataCarrier │ | | │ ┌────────────────────────────────────────────────────┐ │ | | │ │ Channel │ │ | | │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ | | │ │ │ Buffer-0 │ │ Buffer-1 │ │ Buffer-2 │ ... │ │ | | │ │ │ ┌──────┐ │ │ ┌──────┐ │ │ ┌──────┐ │ │ │ | | │ │ │ │ slot │ │ │ │ slot │ │ │ │ slot │ │ │ │ | | │ │ │ │ slot │ │ │ │ slot │ │ │ │ slot │ │ │ │ | | │ │ │ │ ... │ │ │ │ ... │ │ │ │ ... │ │ │ │ | | │ │ │ └──────┘ │ │ └──────┘ │ │ └──────┘ │ │ │ | | │ │ └──────────┘ └──────────┘ └──────────┘ │ │ | | │ └────────────────────────────────────────────────────┘ │ | | └──────────────────────────────────────────────────────────┘ | | | | Buffer: 存储单元 — 环形数组，存放实际数据 | | Channel: 分发单元 — 多个Buffer的数组，按线程ID分发 | | DataCarrier: 协调单元 — 管理生产者和消费者 | +------------------------------------------------------------------+

3.1 Buffer（数据缓冲区）

Buffer是最底层的存储单元，一个定长的环形数组：

// 核心结构（简化版）publicclassBuffer<T>{privatefinalObject[]buffer;// 底层数组privatefinalAtomicIntegerindex;// 写入位置（原子操作）privatefinalintbufferSize;privatefinalintmaxIndex;// 两种写满策略publicstaticfinalintOVERWRITE=0;// 覆盖旧数据publicstaticfinalintBLOCKING=1;// 阻塞等待publicBuffer(intbufferSize,intstrategy){...}publicbooleansave(Tdata){...}publicvoiddrain(List<T>consumeList){...}}

3.2 Channel（通道/分区）

Channel是一组Buffer的集合，实现了生产者分离：

// 核心结构（简化版）publicclassChannels<T>{privatefinalBuffer<T>[]bufferChannels;// Buffer数组privatefinalintchannelSize;// 生产者根据线程ID取模选择BufferpublicintgetChannel(longthreadId){return(int)(threadId%channelSize);}}

3.3 DataCarrier（协调者）

DataCarrier是最外层，负责把Channel和消费者线程连接起来：

// 核心结构（简化版）publicclassDataCarrier<T>{privatefinalChannels<T>channels;// Channel集合privatefinalIDriverdriver;// 消费线程驱动privatefinalIConsumer<T>consumer;// 消费者接口// 生产消息publicbooleanproduce(Tdata){// 按线程ID选择Channel → 选择Buffer → saveintchannel=Thread.currentThread().getId()%channels.size();returnchannels.getBuffer(channel).save(data);}// 消费消息（批量）publicvoidconsume(){channels.drain(consumer);// 驱动消费}}

4. 与JDK内置队列的对比

4.1 对比表

4.2 为什么DataCarrier更快？

+------------------------------------------------------------------+ | LinkedBlockingQueue vs DataCarrier | +------------------------------------------------------------------+ | | | LinkedBlockingQueue: | | ┌─────────────────────────────────────────────────────┐ | | │ put()线程: 获取putLock → 创建Node → 入队 → 释放锁 │ | | │ take()线程: 获取takeLock → 出队 → 释放锁 │ | | │ │ | | │ 问题: │ | | │ ① 每次put都会new Node → GC压力 │ | | │ ② 两个锁竞争 → 并发度受限 │ | | │ ③ 链表遍历 → 缓存不友好 │ | | └─────────────────────────────────────────────────────┘ | | | | DataCarrier: | | ┌─────────────────────────────────────────────────────┐ | | │ save()线程: AtomicInteger CAS自增 → 直接写入数组 │ | | │ drain()线程: 批量读取 → 批量处理 → 批量标记消费 │ | | │ │ | | │ 优势: │ | | │ ① 数组预分配 → 零对象创建 → 零GC │ | | │ ② 原子操作无锁 → 无竞争 │ | | │ ③ 数组连续内存 → CPU缓存友好 │ | | │ ④ 批量消费 → 减少操作频率 │ | | └─────────────────────────────────────────────────────┘ | +------------------------------------------------------------------+

5. 零GC压力设计思想

5.1 对象复用

DataCarrier的"零GC"不是魔法，而是靠对象复用实现：

// 错误的做法（产生GC）publicvoidproduce(Spanspan){SpanObjectobj=newSpanObject();// 每次new对象obj.setSpanId(span.getSpanId());obj.setStartTime(span.getStartTime());// ... 填数据buffer.save(obj);// GC！}// DataCarrier的做法（对象复用）publicvoidproduce(Spanspan){// Buffer内部维护了一个预分配的对象数组// save()时直接覆盖已有对象的字段，不创建新对象SpanObjectobj=buffer.getSlot();// 复用已有对象obj.reset();// 重置字段obj.setSpanId(span.getSpanId());obj.setStartTime(span.getStartTime());buffer.save(obj);// 零GC！}

5.2 内存布局设计

+------------------------------------------------------------------+ | DataCarrier 内存布局 | +------------------------------------------------------------------+ | | | Buffer数组（环形缓冲区） | | ┌──────────────────────────────────────────────────┐ | | │ │ | | │ Index: 0 1 2 3 4 5 6 7 ... N-1 │ | | │ ┌───┬───┬───┬───┬───┬───┬───┬───┬─────┐ │ | | │ │obj│obj│obj│obj│obj│obj│obj│obj│ ... │ │ | | │ └───┴───┴───┴───┴───┴───┴───┴───┴─────┘ │ | | │ ↑ writeCursor (AtomicInteger) │ | | │ ↑ producerIndex | | │ │ │ | │ 预分配: 初始化时创建N个空对象 │ │ | │ 写入时: 覆盖已有对象的字段 │ │ | │ 读取时: 读取对象引用（不复制） │ │ | │ 消费后: 标记位置为空闲 │ │ | │ │ │ | │ 整个生命周期: 0次new / 0次GC │ │ | └──────────────────────────────────────┼───────────────────────┘ │ +------------------------------------------------------------------+

5.3 为什么GC对APM系统是致命的

APM系统的目标是不干扰被监控的应用。如果Agent本身触发了大量GC，就形成了"观察者效应"——监控反而成了性能问题的来源：

+------------------------------------------------------------------+ | GC压力对APM的负面影响 | +------------------------------------------------------------------+ | | | Agent GC → STW暂停 → 业务线程被阻塞 | | → 请求响应变慢 | | → 吞吐量下降 | | → 用户感知变差 | | → 监控数据也出现异常波动 | | → 你分不清是系统真的慢了，还是Agent导致的 | | | | 这个恶性循环一旦形成，APM系统就成了"内鬼"。 | +------------------------------------------------------------------+

DataCarrier的零GC设计就是要把这个循环从源头斩断。

6. DataCarrier在Agent数据上报中的角色

6.1 完整的数据上报链路

+------------------------------------------------------------------+ | Agent端数据上报完整流程 | +------------------------------------------------------------------+ | | | ┌──────────────────────────────────────────────────────────────┐ │ | │ 应用程序 (被监控) │ │ | │ HTTP Request → Controller → Service → DAO → Database │ │ | │ │ │ │ │ │ │ | │ ▼ ▼ ▼ ▼ │ │ | │ ┌────┴────┐ ┌────┴────┐ ┌───┴───┐ ┌┴───────────┐ │ │ | │ │EntrySpan│ │LocalSpan│ │ExitSpan│ │ExitSpan(DB)│ │ │ | │ └────┬────┘ └────┬────┘ └───┬───┘ └─────┬──────┘ │ │ | └───────┼────────────────┼──────────┼────────────┼─────────────┘ │ | │ │ │ │ │ | └────────────────┴──────────┴────────────┘ │ | │ │ | ▼ │ | ┌──────────────────┐ │ | │ TracingContext │ │ | │ TraceSegment创建 │ │ | │ 收集所有Span │ │ | └────────┬─────────┘ │ | │ │ | ▼ │ | ┌──────────────────┐ │ | │ DataCarrier │ ← 主角登场！ │ | │ produce() │ │ | └────────┬─────────┘ │ | │ (异步、批量) │ | ▼ │ | ┌──────────────────┐ │ | │ gRPC Sender │ │ | │ 批量上报OAP │ │ | └────────┬─────────┘ │ | │ │ | ▼ │ | ┌──────────────────┐ │ | │ OAP Server │ │ | │ 11800端口 │ │ | └──────────────────┘ │ +------------------------------------------------------------------+

6.2 生产消费的完整代码示意

// ===== 生产者端（业务线程） =====publicclassTraceSegmentServiceClient{privatefinalDataCarrier<TraceSegment>carrier;publicvoidsend(TraceSegmentsegment){// 非阻塞写入，如果Buffer满则根据策略处理if(!carrier.produce(segment)){// OVERWRITE策略：自动覆盖，不阻塞// BLOCKING策略：这里会返回失败（或阻塞）logger.warn("DataCarrier buffer full, segment dropped");}}}// ===== 消费者端（独立线程） =====publicclassTraceSegmentConsumerimplementsIConsumer<TraceSegment>{@Overridepublicvoidconsume(List<TraceSegment>data){// 批量消费：一次拉取多条TraceSegmentif(data.isEmpty())return;// 转换为gRPC对象，批量发送List<SegmentObject>segmentList=newArrayList<>();for(TraceSegmentsegment:data){segmentList.add(transform(segment));}// gRPC批量上报grpcChannel.sendBatch(segmentList);}}// ===== 启动消费线程 =====ConsumerThreadconsumer=newConsumerThread("TraceSegmentConsumer",carrier,consumer);consumer.start();

7. DataCarrier vs Disruptor

很多读者可能会问：“这不就是Disruptor吗？为什么不直接用？”

7.1 设计哲学差异

+------------------------------------------------------------------+ | DataCarrier vs Disruptor 设计对比 | +------------------------------------------------------------------+ | | | 特性 | Disruptor | DataCarrier | +------------------------------------------------------------------+ | 定位 | 通用高性能消息队列 | APM Agent专用场景 | | 依赖 | 第三方库(~100KB) | 零依赖 | | Sequence管理 | Sequence + Padding | AtomicInteger | | 生产者选择策略 | ClaimStrategy | 线程ID取模 | | 内存屏障 | Memory Barriers | 依赖volatile | | 缓存行填充 | 有(@Contended) | 无 | | 背压处理 | 多种WaitStrategy | 仅OVERWRITE/BLOCKING | +------------------------------------------------------------------+

7.2 为什么不直接用Disruptor

两个原因：

1. 零依赖原则

SkyWalking Agent的设计哲学是"zero dependency"——被监控的应用不应该因为Agent而引入新的依赖。Disruptor虽然性能极致，但它是一个~100KB的第三方JAR包。

2. 场景简化

Disruptor的很多高级特性（多种WaitStrategy、ClaimStrategy、EventProcessor等）在Agent场景下用不上。Agent场景的需求很明确：

• 多生产者（多个业务线程写Span数据）
• 单消费者（一个消费线程批量发数据）
• 允许丢失（采样场景下，数据可丢弃）

场景越简单，实现就应该越简单。DataCarrier用几百行代码就解决了Disruptor需要几千行才能解决的问题。

8. 总结

DataCarrier是SkyWalking Agent中的"隐形英雄"——用户看不到它，但它默默扛起了所有数据中转任务：

1. Buffer：无锁环形数组，OVERWRITE/BLOCKING两种策略
2. Channel：多Buffer组织，线程取模分配生产者
3. DataCarrier：串联生产者和消费者，批量消费提升效率
4. 零GC：对象预分配+复用，不给业务应用增加GC负担
5. 零依赖：纯Java实现，不需要引入任何第三方库

这三篇文章（025-028），我们将逐一深入DataCarrier的源码实现。下一篇，从最底层的Buffer开始解剖。

上一篇【第24篇】SkyWalking Kubernetes部署实战——让监控与云原生共舞
下一篇【第26篇】Buffer源码深度解析——无锁环形缓冲区的精巧设计