上一篇【第24篇】SkyWalking Kubernetes部署实战——让监控与云原生共舞下一篇【第26篇】Buffer源码深度解析——无锁环形缓冲区的精巧设计1. 引言为什么不用现成的队列想象一个场景你的Java应用在欢快地为用户服务时SkyWalking Agent正在默默采集每一个请求的Trace数据。高峰期每秒可能有上千个Span需要从应用程序发送到OAP Server。这些数据去哪里了直接网络发送不行太慢会拖累业务线程。扔进一个队列异步发送用什么队列呢java.util.concurrent.BlockingQueue太慢了有锁竞争。Disruptor确实快但它是外部依赖SkyWalking Agent要保持零依赖。KafkaAgent端跑个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是最底层的存储单元一个定长的环形数组// 核心结构简化版publicclassBufferT{privatefinalObject[]buffer;// 底层数组privatefinalAtomicIntegerindex;// 写入位置原子操作privatefinalintbufferSize;privatefinalintmaxIndex;// 两种写满策略publicstaticfinalintOVERWRITE0;// 覆盖旧数据publicstaticfinalintBLOCKING1;// 阻塞等待publicBuffer(intbufferSize,intstrategy){...}publicbooleansave(Tdata){...}publicvoiddrain(ListTconsumeList){...}}3.2 Channel通道/分区Channel是一组Buffer的集合实现了生产者分离// 核心结构简化版publicclassChannelsT{privatefinalBufferT[]bufferChannels;// Buffer数组privatefinalintchannelSize;// 生产者根据线程ID取模选择BufferpublicintgetChannel(longthreadId){return(int)(threadId%channelSize);}}3.3 DataCarrier协调者DataCarrier是最外层负责把Channel和消费者线程连接起来// 核心结构简化版publicclassDataCarrierT{privatefinalChannelsTchannels;// Channel集合privatefinalIDriverdriver;// 消费线程驱动privatefinalIConsumerTconsumer;// 消费者接口// 生产消息publicbooleanproduce(Tdata){// 按线程ID选择Channel → 选择Buffer → saveintchannelThread.currentThread().getId()%channels.size();returnchannels.getBuffer(channel).save(data);}// 消费消息批量publicvoidconsume(){channels.drain(consumer);// 驱动消费}}4. 与JDK内置队列的对比4.1 对比表特性LinkedBlockingQueueArrayBlockingQueueDisruptorDataCarrier锁机制两把锁(putLock/takeLock)一把ReentrantLock无锁(Sequence Barrier)无锁(AtomicInteger)数据结构链表数组环形数组环形数组GC压力高(节点创建)低(数组复用)极低(预分配)极低(预分配)有界支持可选必选是是外部依赖JDKJDK第三方库零依赖多生产者是是是(多ProducerBarrier)是(线程取模)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不是魔法而是靠对象复用实现// 错误的做法产生GCpublicvoidproduce(Spanspan){SpanObjectobjnewSpanObject();// 每次new对象obj.setSpanId(span.getSpanId());obj.setStartTime(span.getStartTime());// ... 填数据buffer.save(obj);// GC}// DataCarrier的做法对象复用publicvoidproduce(Spanspan){// Buffer内部维护了一个预分配的对象数组// save()时直接覆盖已有对象的字段不创建新对象SpanObjectobjbuffer.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{privatefinalDataCarrierTraceSegmentcarrier;publicvoidsend(TraceSegmentsegment){// 非阻塞写入如果Buffer满则根据策略处理if(!carrier.produce(segment)){// OVERWRITE策略自动覆盖不阻塞// BLOCKING策略这里会返回失败或阻塞logger.warn(DataCarrier buffer full, segment dropped);}}}// 消费者端独立线程 publicclassTraceSegmentConsumerimplementsIConsumerTraceSegment{Overridepublicvoidconsume(ListTraceSegmentdata){// 批量消费一次拉取多条TraceSegmentif(data.isEmpty())return;// 转换为gRPC对象批量发送ListSegmentObjectsegmentListnewArrayList();for(TraceSegmentsegment:data){segmentList.add(transform(segment));}// gRPC批量上报grpcChannel.sendBatch(segmentList);}}// 启动消费线程 ConsumerThreadconsumernewConsumerThread(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中的隐形英雄——用户看不到它但它默默扛起了所有数据中转任务Buffer无锁环形数组OVERWRITE/BLOCKING两种策略Channel多Buffer组织线程取模分配生产者DataCarrier串联生产者和消费者批量消费提升效率零GC对象预分配复用不给业务应用增加GC负担零依赖纯Java实现不需要引入任何第三方库这三篇文章025-028我们将逐一深入DataCarrier的源码实现。下一篇从最底层的Buffer开始解剖。上一篇【第24篇】SkyWalking Kubernetes部署实战——让监控与云原生共舞下一篇【第26篇】Buffer源码深度解析——无锁环形缓冲区的精巧设计