下一篇【第60篇】探针注册通信扩展——基于HTTP的SPI注册实现完全指南上一篇【第62篇】通信扩展最佳实践——gRPC/HTTP/Kafka全景对比与选型决策一、为什么是Kafka先问一个扎心的问题gRPC那么好为什么还要用Kafka答案取决于你的部署规模和数据量。来看两种典型场景------------------------------------------------------------------ | 两种通信模式的适用场景 | ------------------------------------------------------------------ | | | 场景A中小规模100个Agent实例 | | ┌──────────────────────────────────────┐ | | │ Agent → gRPC → OAP │ ✓ 简单直接 | | │ │ ✓ 零额外组件 | | │ Agent数量少OAP压力可控 │ ✓ 延迟低 | | │ 直接gRPC通信完全够用 │ | | └──────────────────────────────────────┘ | | | | 场景B大规模部署500个Agent实例 | | ┌──────────────────────────────────────┐ | | │ Agent → Kafka → OAP │ ✓ 削峰填谷 | | │ │ ✓ OAP 可水平扩展 | | │ Agent数量多数据量大 │ ✓ 消息可持久化 | | │ Kafka做缓冲避免OAP被打爆 │ ✓ 消费可回溯 | | └──────────────────────────────────────┘ | | | | Agent数量 推荐方案 | | ───────────────────────────────────────── | | 1-100 直接gRPC | | 100-500 gRPC 适当调参 | | 500-2000 Kafka Reporter | | 2000 Kafka OAP集群 ES集群 | | | ------------------------------------------------------------------Kafka的优势在大规模场景中非常明显解耦Agent不关心OAP在不在只管往Kafka扔数据削峰Trace数据有波峰波谷Kafka天然抗波动可回溯Kafka可按时间范围回放数据方便故障排查可靠性Kafka的多副本机制保证数据不丢二、Kafka Reporter的整体架构------------------------------------------------------------------ | SkyWalking Kafka 数据流全景 | ------------------------------------------------------------------ | | | ┌───────────────────────────┐ | | │ App JVM 1 │ | | │ ┌─────────────────────┐ │ | | │ │ SkyWalking Agent │ │ | | │ │ ┌─────────────────┐ │ │ | | │ │ │ Trace Segment │ │ │ | | │ │ │ Collector CPU3 │ │ │ | | │ │ │ Memory521MB │ │ │ | | │ │ └────────┬────────┘ │ │ | | │ │ ↓ │ │ | | │ │ ┌─────────────────┐ │ │ | | │ │ │ Kafka Reporter │ │ │ | | │ │ │ (KafkaProducer) │ │──┼───────→ Kafka Broker | | │ │ └─────────────────┘ │ │ Topic: | | │ └─────────────────────┘ │ skywalking-segments | | └───────────────────────────┘ | | | | ┌───────────────────────────┐ | | │ App JVM 2 │ | | │ ┌─────────────────────┐ │ | | │ │ Kafka Reporter │──┼───────→ 同上 | | │ └─────────────────────┘ │ | | └───────────────────────────┘ | | | | ┌───────────────────────────┐ | | │ App JVM N │ | | │ ┌─────────────────────┐ │ | | │ │ Kafka Reporter │──┼───────→ 同上 | | │ └─────────────────────┘ │ | | └───────────────────────────┘ | | | | ┌──────────────────────────────────────────────────┐ | | │ Kafka Cluster │ | | │ ┌────────────────────────────────────────────┐ │ | | │ │ Topic: skywalking-segments (Partition x N) │ │ | | │ │ Topic: skywalking-metrics │ │ | | │ │ Topic: skywalking-profilings │ │ | | │ │ Topic: skywalking-managements │ │ | | │ │ Topic: skywalking-logs │ │ | | │ └────────────────────────────────────────────┘ │ | | └──────────────────────┬───────────────────────────┘ | | │ | | ┌─────────────┼─────────────┐ | | │ │ │ | | ↓ ↓ ↓ | | ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ | | │ OAP Server 1│ │ OAP Server 2│ │ OAP Server 3│ | | │KafkaFetcher │ │KafkaFetcher │ │KafkaFetcher │ | | │Analyzer │ │Analyzer │ │Analyzer │ | | └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ | | │ │ │ | | └───────────────┼───────────────┘ | | │ | | ↓ | | ┌──────────────────┐ | | │ Elasticsearch │ | | │ (持久化存储) │ | | └──────────────────┘ | | | ------------------------------------------------------------------三、Agent端 —— Kafka Reporter配置SkyWalking 8.x版本官方就已经内置了Kafka Reporter。你不需要自己写代码只需要配置即可。3.1 基础配置# agent/config/agent.config# Kafka Reporter 配置 # 1. 指定使用Kafka作为Trace数据上报通道plugin.kafka.topic_segmentskywalking-segments# 2. Kafka集群地址plugin.kafka.bootstrap_servers127.0.0.1:9092,127.0.0.1:9093,127.0.0.1:9094# 3. 其他数据类型也走Kafkaplugin.kafka.topic_metricsskywalking-metrics plugin.kafka.topic_profilingsskywalking-profilings plugin.kafka.topic_managementsskywalking-managements plugin.kafka.topic_logsskywalking-logs# 4. Producer配置plugin.kafka.producer_config.max_request_size104857600# 100MBplugin.kafka.producer_config.batch_size16384 plugin.kafka.producer_config.linger_ms10 plugin.kafka.producer_config.compression_typelz4 plugin.kafka.producer_config.acks-1# all replicasplugin.kafka.producer_config.retries3 plugin.kafka.producer_config.max_in_flight_requests_per_connection1# 5. 命名空间多环境隔离plugin.kafka.namespaceproduction3.2 高级生产配置详解# Kafka Producer 完整调优参数plugin.kafka.producer_config:# 可靠性 acks:all# 等待所有副本确认最高可靠retries:2147483647# 最大重试次数enable.idempotence:true# 幂等生产防止重复# 性能 batch.size:131072# 批次大小128KBlinger.ms:5# 批次等待时间buffer.memory:67108864# 缓冲区大小64MBcompression.type:lz4# 压缩算法lz4/snappy/gzip# 超时 request.timeout.ms:30000# 请求超时delivery.timeout.ms:120000# 投递超时max.block.ms:60000# 缓冲区满时的阻塞时间3.3 Kafka Reporter的实现原理// Kafka Reporter的核心逻辑简化自SkyWalking源码publicclassKafkaTraceSegmentServiceClientimplementsTracingContextListener,GRPCChannelListener{privateKafkaProducerString,Bytesproducer;// 当有新Trace Segment产生时TracingContext会回调此方法OverridepublicvoidafterFinished(TraceSegmenttraceSegment){if(traceSegment.isSkipAnalysis()){return;// 跳过不需要分析的Segment}// 将TraceSegment序列化为ProtobufUpstreamSegmentupstreamtraceSegment.transform();BytesdataBytes.wrap(upstream.toByteArray());// 构建Kafka消息StringtopicString.format(%s-%s,config.getTopicSegment(),config.getNamespace());// 以TraceId作为Key保证同一Trace的所有Segment去同一个PartitionStringtraceIdtraceSegment.getRelatedGlobalTrace().getId();ProducerRecordString,BytesrecordnewProducerRecord(topic,traceId,data);// 异步发送producer.send(record,newCallback(){OverridepublicvoidonCompletion(RecordMetadatametadata,Exceptione){if(e!null){LOGGER.warn(Failed to send trace segment: {},e.getMessage());}}});}// 初始化Kafka ProducerprivatevoidinitProducer(KafkaReporterConfigconfig){PropertiespropsnewProperties();props.put(bootstrap.servers,config.getBootstrapServers());// 合并自定义Producer配置if(config.getProducerConfig()!null){props.putAll(config.getProducerConfig());}props.put(key.serializer,org.apache.kafka.common.serialization.StringSerializer);props.put(value.serializer,org.apache.kafka.common.serialization.BytesSerializer);this.producernewKafkaProducer(props);}}四、OAP端 —— Kafka Fetcher配置Agent把数据发到Kafka了OAP怎么消费呢4.1 application.yml配置# oap-server/config/application.yml# Kafka Fetcher 配置 core:default:# 指定从Kafka消费数据selector:${SW_CLUSTER:standalone}# 不使用gRPC接收gRPCHost:${SW_CORE_GRPC_HOST:0.0.0.0}gRPCPort:${SW_CORE_GRPC_PORT:11800}# Kafka 消费者配置kafka:bootstrapServers:${SW_KAFKA_FETCHER_SERVERS:127.0.0.1:9092}# 消费配置 # 消费者组IDgroupId:${SW_KAFKA_FETCHER_GROUP_ID:skywalking-oap}# 每个Topic的Partition数量用于分配消费者partitions:${SW_KAFKA_FETCHER_PARTITIONS:3}# 每批消费的消息数batchSize:${SW_KAFKA_FETCHER_BATCH_SIZE:1000}# 轮询间隔pollInterval:${SW_KAFKA_FETCHER_POLL_INTERVAL:1000}# Topic映射 consumers:-topic:${SW_KAFKA_TOPIC_SEGMENT:skywalking-segments}processor:trace-topic:${SW_KAFKA_TOPIC_METRICS:skywalking-metrics}processor:metrics-topic:${SW_KAFKA_TOPIC_PROFILING:skywalking-profilings}processor:profiling-topic:${SW_KAFKA_TOPIC_MANAGEMENT:skywalking-managements}processor:management-topic:${SW_KAFKA_TOPIC_LOGS:skywalking-logs}processor:logs# 消费者高级配置 kafkaConsumerConfig:enable.auto.commit:trueauto.commit.interval.ms:5000session.timeout.ms:30000max.poll.records:500fetch.max.bytes:52428800# 50MBmax.partition.fetch.bytes:10485760# 10MB4.2 Kafka Fetcher的实现逻辑// KafkaFetcherHandlerRegister.java (简化核心逻辑)publicclassKafkaFetcherHandlerRegisterimplementsFetcherHandlerRegister{privatefinalMapString,KafkaConsumerString,BytesconsumersnewConcurrentHashMap();Overridepublicvoidregister(FetcherConfigconfig){KafkaFetcherConfigkafkaConfig(KafkaFetcherConfig)config;for(ConsumerConfigconsumerConfig:kafkaConfig.getConsumers()){// 为每个Topic创建一个消费者KafkaConsumerString,BytesconsumercreateConsumer(kafkaConfig,consumerConfig);consumers.put(consumerConfig.getTopic(),consumer);// 启动消费者线程startConsumerThread(consumer,consumerConfig);}}privatevoidstartConsumerThread(KafkaConsumerString,Bytesconsumer,ConsumerConfigconfig){ThreadconsumerThreadnewThread(()-{consumer.subscribe(Collections.singletonList(config.getTopic()));while(!stopped){ConsumerRecordsString,Bytesrecordsconsumer.poll(Duration.ofMillis(1000));for(ConsumerRecordString,Bytesrecord:records){// 根据processor类型分发给不同的处理器switch(config.getProcessor()){casetrace:traceAnalyzer.analyze(record.key(),record.value());break;casemetrics:metricsAggregator.aggregate(record.value());break;caseprofiling:profilingHandler.handle(record.value());break;}}// 提交offsetconsumer.commitAsync();}},kafka-fetcher-config.getTopic());consumerThread.setDaemon(true);consumerThread.start();}}五、完整的端到端数据流把Agent和OAP都配好后数据是如何流动的------------------------------------------------------------------ | Trace数据从Agent到ES的完整生命周期 | ------------------------------------------------------------------ | | | Step 1: Agent采集 | | ┌────────────────────────────────────────────┐ | | │ TracingContext生成TraceSegment │ | | │ ↓ │ | | │ 序列化为UpstreamSegment (Protobuf) │ | | │ ↓ │ | | │ KafkaReporter.send() │ | | └────────────────────┬───────────────────────┘ | | │ | | Step 2: Kafka传输 | | ┌────────────────────▼───────────────────────┐ | | │ Topic: skywalking-segments-prod │ | | │ Key: traceId → 路由到固定Partition │ | | │ Value: UpstreamSegment (二进制) │ | | │ ↓ │ | | │ 副本同步 → Leader确认 │ | | └────────────────────┬───────────────────────┘ | | │ | | Step 3: OAP消费 | | ┌────────────────────▼───────────────────────┐ | | │ KafkaFetcher.poll() → 批量拉取 │ | | │ ↓ │ | | │ 反序列化UpstreamSegment │ | | │ ↓ │ | | │ TraceAnalyzer.doAnalysis() │ | | │ ├── Span聚合 │ | | │ ├── 调用链构建 │ | | │ ├── 指标计算(OAL) │ | | │ └── 拓扑推断 │ | | │ ↓ │ | | │ 写入Elasticsearch │ | | └────────────────────────────────────────────┘ | | | | Step 4: 查询展示 | | ┌────────────────────────────────────────────┐ | | │ SkyWalking UI → GraphQL API → ES查询 │ | | │ ↓ │ | | │ 展示Trace详情/拓扑图/指标图表 │ | | └────────────────────────────────────────────┘ | | | ------------------------------------------------------------------六、性能影响与注意事项6.1 性能对比# 不同通信方式的性能特征gRPC (长连接,Protobuf):延迟:~5ms (同机房)吞吐:~10000 segments/秒 (单OAP)CPU:Agent端 1%,OAP端 中等内存:Agent端 ~10MB,OAP端 ~200MB Kafka (异步,批量):延迟:~10-50ms (消息队列缓冲)吞吐:~50000 segments/秒 (单OAP消费)CPU:Agent端 1%,OAP端 低内存:Agent端 ~20MB (Producer缓冲区),OAP端 ~300MB (Consumer缓冲区) HTTPS (短连接,JSON):延迟:~20ms吞吐:~3000 segments/秒CPU:Agent端 低,OAP端 高 (JSON解析)内存:较低6.2 Kafka环境检查清单# 部署前验证清单# 1. 检查Kafka集群状态kafka-topics.sh --bootstrap-server localhost:9092--list# 2. 验证Topic创建Partition数量 OAP实例数kafka-topics.sh --bootstrap-server localhost:9092\--create\--topicskywalking-segments\--partitions6\--replication-factor3\--configretention.ms172800000# 保留2天# 3. 检查消费者的Lagkafka-consumer-groups.sh --bootstrap-server localhost:9092\--groupskywalking-oap--describe# 4. 监控Topic的写入速率kafka-run-class.sh kafka.tools.GetOffsetShell\--broker-list localhost:9092\--topicskywalking-segments--time-1# 5. 验证网络连通性telnet kafka-broker-190926.3 常见问题与解决------------------------------------------------------------------ Kafka集成常见问题 ------------------------------------------------------------------ | | | 问题1: Topic不存在 | | 原因: Kafka未启用auto.create.topics.enable | | 解决: 手动创建Topic或启用自动创建 | | | | 问题2: OAP不消费数据 | | 原因: | | - OAP配置中未启用KafkaFetcher | | - 消费者Group配置冲突 | | - Offset异常未从头消费 | | 解决: | | - 检查SW_CORE环境变量 | | - 重置Consumer Group Offset | | | | 问题3: 数据延迟严重 | | 原因: | | - Kafka Partition不够OAP实例并行度受限 | | - OAP处理能力不足 | | - 网络带宽不足 | | 解决: 增加Partition数量 增加OAP实例 | | | ------------------------------------------------------------------七、总结用Kafka作为SkyWalking的数据上报通道是大规模生产环境的最佳实践方面要点适用场景Agent实例数 500或需要缓冲/削峰Agent配置指定bootstrap_servers和各topic名称OAP配置配置KafkaFetcher消费者映射processor监控重点Consumer Lag、数据延迟、错误率运维要点确保Partition OAP实例数定期清Topic下一篇文章将汇总所有通信方案的对比和最佳实践。下一篇【第60篇】探针注册通信扩展——基于HTTP的SPI注册实现完全指南上一篇【第62篇】通信扩展最佳实践——gRPC/HTTP/Kafka全景对比与选型决策