分布式追踪 APM 系统 SkyWalking 源码分析

2024-05-13 09:58

本文主要是介绍分布式追踪 APM 系统 SkyWalking 源码分析,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

1. 概述

本文主要分享 SkyWalking DataCarrier 异步处理库。
基于生产者消费者的模式,大体结构如下图:
  • 实际项目中,没有 Producer 这个类。所以本文提到的 Producer ,更多的是一种角色。
下面我们来看看整体的项目结构,如下图所示 :

2. buffer

org.skywalking.apm.commons.datacarrier.buffer 包,主要包含 Channels 、Buffer 两个类。Channels 是 Buffer 数组的封装。

2.1 Buffer

org.skywalking.apm.commons.datacarrier.buffer.Buffer ,缓存区。
  • buffer 属性,缓冲数组。Producer 保存的数据到 buffer 里。
  • strategy ,缓冲策略( org.skywalking.apm.commons.datacarrier.buffer.BufferStrategy ) 。
  • index 属性,递增位置( org.skywalking.apm.commons.datacarrier.common.AtomicRangeInteger)。
Buffer 在保存数据时,把 buffer 作为一个 “环“,使用 index 记录最后存储的位置,不断向下,循环存储到 buffer 中。通过这样的方式,带来良好的存储性能,避免扩容问题。But ,存储会存在冲突的问题:buffer 写入位置,暂未被消费,已经存在值。此时,根据不同的 BufferStrategy 进行处理。整体流程见 #save(data) 方法。
当 Buffer 被 Consumer 消费时,被调用 #obtain(start, end) 方法,获得数据并清空。为什么会带 start 、end 方法参数呢?下文揭晓答案。

2.2 Channels

org.skywalking.apm.commons.datacarrier.buffer.Channels ,内嵌多个 Buffer 的通道。
  • bufferChannels 属性,Buffer 数组。
  • dataPartitioner 属性,数据分区(org.skywalking.apm.commons.datacarrier.partition.IDataPartitioner )。
  • strategy 属性,缓冲策略( org.skywalking.apm.commons.datacarrier.buffer.BufferStrategy ) 。
Channels 在保存数据时,相比 Buffer ,从 buffer 变成了多 buffer ,因此需要先选一个buffer 。通过使用不同的 IDataPartitioner 实现类,进行 Buffer 的选择。当缓冲策略为BufferStrategy.IF_POSSIBLE 时,根据 IDataPartitioner 定义的重试次数,进行多次保存数据直到成功。整体流程见 #save(data) 方法。

3. partition

org.skywalking.apm.commons.datacarrier.partition.IDataPartitioner ,数据分配者接口。定义了如下方法:
  • #partition(total, data) 接口方法,获得数据被分配的分区位置。
  • #maxRetryCount() 接口方法,获得最大重试次数。
IDataPartitioner 目前有两个子类实现:
  • ProducerThreadPartitioner ,基于线程编号分配策略的数据分配者实现类。
  • SimpleRollingPartitioner ,基于顺序分配策略的数据分配者实现类。

4. consumer

org.skywalking.apm.commons.datacarrier.consumer 包,主要包含 ConsumerPool 、ConsumerThread 、IConsumer 三个类。
  • ConsumerThread 使用 IConsumer ,消费数据
  • ConsumerPool 是 ConsumerThread 的线程池封装

4.1 IConsumer

org.skywalking.apm.commons.datacarrier.consumer.IConsumer ,消费者接口。定义了如下方法:
  • #init() 接口方法,初始化消费者。
  • #consume(List<T>) 接口方法,批量消费消息。
  • #onError(List<T>, Throwable) 接口方法,处理当消费发生异常。
  • #onExit() 接口方法,处理当消费结束。此处的结束时,ConsumerThread 关闭。
我们在使用时,自定义 Consumer 类,实现 IConsumer 接口。例如:RemoteMessageConsumer 。

4.2 ConsumerThread

org.skywalking.apm.commons.datacarrier.consumer.ConsumerThread ,继承 java.lang.Thread ,消费线程。
  • running 属性,是否运行中。
  • consumer 属性,消费者对象。
  • dataSources 属性,消费消息的数据源( DataSource )数组。一个 ConsumerThread ,可以消费多个 Buffer ,并且单个 Buffer 消费的分区范围可配置,即一个 Buffer 可以被多个 ConsumerThread 同时无冲突的消费。在 「4.3 ConsumerPool」 详细解析 ConsumerThread 分配 Buffer 的方式。
    • #addDataSource(sourceBuffer, start, end) 方法,添加 Buffer 部分范围。
    • #addDataSource(sourceBuffer) 方法,添加 Buffer 全部范围。
#run() 实现方法,不断、批量的消费数据。代码如下:
  • 第 78 至 88 行:不断消费,直到线程关闭( #shutdown() )。
    • 第 80 行:调用 #consume() 方法,批量消费数据。
    • 第 82 至 87 行:当未消费到数据,说明 dataSources 为空,等待 20 ms ,避免 CPU 空跑。
  • 第 93 行:当线程关闭,调用 #consume() 方法,消费完 dataSources 剩余的数据。
  • 第 95 行:调用 IConsumer#onExit() 方法,处理当消费结束。
#consume() 方法,批量消费数据。代码如下:
  • 第 107 至 117 行:从 dataSources 中,获取要消费的数据。
  • 第 120 至 126 行:当有数据可消费时,调用 IConsumer#consume(List<T>) 方法。当消费发生异常时,调用 IConsumer#onError(List<T>, Throwable) 方法。
  • 第 127 行:返回是否有消费数据。

4.3 ConsumerPool

org.skywalking.apm.commons.datacarrier.consumer.ConsumerPool ,消费者池,提供了对 Channels 启动指定数量的 ConsumerThread 进行消费。
  • running 属性,是否运行中。
  • consumerThreads 属性,ConsumerThread 数组,通过构造方法的 num 参数进行指定。
  • channels 属性,数据通道。
  • lock 属性,锁。保证 ConsumerPool 启动或关闭时的线程安全。
#begin() 方法,启动 ConsumerPool ,进行数据消费。代码如下:
  • 第 97 至 99 行:正在运行中,直接返回。
  • 第 101 行:获得锁。
  • 第 104 行:调用 #allocateBuffer2Thread() 方法,将 channels 的多个 Buffer ,分配给consumerThreads 的多个 ConsumerThread。
  • 第 107 至 109 行:启动每个 ConsumerThread ,开始消费。
  • 第 112 行:标记正在运行中。
  • 第 114 行:释放锁。
close() 方法,关闭 ConsumerPool 。代码如下:
  • 第 168 行:获得锁。
  • 第 169 行:标记不在运行中。
  • 第 170 至 172 行:关闭每个 ConsumerThread ,结束消费。
  • 第 174 行:释放锁。

#allocateBuffer2Thread() 方法,将 channels 的多个 Buffer ,分配给 consumerThreads 的多个ConsumerThread。一共会有三种情况:
  • Buffer 数量等于 ConsumerThread 数量,这个十分好分配,一比一。
  • Buffer 数量大于 ConsumerThread 数量,那么按照 Buffer 数量 % ConsumerThread 数量进行分组,分配给 ConsumerThread ,如下图所示:
  • Buffer 数量大于 ConsumerThread 数量,那么按照 ConsumerThread 数量 % Buffer 数量进行分组,分配给 Buffer 。其中,一个 Buffer 会被均分给多个 ConsumerThread ,如下图所示:
    • 这个就是为什么 Buffer 里面,提供了 Buffer#obtain(start, end) 方法的原因。

4. DataCarrier

org.skywalking.apm.commons.datacarrier.DataCarrier ,DataCarrier 异步处理库的入口程序。通过创建 DataCarrier 对象,使用生产者消费者的模式,执行异步执行逻辑。
构造方法 ,代码如下:
  • channels 属性,数据通道。在构造方法中,我们可以看到默认使用 SimpleRollingPartitioner 作为数据分区分配者,使用 BufferStrategy.BLOCKING 作为缓冲策略。
    • #setPartitioner(IDataPartitioner) 方法,设置数据分区分配者。
    • #setBufferStrategy(BufferStrategy) 方法,设置缓冲策略。
  • channelSize 方法参数,通道大小。
  • bufferSize 方法参数,缓冲区大小。
设置消费者和消费线程数量:
  • #consume(Class<? extends IConsumer<T>>, num)
  • #consume(IConsumer<T>, num)
生产消息
  • #produce(data)
关闭消费
  • #shutdownConsumers()

这篇关于分布式追踪 APM 系统 SkyWalking 源码分析的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/985386

相关文章

怎样通过分析GC日志来定位Java进程的内存问题

《怎样通过分析GC日志来定位Java进程的内存问题》:本文主要介绍怎样通过分析GC日志来定位Java进程的内存问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、GC 日志基础配置1. 启用详细 GC 日志2. 不同收集器的日志格式二、关键指标与分析维度1.

MySQL中的表连接原理分析

《MySQL中的表连接原理分析》:本文主要介绍MySQL中的表连接原理分析,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录1、背景2、环境3、表连接原理【1】驱动表和被驱动表【2】内连接【3】外连接【4编程】嵌套循环连接【5】join buffer4、总结1、背景

MySQL追踪数据库表更新操作来源的全面指南

《MySQL追踪数据库表更新操作来源的全面指南》本文将以一个具体问题为例,如何监测哪个IP来源对数据库表statistics_test进行了UPDATE操作,文内探讨了多种方法,并提供了详细的代码... 目录引言1. 为什么需要监控数据库更新操作2. 方法1:启用数据库审计日志(1)mysql/mariad

python中Hash使用场景分析

《python中Hash使用场景分析》Python的hash()函数用于获取对象哈希值,常用于字典和集合,不可变类型可哈希,可变类型不可,常见算法包括除法、乘法、平方取中和随机数哈希,各有优缺点,需根... 目录python中的 Hash除法哈希算法乘法哈希算法平方取中法随机数哈希算法小结在Python中,

Java Stream的distinct去重原理分析

《JavaStream的distinct去重原理分析》Javastream中的distinct方法用于去除流中的重复元素,它返回一个包含过滤后唯一元素的新流,该方法会根据元素的hashcode和eq... 目录一、distinct 的基础用法与核心特性二、distinct 的底层实现原理1. 顺序流中的去重

linux重启命令有哪些? 7个实用的Linux系统重启命令汇总

《linux重启命令有哪些?7个实用的Linux系统重启命令汇总》Linux系统提供了多种重启命令,常用的包括shutdown-r、reboot、init6等,不同命令适用于不同场景,本文将详细... 在管理和维护 linux 服务器时,完成系统更新、故障排查或日常维护后,重启系统往往是必不可少的步骤。本文

关于MyISAM和InnoDB对比分析

《关于MyISAM和InnoDB对比分析》:本文主要介绍关于MyISAM和InnoDB对比分析,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录开篇:从交通规则看存储引擎选择理解存储引擎的基本概念技术原理对比1. 事务支持:ACID的守护者2. 锁机制:并发控制的艺

Mac系统下卸载JAVA和JDK的步骤

《Mac系统下卸载JAVA和JDK的步骤》JDK是Java语言的软件开发工具包,它提供了开发和运行Java应用程序所需的工具、库和资源,:本文主要介绍Mac系统下卸载JAVA和JDK的相关资料,需... 目录1. 卸载系统自带的 Java 版本检查当前 Java 版本通过命令卸载系统 Java2. 卸载自定

MyBatis Plus 中 update_time 字段自动填充失效的原因分析及解决方案(最新整理)

《MyBatisPlus中update_time字段自动填充失效的原因分析及解决方案(最新整理)》在使用MyBatisPlus时,通常我们会在数据库表中设置create_time和update... 目录前言一、问题现象二、原因分析三、总结:常见原因与解决方法对照表四、推荐写法前言在使用 MyBATis

Python主动抛出异常的各种用法和场景分析

《Python主动抛出异常的各种用法和场景分析》在Python中,我们不仅可以捕获和处理异常,还可以主动抛出异常,也就是以类的方式自定义错误的类型和提示信息,这在编程中非常有用,下面我将详细解释主动抛... 目录一、为什么要主动抛出异常?二、基本语法:raise关键字基本示例三、raise的多种用法1. 抛