数据采集层 · Receiver

这一篇解决什么
OAP 后端怎么收数据？答案是一堆 receiver 插件，每个对应一种协议或数据源。这篇先讲贯穿所有 receiver 的共同机制（共享服务器 + Source/Dispatcher 总线），再按"协议大类"分组讲每个插件，最后对比 push 和 pull 两种采集模式。读完应该能回答：trace/jvm/log/meter/otel/zipkin/envoy 各自怎么进来，差别在哪。

一句话定位

Receiver 体系解决的是"用统一接口接收来自各种探针、协议、数据源的可观测数据，把它们规范化成内部 Source 模型送入流处理链"。做成插件化是因为数据来源的协议（SkyWalking 原生 gRPC / OpenTelemetry OTLP / Zipkin / Envoy ALS / Kafka 拉取等）和上报方式彼此完全不同——每加一种协议都改一个 monolith 会越来越乱。插件化让 OAP 按需启用、独立演进、互不干扰，又能共享同一套底层服务器和 Source→Dispatcher 流转机制。

贯穿所有 receiver 的共同机制

先看骨架，再看具体插件。

1. 模块系统的标准用法

每个 receiver 是一个 ModuleDefine + ModuleProvider。启动流程固定四步：prepare() → start() → notifyAfterCompleted()，模块间依赖通过 requiredModules() 声明排序。几乎所有 receiver 的 start() 做的都是同一件事：从 SharingServerModule 拿到 GRPCHandlerRegister/HTTPHandlerRegister，然后 addHandler(自己的 gRPC/HTTP 处理器)。

2. 数据流转总线：SourceReceiver → DispatcherManager

所有 receiver 收到原始数据后的统一去向：

• 接口 SourceReceiver（oap-server/server-core/.../source/SourceReceiver.java:28）只有一个方法 receive(ISource source)。
• 实现 SourceReceiverImpl（SourceReceiverImpl.java:29）持 DispatcherManager，receive() 转发给 dispatcherManager.forward(source)。
• DispatcherManager.forward（oap-server/server-core/.../analysis/DispatcherManager.java:46）按 source.scope() 找到一组 SourceDispatcher，先 source.prepare()，再对每个 dispatcher 调 dispatch(source)（:60-62）。一个 Source 可被多个 dispatcher 处理——比如一个 Service 同时产生 resp_time、sla、cpm 等多个 metric。

Source / Dispatcher 是什么
Source 是"一份数据事件的规范化形态"，比如 Service、ServiceRelation、Endpoint、Segment、Log、ServiceInstanceJVMMemory 等几十种（server-core/.../source/ 下有 60+ 个 Source 类）。每个有 scope() 整数 ID。

SourceDispatcher 是"针对某类 Source 做聚合计算的逻辑"。OAL 引擎扫描用户写的 .oal 脚本，为每条指标规则生成一个 dispatcher 类注册进 DispatcherManager。这样 receiver 只管"产 Source"，不关心要算哪些指标——指标规则与数据接收彻底解耦。

启动时 SourceReceiverImpl.scan()（:51）用 Guava ClassPath 扫描 org.apache.skywalking 包下所有 SourceDispatcher 实现类自动注册。这就是"为什么 OAL 生成的 dispatcher 不需要手动接线"。

为什么用"一个 Source 可被多个 dispatcher 处理"这种一发多收的设计？因为同一个原始事件往往要算多种指标——一次服务调用既算响应时间均值、又算成功率、又算 QPS。如果一对一，要么 receiver 重复构造多个 Source、要么把多种聚合塞进一个 dispatcher。一发多收让每条聚合规则独立成 dispatcher，新增一条规则只加一个 dispatcher，不动其他。这是观察者模式在流式聚合里的典型应用。

3. 共享服务器：skywalking-sharing-server-plugin

绝大多数 receiver 自己不建 gRPC/HTTP 服务器，而是向 SharingServerModule 注册 handler。这是整个 receiver 体系的"地基"。

• 模块：oap-server/server-receiver-plugin/skywalking-sharing-server-plugin/.../sharing/server/SharingServerModule.java
• Provider：SharingServerModuleProvider.java:41

它做三件事：

SharingServer 的三段活

1. prepare()（:76）：根据配置 restPort/gRPCPort 决定新建 HTTPServer/GRPCServer，还是用"代理注册器"模式（当本 receiver 没配端口时，把 handler 暂存，等有服务器的 CoreModule 启动后再委托注册）。GRPCServer 由 Netty 实现，注释明确说"最多给 4 个 OAP 端点用：core-grpc、receiver-grpc、ebpf-grpc、als-grpc"（oap-server/server-library/library-server/.../grpc/GRPCServer.java:44）。
2. 把 GRPCHandlerRegister 和 HTTPHandlerRegister 作为 Service 注册出去，供其他 receiver 通过 getManager().find(SharingServerModule.NAME).provider().getService(GRPCHandlerRegister.class) 取用。
3. notifyAfterCompleted()（:173）：真正 grpcServer.start() / httpServer.start()。

代理注册器 ReceiverGRPCHandlerRegister
oap-server/server-receiver-plugin/skywalking-sharing-server-plugin/.../sharing/server/ReceiverGRPCHandlerRegister.java:30。

当 SharingServerModuleProvider 自己没配 gRPC 端口（比如某些部署把 receiver 合并进 core），它先创建一个空壳的 ReceiverGRPCHandlerRegister 注册成 Service，receiver 往里 addHandler 会先缓存；等 start() 时（:160）再从 CoreModule 取到真正的 GRPCHandlerRegister 赋值进来，缓存的处理就委托过去。

这让 receiver 代码完全不感知"服务器在哪建"，只管注册。还支持 addFilter 在 addHandler 之前调用，给每个 handler 绑上拦截器（如鉴权 AuthenticationInterceptor）。

4. gRPC stream 与 proto 基础

receiver handler 继承 proto 生成的 *Grpc.*ImplBase，实现 GRPCHandler 标记接口。

gRPC stream / StreamObserver / proto 速成

• proto：用 Protocol Buffers 描述语言写数据结构和接口契约，编译后生成 Java 类。比如 Tracing.proto 定义 TraceSegmentReportService 服务，含 rpc collect(stream SegmentObject) returns (Commands)。
• stream 关键字：表示客户端流式上传——客户端可以连续 onNext 发多个对象，最后 onCompleted。StreamObserver<T> 就是 gRPC 给的回调接口：每来一条调 onNext，结束调 onCompleted，出错调 onError。
• unary RPC：一次请求一次响应，没有 stream 关键字，如 JVM 的 collect(JVMMetricCollection) returns (Commands)。

SkyWalking 原生协议里，trace、mesh、meter、log 用 client-streaming（流量大、可背压），JVM/CLR/event 用 unary（定时批量、简单）。

5. 与 apm-protocol 的关系

apm-protocol/apm-network/src/main/proto/ 是所有 SkyWalking 原生协议的 proto 源头。关键映射：

Compat 协议：向后兼容
还有一组 *Compat.proto（TracingCompat / JVMMetricCompat / …）。它们用于协议向后兼容：proto 字段演进时，旧版 agent 仍能用兼容服务名上报，handler 里 *Compat 类（如 TraceSegmentReportServiceHandlerCompat）把旧协议请求转成新处理逻辑。每个原生 receiver 都成对注册主 handler 和 *Compat handler。

receiver-proto/src/main/proto/ 存 OpenTelemetry OTLP 协议的 proto 和 Envoy 相关 proto，供 otel-receiver 和 envoy-metrics-receiver 编译使用。

三层分工

理解 receiver 体系的关键是分清三层职责：

receiver 与 analyzer 通过 ISegmentParserService/IMeterProcessService/ILogAnalyzerService 等 Service 接口解耦。这就是 requiredModules 几乎都列 AnalyzerModule/LogAnalyzerModule 的原因。

A 组：SkyWalking 原生 gRPC 协议接收器

共同点：handler 继承 *Grpc.*ImplBase，通过 SharingServerModule 注册，数据流向 SourceReceiver。区别只在"中间解析层"放哪。

A1. trace-receiver · 链路追踪

• Module/Provider：TraceModule / TraceModuleProvider（.../trace/provider/TraceModuleProvider.java:39）
• 传输：gRPC client-streaming + HTTP REST
• 关键类：TraceSegmentReportServiceHandler（.../handler/v8/grpc/TraceSegmentReportServiceHandler.java:39）继承 TraceSegmentReportServiceGrpc.TraceSegmentReportServiceImplBase，collect 返回 StreamObserver，每收到一个 SegmentObject 调 segmentParserService.send(segment)（:73）。

数据流转（最完整的一条链）：

receiver 与 analyzer 的分工
trace receiver 自身只做"接收 + 转发"，真正解析成 Source 的逻辑在 analyzer 模块。TraceModuleProvider.java:46 从 AnalyzerModule 取 ISegmentParserService；SegmentParserServiceImpl.send（oap-server/analyzer/agent-analyzer/.../trace/parser/SegmentParserServiceImpl.java:38）new 一个 TraceAnalyzer 并 doAnalysis；TraceAnalyzer.doAnalysis（.../trace/parser/TraceAnalyzer.java:45）遍历每个 SpanObject，按 SpanType（Entry/Exit/Local）分发到不同 listener，最后 build()。这就是为什么 TraceModuleProvider.requiredModules() 含 AnalyzerModule.NAME。

A2. jvm-receiver · JVM 监控

• Module/Provider：JVMModule / JVMModuleProvider（.../jvm/provider/JVMModuleProvider.java:32）
• 传输：gRPC unary（collect(JVMMetricCollection) returns Commands）
• 关键类：JVMMetricReportServiceHandler（.../jvm/provider/handler/JVMMetricReportServiceHandler.java:33）

与 trace 的区别：JVM 不走 AnalyzerModule，handler 自己直接 new JVMSourceDispatcher（在 analyzer 模块）。collect() 先做命名规范化（NamingControl.formatServiceName），再 jvmSourceDispatcher.sendMetric(...) 把每个 JVMMetric 转成 ServiceInstanceJVMMemory/ServiceInstanceJVMCPU/ServiceInstanceJVMGC/ServiceInstanceJVMThread 等 Source 并 sourceReceiver.receive()。

Provider start() 里 OALEngineLoaderService.load(JVMOALDefine.INSTANCE)（JVMModuleProvider.java:57）——JVM 的指标规则写在 JVMOALDefine 里，启动时编译成 dispatcher 注入 DispatcherManager。

A3. clr-receiver · .NET CLR

与 JVM 几乎同构，数据是 .NET CLR 指标（CPU/GC/Thread）。CLRMetricReportServiceHandler + CLRSourceDispatcher（.../clr/provider/handler/CLRSourceDispatcher.java:36），也是 gRPC unary。

A4. mesh-receiver · 服务网格

易混点
这个 receiver 处理的是 SkyWalking 自定义的 service-mesh proto（ServiceMeshMetrics），不是 Envoy ALS 的 ALS proto。Envoy ALS 在 D 组的 envoy-metrics-receiver-plugin。

• Module/Provider：MeshReceiverModule / MeshReceiverProvider（.../mesh/MeshReceiverProvider.java:32）
• 传输：gRPC client-streaming（collect(stream ServiceMeshMetrics) returns MeshProbeDownstream）
• 关键类：MeshGRPCHandler（.../mesh/MeshGRPCHandler.java:30），每条 onNext 调 TelemetryDataDispatcher.process(metrics)。
• TelemetryDataDispatcher（.../mesh/TelemetryDataDispatcher.java:60）静态工具，init() 时从 CoreModule 取 SourceReceiver/NamingControl。dispatchHTTPMetrics/dispatchTCPMetrics 把一条 mesh metric 拆成多个 Source（Service、ServiceInstance、Endpoint、ServiceRelation，或 TCP 情况的 TCPService/TCPServiceRelation），分别 SOURCE_RECEIVER.receive(...)。这是"一条入参 → 多个 Source"的典型。

A5. log-receiver · 日志

• Module/Provider：LogModule / LogModuleProvider（.../log/provider/LogModuleProvider.java:36）
• 传输：gRPC client-streaming + HTTP REST
• 关键类：LogReportServiceGrpcHandler（.../log/provider/handler/grpc/LogReportServiceGrpcHandler.java:41），每条 onNext 调 logAnalyzerService.doAnalysis(...)（:96）。细节：用 setServiceName 在 stream 内复用上一个非空 service 名（:79），因为日志流里 service 字段可能只第一条有。

像 trace 一样依赖独立的 LogAnalyzerModule，取 ILogAnalyzerService.doAnalysis()。日志最终走 LAL 规则转成 Log/LogMetadata 等 Source。

A6. meter-receiver · 自定义指标

• Module/Provider：MeterReceiverModule / MeterReceiverProvider（.../meter/provider/MeterReceiverProvider.java:34）
• 传输：gRPC client-streaming（collect(stream MeterData)）+ 批量（collectBatch(stream MeterDataCollection)）
• 关键类：MeterServiceHandler（.../meter/provider/handler/MeterServiceHandler.java:42）

数据流转（委托 analyzer 的"攒批后统一处理"模式）：Provider 从 AnalyzerModule 取 IMeterProcessService（MeterReceiverProvider.java:59）。collect() 里 processService.createProcessor() 得到 MeterProcessor，每条 MeterData 调 processor.read 累积，onCompleted() 时 processor.process() 真正触发 MAL 规则计算 → Source → dispatcher。

A7–A13. 其他原生 receiver 速览

B 组：OpenTelemetry 协议接入

otel-receiver-plugin 与原生协议的核心区别：

四点不同：

1. proto 完全不同：用 OpenTelemetry 官方 OTLP proto（io.opentelemetry.proto.collector.*.*Grpc），不是 SkyWalking 自己的 apm.network.*。proto 源在 receiver-proto/src/main/proto/opentelemetry/proto/。
2. 可拔插 Handler 机制：OtelMetricReceiverProvider（.../otel/OtelMetricReceiverProvider.java:34）在 prepare() 用 ServiceLoader.load(Handler.class)（:80）发现所有 Handler 实现，按 enabledHandlers 过滤；start() 里对每个 handler 调 init() + active()（:91-93）。加新 OTLP 数据类型只要实现 Handler 接口并注册 SPI。
3. 数据流转分两类：
   • Trace：OpenTelemetryTraceHandler（.../otel/otlp/OpenTelemetryTraceHandler.java:72）把 OTLP span 先转成 Zipkin Span（convertSpan，:191），再交给 ZipkinReceiverModule 的 SpanForwardService.send()（:184）——OTLP trace 复用 Zipkin 处理链。
   • Metric：OpenTelemetryMetricHandler（.../otel/otlp/OpenTelemetryMetricHandler.java:38）调 OpenTelemetryMetricRequestProcessor.processMetricsRequest()，用 MAL 规则把 OTLP metrics 转成 SkyWalking Meter 体系。
4. gRPC 都是 unary：OTLP 的 export(ExportTraceServiceRequest) 是 unary，不像原生 trace 是 client-streaming。

用 OTLP 接 trace 必须同时启用 zipkin receiver
OpenTelemetryTraceHandler 把 OTLP span 转成 zipkin Span 后调 SpanForwardService.send()，这个 service 来自 ZipkinReceiverModule。所以用 OTLP 接 trace 时必须同时启用 zipkin receiver。

AWS Firehose 是个旁支：AWSFirehoseReceiverModuleProvider 自己起独立 HTTP 服务器（firehose-http），收 AWS Kinesis Firehose 推送，但内部复用 OtelMetricReceiverModule 的 OpenTelemetryMetricRequestProcessor——本质是"另一种传输入口（AWS Firehose HTTP）+ OTLP 处理内核"。

C 组：Zipkin 协议

• Module/Provider：ZipkinReceiverModule / ZipkinReceiverProvider（.../zipkin/ZipkinReceiverProvider.java:37）
• 传输：HTTP REST（ZipkinSpanHTTPHandler，POST/GET）+ 可选 Kafka 消费。不依赖 sharing server——自己 new HTTPServer（线程名 zipkin-http），是少数完全独立服务器的 receiver。

核心流转类 SpanForward（.../zipkin/trace/SpanForward.java:55）实现 SpanForwardService：

OTLP trace 复用的就是这条链（见 B 组）。

D 组：Envoy ALS

• Module/Provider：EnvoyMetricReceiverModule / EnvoyMetricReceiverProvider（.../envoy/EnvoyMetricReceiverProvider.java:42）
• 传输：gRPC，两类服务：
  • Envoy ALS（Access Log Service）：AccessLogServiceGRPCHandler + V3 兼容，收 Envoy 访问日志（L7 HTTP / L4 TCP），用 Envoy 官方 als.proto。
  • Envoy Metrics Service：MetricServiceGRPCHandler + V3，可选，收 Envoy stats。
• 服务器策略：和 eBPF 一样可配独立 gRPC 端口（线程名 als-grpc），否则 fallback 到 sharing。
• 依赖 mesh receiver：requiredModules 含 MeshReceiverModule.NAME——Envoy ALS 复用 mesh 的 OAL 规则与部分逻辑。还加载 TCPOALDefine（当配了 alsTCPAnalysis）和 metadata-service-mapping.yaml 字段映射。

mesh-receiver vs envoy-metrics-receiver

• skywalking-mesh-receiver-plugin（A4）收 SkyWalking 私有 service-mesh proto（适合被 SkyWalking Rover/ALS adapter 转换过的数据）。
• envoy-metrics-receiver-plugin（D 组）收 Envoy 原生 ALS/Metrics proto（直接对接 Envoy）。

两者都产出 mesh 类 Source，但入口协议不同。

E 组：拉取式 Fetcher（pull）

与上面所有 receiver 不同——fetcher 不开放端口等数据来，而是主动去外部系统拉数据。

E1. kafka-fetcher · 把原生入口换成 Kafka

• Module/Provider：KafkaFetcherModule / KafkaFetcherProvider（oap-server/server-fetcher-plugin/kafka-fetcher-plugin/.../provider/KafkaFetcherProvider.java:42）
• 机制：prepare() 创建 KafkaFetcherHandlerRegister（:73）；start() 注册一系列 KafkaHandler（:78-89）然后 handlerRegister.start()。这些 handler 与原生 receiver 一一对应，复用同一处理内核：

这意味着 agent 可以不直连 OAP，而是把数据写 Kafka，OAP 从 Kafka 消费——适合大流量/解耦部署。处理内核完全复用，所以 requiredModules 仍含 AnalyzerModule/LogAnalyzerModule。它不依赖 SharingServerModule（不需要 gRPC 端口）。

E2. cilium-fetcher · gRPC stream 客户端

• Module/Provider：CiliumFetcherModule / CiliumFetcherProvider（.../cilium/CiliumFetcherProvider.java:38）
• 机制：OAP 作为 gRPC client 主动连 Cilium Hubble Relay 的 observer gRPC stream，订阅 flow 事件。start() 创建 CiliumNodeManager（:92）管理到多个 Hubble 节点的连接，加 CiliumFlowListener（:93），flow 事件解析成 Cilium 相关 Source（CiliumService/CiliumServiceRelation/CiliumServiceInstanceRelation）。
• 依赖：CoreModule + ClusterModule（多 OAP 节点要分摊 Hubble 连接），加载 CiliumOALDefine。

E3. fetcher-proto

仅 proto 定义模块（observer.proto 等），编译出 Cilium Hubble 的 Java stub，无运行时逻辑。

F 组：私有 HTTP 协议接收器

F1. telegraf-receiver · InfluxDB line protocol

TelegrafReceiverProvider（.../telegraf/provider/TelegrafReceiverProvider.java:43）复用 sharing server 的 HTTPHandlerRegister。Telegraf 用 InfluxDB line protocol 格式上报指标，Provider start() 加载 MAL 规则（Rules.loadRules，:83），取 MeterSystem（:92）注入 handler。本质是"telegraf 格式 → MAL 体系"。

F2. zabbix-receiver · Zabbix 私有 JSON 协议

ZabbixReceiverProvider（.../zabbix/provider/ZabbixReceiverProvider.java:37）自实现 Zabbix 协议服务器（ZabbixServer，:84 自己 start()），不依赖 sharing server——因为 Zabbix agent 用 Zabbix 私有 JSON 协议，不是标准 HTTP/gRPC。Zabbix 指标进 MeterSystem → MAL。

关键设计模式小结

写文档可用的六条结论：

1. 统一入口注册，统一出口流转：几乎所有 receiver 在 start() 里 getManager().find(SharingServerModule.NAME).provider().getService(GRPCHandlerRegister.class).addHandler(myHandler)，数据最终都进 SourceReceiver.receive() → DispatcherManager.forward()。

2. 三层分工：receiver 管协议入口，analyzer 管语义解析，OAL/MAL/LAL 管指标计算。

3. 服务器复用的两种形态：

   • 默认：receiver 向 SharingServerModule 注册 handler，共享 receiver-grpc 端口（trace/jvm/clr/mesh/log/meter/event/profile/browser/pprof/async-profiler/management/config-discovery/otel）。
   • 独立服务器：协议有特殊性能/隔离需求时，receiver 自建 GRPCServer/HTTPServer，如 ebpf-grpc、als-grpc、zipkin-http、firehose-http、Zabbix 私有协议。还有"有配置就���立、没配置就 fallback 到 sharing"的弹性模式（eBPF、Envoy）。

4. 协议演进的兼容机制：*Compat.proto + *HandlerCompat 让旧版 agent 不断链。每个原生 receiver 注册一对（主 + Compat）handler。

5. OTLP 与原生的本质区别：原生是"私有 proto + 流式 + analyzer 直连"；OTLP 是"标准 proto + unary + 复用 Zipkin(trace)/MAL(metric) 链 + SPI 扩展"。OTLP trace 必须配 zipkin receiver 才能工作。

6. Receiver（push）vs Fetcher（pull）：receiver 开端口等数据，fetcher 主动连外部系统。kafka-fetcher 特殊——把"原生 receiver 的处理内核"接到"Kafka 消费入口"上，实现入口与处理解耦。

关键文件速查

接下来

数据进来了，但"算哪些指标"由谁决定？答案是 OAL/MAL/LAL 三种 DSL——它们在启动时编译成 Dispatcher，接在 receiver 产出的 Source 后面。详见 03-分析引擎与四大DSL。

心智模型
把 receiver 想成"快递驿站"——各种运输公司（gRPC/HTTP/Kafka/OTLP）把货送来，驿站只管签收、贴统一标签（转成 Source）。至于这批货要加工成什么产品（指标），那是后头工艺单（OAL/MAL/LAL）的事，驿站不操心。