笔记
Skywalking
  Java Agent
    启动流程
      加载配置
      加载插件
        加载插件文件
      配置增强并注册Transfomer
        增强 Bootstrap类
        兼容jdk9
        使用PluginFinder用匹配的插件增强匹配的类
          判断插件的见证类是否存在classpath中,以此来区分目标类的版本
            插件依赖目标类的包时使用<scope>provided</scope>,这样就不会将依赖打包进插件,此时根据应用程序自身依赖来使用对应的插件版本
          判断插件的见证方法是否存在classpath中,处理同名见证类不同方法
      启动核心服务(BootService)
        使用spi方式加载所有的BootService
          BootService生命周期
            依次调用prepare
            依次调用boot
            依次调用onComplete
          数据上报组件
            GRPCChannelManager管理与oap服务的grpc连接
              属于基础设施,负责底层的连接,供其他服务获取oap连接
                其他需要远程调用OAP的服务通过注册为GRPCChannelManager的监听者来获取连接
            TraceSegmentServiceClient链路数据上报客户端
              boot阶段
                public void boot() {
lastLogTime = System.currentTimeMillis();
segmentUplinkedCounter = 0;
segmentAbandonedCounter = 0;
carrier = new DataCarrier<>(CHANNEL_SIZE, BUFFER_SIZE, BufferStrategy.IF_POSSIBLE);
carrier.consume(this, 1);
}
                  创建上报数据缓冲层
carrier = new DataCarrier<>(CHANNEL_SIZE, BUFFER_SIZE, BufferStrategy.IF_POSSIBLE);
                    使用数据分片缓存
                      public DataCarrier(String name, String envPrefix, int channelSize, int bufferSize, BufferStrategy strategy) {
this.name = name;
// 每个分区的内存大小
bufferSize = EnvUtil.getInt(envPrefix + "_BUFFER_SIZE", bufferSize);
// 缓存的数据分区,提高并发度
channelSize = EnvUtil.getInt(envPrefix + "_CHANNEL_SIZE", channelSize);
// 数据分区
channels = new Channels<>(channelSize, bufferSize, new SimpleRollingPartitioner<T>(), strategy);
}
                        BufferStrategy // 缓存策略
public Channels(int channelSize, int bufferSize, IDataPartitioner<T> partitioner, BufferStrategy strategy) {
this.dataPartitioner = partitioner;
this.strategy = strategy;
// 数据分区
bufferChannels = new QueueBuffer[channelSize];
for (int i = 0; i < channelSize; i++) {
if (BufferStrategy.BLOCKING.equals(strategy)) {
// 根据缓存策略选择用CAS结构还是锁结构来缓存数据
bufferChannels[i] = new ArrayBlockingQueueBuffer<>(bufferSize, strategy);
} else {
bufferChannels[i] = new Buffer<>(bufferSize, strategy);
}
}
// noinspection PointlessArithmeticExpression
size = 1L * channelSize * bufferSize; // it's not pointless, it prevents numeric overflow before assigning an integer to a long
}
                        SimpleRollingPartitioner分区选择器
                          分片选择方案
                          最大重试次数,因为缓存有大小,如果没有及时采集数据或者写入数据太快都可能写入失败
                  开启数据采集线程
carrier.consume(this, 1)
                    public DataCarrier consume(IConsumer<T> consumer, int num, long consumeCycle) {
if (driver != null) {
driver.close(channels);
}
// 创建数据采集线程
driver = new ConsumeDriver<T>(this.name, this.channels, consumer, num, consumeCycle);
driver.begin(channels);
return this;
}
                      IDriver(缓存数据消费者)
                        ConsumeDriver
                          ConsumerThread[] consumerThreads
固定线程模式,线程数大于分片数时会有空闲线程
根据开启的消费线程数分配缓存的数据分片
                            public void run() {
running = true;
final List<T> consumeList = new ArrayList<T>(1500);
while (running) {
// 消费缓冲数据
if (!consume(consumeList)) {
try {
Thread.sleep(consumeCycle);
} catch (InterruptedException e) {
}
}
}
// consumer thread is going to stop
// consume the last time
consume(consumeList);
consumer.onExit();
}
                              private boolean consume(List<T> consumeList) {
for (DataSource dataSource : dataSources) {
// 将数据读取到consumeList
dataSource.obtain(consumeList);
}
if (!consumeList.isEmpty()) {
try {
// 数据消费
consumer.consume(consumeList);
} catch (Throwable t) {
consumer.onError(consumeList, t);
} finally {
consumeList.clear();
}
return true;
}
consumer.nothingToConsume();
return false;
}
                                dataSource.obtain(consumeList);
// 将数据分区中的数据读出来并清除缓存
                                TraceSegmentServiceClient作为消费者进行消费,将数据上报给oap中心
                        BulkConsumePool
目前没有用到
                          支持多数据分区
                            public DataCarrier consume(ConsumerPool consumerPool, IConsumer<T> consumer) {
driver = consumerPool;
consumerPool.add(this.name, channels, consumer);
driver.begin(channels);
return this;
}
                          优先选择负载低的消费线程
                            public synchronized void add(String name, Channels channels, IConsumer consumer) {
MultipleChannelsConsumer multipleChannelsConsumer = getLowestPayload();
multipleChannelsConsumer.addNewTarget(channels, consumer);
}
            DataCarrier上报数据缓冲层
            ContextManager上下文控制器
              TracingContext链路数据上下文
            ServiceManagementClient负责上报实例信息,能让OAP管理客户端实例
              让oap能第一时间知道连接实例的状态,并支持远程指令
1.通过AgentClassLoader#initDefaultLoader读取位于agent.jar相对路径下plugins和activations目录存储到classpath中
2.通过AgentClassLoader#findResources加载classpath目录下的所有jar包
3.读取插件jar中的所有skywalking-plugin.def文件的URL
4.将skywalking-plugin.def文件的内容读取为PluginDefine
5.将PluginDefine的定义通过AgentClassLoader反射加载为AbstractClassEnhancePluginDefine(实现埋点增强的基类)
6.将插件分为三类存放在PluginFinder中
ServiceManager.INSTANCE.findService(GRPCChannelManager.class).addChannelListener(this);