转载

Skywalking源码分析指北第十一篇——DataCarrier

Skywalking源码分析指北第十一篇——DataCarrier

前情回顾

Skywalking源码分析指北第十一篇——DataCarrier

前文介绍了 Skywalking 中提供的一些插件，例如，Tomcat 插件、Dubbo 插件等等，同时也简单介绍了 Tomcat 的架构以及 Dubbo Filter 相关的基础知识。从本节开始，我们来看 Skywalking Agent 发送 Trace 数据的相关实现。

Skywalking源码分析指北第十一篇——DataCarrier

DataCarrier

Skywalking源码分析指北第十一篇——DataCarrier

DataCarrier 是一个轻量级的生产者-消费者模式的实现库， Skywalking Agent 在收集到 Trace 数据之后，会先将 Trace 数据写入到 DataCarrier 中缓存，然后由后台线程定时发送到 Skywalking 服务端。呦吼，和前面介绍的 Dubbo MonitorFilter 的实现原理类似啊，emmm，其实多数涉及到大量网络传输的场景都会这么设计：先本地缓存聚合，然后定时发送。

DataCarrier 实现在 Skywalking 中是一个单独的模块，如下图所示，行吧，这么多？！有的看了：

Skywalking源码分析指北第十一篇——DataCarrier

首先来看 Buffer 这个类，它是一个环形缓冲区，是整个 DataCarrier 最底层的一个类，其核心字段如下：

// 真正缓存数据的数组
private final Object[] buffer;
// BufferStrategy指定了缓冲区满了之后的填充策略，是个枚举，有三个可选项：
// 1.BLOCKING（默认），写入线程阻塞等待，直到数据被消费为止
// 2.OVERRIDE，复写旧数据，旧数据被丢弃
// 3.IF_POSSIBLE，如果无法写入则直接返回false，由应用程序判断如何处理
private BufferStrategy strategy;
// 环形的指针，后面展开详细说其实现
private AtomicRangeInteger index;
// 回调函数集合
private List<QueueBlockingCallback<T>> callbacks;

这里的 AtomicRangeInteger 是环形指针，我们可以指定其中的 value字段（AtomicInteger类型）从 start值（int类型）开始递增，当 value 递增到 end值（int类型）时，value字段会被重置为 start值，从而实现环形指正的效果。

Buffer.save() 方法负责向当前 Buffer 缓冲区中填充数据，实现如下：

boolean save(T data) {
    int i = index.getAndIncrement();
     // 如果当前位置空着，则不用走strategy指定冲突处理策略，直接填充即可，皆大欢喜
    if (buffer[i] != null) {
        switch (strategy) {
            case BLOCKING:
                boolean isFirstTimeBlocking = true;
                while (buffer[i] != null) {// 自选等待当前位置被释放
                    if (isFirstTimeBlocking) { // 第一个阻塞在当前位置的线程，会通知所有注册的Callback
                        isFirstTimeBlocking = false;
                        for (QueueBlockingCallback<T> callback : callbacks) {
                            callback.notify(data);
                        }
                    }
                    Thread.sleep(1L); // sleep 1s
                }
                break;
            case IF_POSSIBLE:
                return false; // 直接返回false
            case OVERRIDE: // 啥都不做，直接走下面的赋值，覆盖旧数据
            default:
        }
    }
    buffer[i] = data; // 向当前位置填充数据
    return true;
}

个人感觉： BLOCKING 策略下，Buffer 在写满之后，index 发生循环， 可能会出现两个线程同时等待写入一个位置的情况，有并发问题啊，会丢数据~~~看官大佬们可以自己思考一下~~~

生产者用的是 save() 方法，消费者用的是 Buffer.obtain() 方法：

public LinkedList<T> obtain(int start, int end) {
    LinkedList<T> result = new LinkedList<T>();
    // 将 start~end 之间的元素返回，消费者消费这个result集合就行了
    for (int i = start; i < end; i++) {
        if (buffer[i] != null) {
            result.add((T)buffer[i]);
            buffer[i] = null;
        }
    }
    return result;
}

Channels 是另一个比较基础的类，它底层封装了多个 Buffer 实例以及一个分区选择器，字段如下所示：

// 底层封装的多个环形缓冲区
private final Buffer<T>[] bufferChannels;
// 分区选择器，由它确定数据到底写入哪个缓冲区
private IDataPartitioner<T> dataPartitioner;
// 缓冲区满时的写入策略
private BufferStrategy strategy;
// 整个Channels的总大小，实际就是所有底层Buffer的和
private final long size;

IDataPartitioner 提供了类似于 Kafka 的分区功能，当数据并行写入的时候，由分区选择器定将数据写入哪个分区，这样就可以有效降低并发导致的锁(或CAS)竞争，降低写入压力，可提高效率。IDataPartitioner 接口有两个实现，如下下图所示：

Skywalking源码分析指北第十一篇——DataCarrier

  • ProducerThreadPartitioner会根据写入线程的ID进行选择，这样可以保证相同的线程号写入的数据都在一个分区中。

  • SimpleRollingPartitioner简单循环自增选择器，使用无锁整型（volatile修饰）的自增，顺序选择线程号，在高负载时，会产生批量连续写入一个分区的效果，在中等负载情况下，提供较好性能。

下面来看 IConsumer 接口， DataCarrier 的消费者逻辑需要封装在其中。IConsumer 接口定义了消费者的基本行为：

public interface IConsumer<T> {
    void init();    // 初始化消费者。
    void consume(List<T> data);    // 批量消费消息
    void onError(List<T> data, Throwable t);     // 处理当消费发生异常。
    void onExit();    // 处理当消费结束，关闭ConsumerThread
}

后面即将介绍的  TraceSegmentServiceClient 类就实现了 IConsumer 接口，后面展开说。

ConsumerThread 继承了Thread，表示的是消费者线程，其核心字段如下：

private volatile boolean running; // 当前线程运行状态
private IConsumer<T> consumer; // 消费者逻辑
// 对Buffer的封装，负责从Buffer取数据
// 一个 ConsumerThread可以消费多个 Buffer
private List<DataSource> dataSources; 
private long consumeCycle; // 两次执行 IConsumer.consume()方法的时间间隔

ConsumerThread.run() 方法会循环 consume() 方法：

public void run() {
    running = true;
    while (running) { // 判断当前线程的状态
        boolean hasData = consume(); // 消费数据
        if (!hasData) {
            Thread.sleep(consumeCycle); // 没数据的话，等一下再消费
        }
    }
    consume();
    consumer.onExit(); // 退出前的处理
}

ConsumerThread.consume() 方法是消费的核心：

private boolean consume() {
    boolean hasData = false;
    LinkedList<T> consumeList = new LinkedList<T>();
    for (DataSource dataSource : dataSources) {
        // DataSource.obtain()方法是对Buffer.obtain()方法的封装
        LinkedList<T> data = dataSource.obtain();
        if (data.size() == 0) { continue; }
        consumeList.addAll(data); // 将所有待消费的数据转存到consumeList
        hasData = true; // 标记此次消费是否有数据
    }

    if (consumeList.size() > 0) {
        try {
            // 消费
            consumer.consume(consumeList);
        } catch (Throwable t) {// 消费过程中出现异常的时候
            consumer.onError(consumeList, t);
        }
    }
    return hasData;
}

另一个消费者线程是 MultipleChannelsConsumer，与 ConsumerThread 的区别在于:

  • ConsumerThread 处理的是一个确定的 IConsumer 消费一个 Channel 中的多个 Buffer。

  • MultipleChannelsConsumer 可以处理多组 Group，每个 Group 都是一个IConsumer+一个 Channels。

先来看 MultipleChannelsConsumer 的核心字段：

private volatile boolean running; // 线程状态
// 一个Channels以及其对应的IConsumer封装成一个Group实例
private volatile ArrayList<Group> consumeTargets;

MultipleChannelsConsumer.consume()方法消费的依然是单个IConsumer，这里不再展开分析。在  MultipleChannelsConsumer.run()方法中会循环每个 Group： 

public void run() {
    running = true;
    while (running) {
        boolean hasData = false;
        for (Group target : consumeTargets) {
            // 调用consume()方法，处理单个 Group
            hasData = hasData || consume(target);
        }
        if (!hasData) { // 没数据，则等待一会儿
            Thread.sleep(consumeCycle);
        }
    }
    // MultipleChannelsConsumer线程关闭之前，再清理一下
    for (Group target : consumeTargets) {
        consume(target);
        target.consumer.onExit();
    }
}

在  MultipleChannelsConsumer.addNewTarget() 方法中会添加新的 Group。这里用了 Copy-on-Write，因为在添加的过程中，MultipleChannelsConsumer线程可能正在循环处理 consumeTargets 集合（这也是为什么 consumeTargets 用 volatile 修饰的原因）： 

public void addNewTarget(Channels channels, IConsumer consumer) {
    Group group = new Group(channels, consumer);
    ArrayList<Group> newList = new ArrayList<Group>();
    for (Group target : consumeTargets) {
        newList.add(target);
    }
    newList.add(group);
    consumeTargets = newList;
    size += channels.size();
}

IDriver 负责将 ConsumerThread 和 IConsumer 集成到一起，其实现类如下：

Skywalking源码分析指北第十一篇——DataCarrier

先来看 ConsumeDriver，核心字段如下：

private ConsumerThread[] consumerThreads; // 消费线程
private Channels<T> channels; // 被消费的缓冲区

在 ConsumeDriver 的构造方法中，会初始化上述两个集合：

private ConsumeDriver(String name, Channels<T> channels, Class<? extends IConsumer<T>> consumerClass, int num,
        long consumeCycle) {
    running = false;
    this.channels = channels; // 初始化channels集合和consumerThreads集合
    consumerThreads = new ConsumerThread[num];
    lock = new ReentrantLock();
    for (int i = 0; i < num; i++) {
        consumerThreads[i] = new ConsumerThread("DataCarrier." + name + ".Consumser." + i + ".Thread", 
            // 反射方式实例化 IConsumer对象
            getNewConsumerInstance(consumerClass),  
            consumeCycle);
        consumerThreads[i].setDaemon(true);
    }
}

完成初始化之后，要调用 ConsumeDriver.begin() 方法，其中会根据 Buffer 的数量以及ConsumerThread 线程数进行分配：

  • 如果 Buffer 个数较多，则一个 ConsumerThread 线程需要处理多个 Buffer。

  • 如果 ConsumerThread 线程数较多，则一个 Buffer 被多个 ConsumerThread 线程处理（每个 ConsumerThread 线程负责消费这个 Buffer 的一段）。

  • 如果两者数量正好相同，则是一对一关系。

逻辑虽然很简单，但是具体的分配代码比较长，这里就不粘贴了。

BulkConsumePool 是 IDriver 接口的另一个实现，其中的 allConsumers 字段（List<MultipleChannelsConsumer>类型）记录了当前启动的  MultipleChannelsConsumer 线程，在  BulkConsumePool 的构造方法中会根据配置初始化该集合： 

public BulkConsumePool(String name, int size, long consumeCycle) {
    size = EnvUtil.getInt(name + "_THREAD", size); // 配置的线程个数
    allConsumers = new ArrayList<MultipleChannelsConsumer>(size);
    for (int i = 0; i < size; i++) {
        // 创建 MultipleChannelsConsumer对象
        MultipleChannelsConsumer multipleChannelsConsumer = new MultipleChannelsConsumer("...", consumeCycle);
        multipleChannelsConsumer.setDaemon(true); // 后台线程
        allConsumers.add(multipleChannelsConsumer);
    }
}

BulkConsumePool.add() 方法提供了添加新 Group的功能：

synchronized public void add(String name, Channels channels, IConsumer consumer) {
    // 查找负载最低的 MultipleChannelsConsumer线程，其实就是负责的 Group个数最少的线程
    MultipleChannelsConsumer multipleChannelsConsumer = getLowestPayload();
    // 添加新的 Group，Copy-on-Write
    multipleChannelsConsumer.addNewTarget(channels, consumer);
}

最后来看 DataCarrier ，它是整个 DataCarrier 库的入口和门面，其核心字段如下：

private String name; // 当前 DataCarrier实例的名称
private final int bufferSize; // Buffer大小
private final int channelSize;// Buffer的个数
private Channels<T> channels; // Channels对象
private IDriver driver; // ConsumeDriver对象

在 DataCarrier 的构造方法中会初始化 Channels 对象，默认使用 SimpleRollingPartitioner 以及 BLOCKING 策略，太简单，不展开了。

DataCarrier.produce() 方法实际上是调用 Channels.save()方法，实现数据写入，太简单，不展开了。

在 DataCarrier.consume()方法中，会初始化 ConsumeDriver 对象并调用 begin() 方法，启动 ConsumeThread 线程：

public DataCarrier consume(Class<? extends IConsumer<T>> consumerClass, int num, long consumeCycle) {
    // 初始化 ConsumeDriver
    driver = new ConsumeDriver<T>(this.name, this.channels, consumerClass, num, consumeCycle);
    // 启动 ConsumerThread线程
    driver.begin(channels);
    return this;
}

Skywalking源码分析指北第十一篇——DataCarrier

总结

Skywalking源码分析指北第十一篇——DataCarrier

本节主要介绍了  Da taCarrier  这个轻量级的生产者-消费者模式的实现库，也没啥可总结的，后面继续介绍 Skywalking 的 Trace 收集以及发送实现。

扫描下图二维码，关注【程序员吴小胖】

从底部 ”源码分析“菜单 即可获取

《Skywalking源码分析指北》全部文章哟~

Skywalking源码分析指北第十一篇——DataCarrier

看懂看不懂，都点个赞吧:+1:

原文  http://mp.weixin.qq.com/s?__biz=MzU5Mjc5OTY5Ng==&mid=2247484673&idx=2&sn=5cbd4d92d2362a7815348cfbd7512bfc
正文到此结束
所属分类： Java 编程技术

热门推荐

相关文章

阿里云首购8折
说给你听

本文目录
随机标签
书籍教程
springboot-demo Java入门教程 bootstrap3 CSS Apache基础教程 php ionic 教程 Python mysql教程 eclipse Ubuntu VPS系统配置 AngularJS 教程 MongoDB教程 Struts2教程 Redis教程 Spring教程 Git教程 Jenkins进阶系列 openfire参考指南 Java设计模式 HBase教程 Maven教程 hibernate教程 Docker 教程 memcached教程 Quartz指南 Hive教程 Ant教程 ANTLR教程 SpringCloud java实例教程 springcloud-demo XStream教程 Hazelcast教程 Elastic-Job-Lite 深入浅出MyBatis SVN教程 ibaties教程 rabittmq教程 solr教程 Hadoop教程 WebService CXF学习 JPA教程 ActiveMQ中文指南 java-demo Java内存模型 dubbo教程 python3-demo Linux入门视频教程
近期评论
  1. 1 Spring Cloud Consul实现选举机制
  2. 2 Spring Boot集成ShedLock实现分布式定时任务
  3. 3 利用oss进行数据库和网站图片备份
  4. 4 Spring Cloud Stream实现数据流处理
  5. 5 Python3访问MySQL数据库快速入门Demo
  6. 6 Github开源项目作者可以免费申请 JetBrains 全家桶
  7. 7 Spring Cloud Vault快速入门Demo
  8. 8 Spring Cloud Gateway快速入门Demo
  9. 9 Spring Cloud Contract快速入门Demo
  10. 10 Spring Cloud Consul快速入门Demo
网站信息
  • 文章总数：82,714 篇
  • 文件总数：284,287 个
  • 标签总数：2,393 个
  • 分类总数：85 个
  • 留言数量：2,560 条
  • 在线人数：654 人
  • 运行天数：4,408天
  • 最后更新：2024年11月21日23点
Loading...

其他链接

关于本站

本站定位：个人技术类博客

本站作用：写博客、记日志、闲聊扯淡鼓捣技术。

问题交流

[HBLOG]公众号 HBLOG HBLOG