每日一句

不要因为怕被玫瑰的刺伤到你，就不敢去摘玫瑰。

目录

前言

上一节我们知道了Ribbon是通过ServerList来管理获取服务器列表的方式。虽然现在没有说到Ribbon负载均衡器是如何工作的，但是大致应该是 获取服务列表 过滤服务列表 根据负载策略选择具体的服务器。那么如果我需要根据服务器的负载量来选择服务，或者根据请求服务器的失败率来过滤服务器。 这些服务器的指标数据 Ribbon是存储在哪里的，又是如何统计的。今天我们就来看一下和服务器指标有关的两个类ServerStats&LoadBalancerStats

netflix-statistics

在聊ServerStats之前我们先来看看Ribbon是如何对数据进行收集的。这就需要了解netflix-commons工程下面的netflix-statistics模块，该模块用来对指标数据收集。很简单 简略的介绍一下。

继承关系

在这里插入图片描述
可以看到这个模块就6个核心类，一共10个类。

  • DataCollector 数据收集父接口 只提供noteValue方法
  • Distribution以增量的方式生成 实现是线程不安全的
  • DataBufferd固定大小的数据收集缓存区 一个最近添加值得一个滑动窗口
  • Histogram跟踪每一个存储桶的计数 来查找中位数
  • DataAccumulator数据累加器 维护了双缓存区
  • DataDistribution数据发布 具有实际的发布能力

DataCollector

数据收集 该接口非常简单 只有一个方法

public interface DataCollector {
  //记录一个数据值
  void noteValue(double val);
}

Distribution

统计量累加器 以增量的方式产生值。实现是线程不安全的，同时更新会产生错误的结果。在大多数情况下，这些不正确的结果往往是不重要的。

 	//数量累加
 	private lo ng numValues;
 	//值累加
    private double sumValues;
    //平方累加
    private double sumSquareValues;
    //最小值
    private double minValue;
    //最大值
    private double maxValue;
    public Distribution() {
        numValues = 0L;
        sumValues = 0.0;
        sumSquareValues = 0.0;
        minValue = 0.0;
        maxValue = 0.0;
    }
    /** 记录值 */
    public void noteValue(double val) {
        numValues++;
        sumValues += val;
        sumSquareValues += val * val;
        if (numValues == 1) {
            minValue = val;
            maxValue = val;
        } else if (val < minValue) {
            minValue = val;
        } else if (val > maxValue) {
            maxValue = val;
        }
    }
    /** 恢复默认配置 */
    public void clear() {
        numValues = 0L;
        sumValues = 0.0;
        sumSquareValues = 0.0;
        minValue = 0.0;
        maxValue = 0.0;
    }

该类非常简单就是对要记录的值得一个简单操作 统计。最大值，最小值，平方值，统计总数，总值不难发现这些值都是最终的值 没法记录过程。某段时间内的指标是无法做到的。

DataBuffer

光看名字就知道 它提供了数据缓存能力 具有记录过程中值得能力。

    //锁 可以外部可以通过getLock来管理对 buf内容的访问
 	private final Lock lock;
 	//数据缓存区
    private final double[] buf;
    //开始统计的时间
    private long startMillis;
    //结束统计的时间
    private long endMillis;
    //缓存区大小
    private int size;
    //当前插入的位置
    private int insertPos;
    @Override
    public void noteValue(double val) {
        //对值得最终统计
        super.noteValue(val);
        //如果值过多就会对之前的内容造成覆盖 所以这是一个滑动的统计
        buf[insertPos++] = val;
        if (insertPos >= buf.length) {
            insertPos = 0;
            size = buf.length;
        } else if (insertPos > size) {
            size = insertPos;
        }
    }
    //开始收集
     public void startCollection() {
        clear();
        startMillis = System.currentTimeMillis();
    }
    //结束收集
  public void endCollection() {
     endMillis = System.currentTimeMillis();
     Arrays.sort(buf, 0, size);
 }

上面的方法很简单。下面我们来搞清楚一个概念：分位数 这个是统计学中的术语。

  • 二分位数： 对于有限的数集，可以通过把所有观察值高低排序后找出正中间的一个作为中位数。如果观察值有偶数个，则中位数不唯一，通常取最中间的两个数值的平均数作为中位数，即二分位数。
  • **四分位数：**是统计学中分位数的一种，即把所有数值由小到大排列并分成四等份，处于三个分割点位置的数值就是四分位数。
  • *百分位数：如果将一组数据从小到大排序，并计算相应的累计百分位，则某一百分位所对应数据的值就称为这一百分位的百分位数。可表示为：一组n个观测值按数值大小排列。如，处于p%位置的值称第p百分位数。（***百分位通常用第几百分位来表示，如第五百分位，它表示在所有测量数据中，测量值的累计频次达5%。以身高为例，身高分布的第五百分位表示有5%的人的身高小于此测量值，95%的身高大于此测量值。）

下面看DataBuffer为我们提供计算百分位数的方法：

public double[] getPercentiles(double[] percents, double[] percentiles) {
        for (int i = 0; i < percents.length; i++) {
            percentiles[i] = computePercentile(percents[i]);
        }
        return percentiles;
    }

    private double computePercentile(double percent) {
        // Some just-in-case edge cases
        if (size <= 0) {
            return 0.0;
        } else if (percent <= 0.0) {
            return buf[0];
        } else if (percent >= 100.0) {        
            return buf[size - 1];
        } 
        double index = (percent / 100.0) * size; 
        int iLow = (int) Math.floor(index);
        int iHigh = (int) Math.ceil(index);
        assert 0 <= iLow && iLow <= index && index <= iHigh && iHigh <= size;
        assert (iHigh - iLow) <= 1;
        if (iHigh >= size) {
            // Another edge case
            return buf[size - 1];
        } else if (iLow == iHigh) {
            return buf[iLow];
        } else {
            return buf[iLow] + (index - iLow) * (buf[iHigh] - buf[iLow]);
        }
    }

***上面的方法 解释一下 例如 缓存 50个值 我们传的是percent 是10 也就是要计算第10百分位的值 也就是要求x 这个x需要满足 10%的值小于x 90%的值大于x *** 明白了吗，没明白 我们再来个例子。

		DataBuffer dataBuffer = new DataBuffer(20);
        //开始收集
        dataBuffer.startCollection();
        for (int i = 0; i < 20; i++) {
            dataBuffer.noteValue(i);
        }
        //结束收集
        dataBuffer.endCollection();
        //我们这里需要计算10分位数  50分位数 90分位数的值
        //根据我们的预测 10分位数应该是 2  50分位数对应的是10  90分位数对应的是 18
        double[] percents = new double[]{10,50,90};
        double[] percentiles = new double[percents.length];
        dataBuffer.getPercentiles(percents , percentiles);
        for (int i = 0; i < percentiles.length; i++) {
            System.out.println(percentiles[i]);
        }

上面的例子模拟了一个计算百分位数的过程。实际输出 和我们预测的结果是一毛一样的。好了 它就是这么简单。

Histogram

这个我们就跳过 Ribbon没有直接使用到。

DataAccumulator

数据统计 ,拥有双缓冲区，一个是当前缓存区 用来向其中添加新数据。另一个是上一个缓存区，用于计算统计信息。

    //当前缓存区 用来添加新数据   这个DataBuffer正是我们上面说到的
	private DataBuffer current;
	// 上一个缓存区 
    private DataBuffer previous;
    private final Object swapLock = new Object();
    //记录一个值 这里加了swapLock 交换锁 大家应该能猜到 在执行数据交换的时候是无法新增的。
    //所谓的数据交换就是 将current缓存的值 复制到previous中的过程。
    public void noteValue(double val) {
        synchronized (swapLock) {
            Lock l = current.getLock();
            l.lock();
            try {
                //记录当前值
                current.noteValue(val);
            } finally {
                l.unlock();
            }
        }
    }
    //数据交换 并且计算上一个缓存区
    public void publish() {
        DataBuffer tmp = null;
        Lock l = null;
        //下面是一个交换值的过程 上面说了交换的时候不能新增 所以这里加了swapLock
        synchronized (swapLock) {
            // Swap buffers
            tmp = current;
            current = previous;
            previous = tmp;
            l = current.getLock();
            l.lock();
            try {
                //当前缓存清空
                current.startCollection();
            } finally {
                l.unlock();
            }
            l = tmp.getLock();
            l.lock();
        }
        try {
            //结束收集  
            tmp.endCollection();
            publish(tmp);
        } finally {
            l.unlock();
        }
    }
    //具体的发布（计算）交给子类实现
    protected abstract void publish(DataBuffer buf);

上面代码主要是一个收集和交换的过程。收集和交换这两个操作是互斥的。
交换过程：

  1. 使用临时变量temp进行交换操作
  2. 交换完成当前缓存区调用startCollection清空缓存区 准备再次开始收集。
  3. 临时缓存区调用endCollection结束收集，并把临时缓存区传入publish方法交由子类处理

DataDistribution

DataDistribution是DataAccumulator的唯一实现。那它最重要的就是实现了父类留下来的publish方法 来计算当前缓存区留下来的值。

    //统计数据总个数
	private long numValues = 0L;
	//统计数据平均数
    private double mean = 0.0;
    //方差
    private double variance = 0.0;
    //标准差
    private double stddev = 0.0;
    //最小值
    private double min = 0.0;
    //最大值
    private double max = 0.0;
    //数据发布时间
    private long ts = 0L;
    //样本时间
    private long interval = 0L;
    private int size = 0;
    //下面两个值 我们上面介绍过 百分位数 和百分位数对应的值
    private final double[] percents;
    private final double[] percentiles;
    public DataDistribution(int bufferSize, double[] percents) {
        super(bufferSize);
        assert percentsOK(percents);
        this.percents = percents;
        this.percentiles = new double[percents.length];
    }
    //参数校验
    private static boolean percentsOK(double[] percents) {
        if (percents == null) {
            return false;
        }
        for (int i = 0; i < percents.length; i++) {
            if (percents[i] < 0.0 || percents[i] > 100.0) {
                return false;
            }
        }
        return true;
    }
    //数据发布
    protected void publish(DataBuffer buf) {
    //记录发布时间
        ts = System.currentTimeMillis();
   //数据发布总数 这个总数最终是由Distribution提供      
        numValues = buf.getNumValues();
   //下面是对上面介绍属性赋值的过程 就不一一注释了。     
        mean = buf.getMean();
        variance = buf.getVariance();
        stddev = buf.getStdDev();
        min = buf.getMinimum();
        max = buf.getMaximum();
        interval = buf.getSampleIntervalMillis();
        size = buf.getSampleSize();
        //通过我们上面介绍的这个方法 来计算百分位数
        buf.getPercentiles(percents, percentiles);
    }
    //初始化
    public void clear() {
       .....
    }

DataPublisher

数据发布器 听名字就知道是调用DataAccumulator的publish方法进行数据定时发布。使用定时器来定时对数据发布。这里就不贴源码了，比较简单。直接上一个demo

		final DataDistribution accumulator = new DataDistribution(20 ,
            new double[]{10 , 50 , 90});
        DataPublisher dataPublisher = new DataPublisher(accumulator , 2000);
        dataPublisher.start();
        new Thread(new Runnable() {
            @Override
            public void run() {
                for (int i = 1; i < 1000; i++) {
                    try {
                        accumulator.noteValue(i);
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }).start();
        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
        executorService.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                System.out.println("=======时间：" + accumulator.getTimestamp() + "=======");
                System.out.println("总个数：" + accumulator.getNumValues());
                System.out.println("统计周期：" + accumulator.getSampleIntervalMillis() );
                System.out.println("样本数据个数：" + accumulator.getSampleSize());
                System.out.println("最大值：" + accumulator.getMaximum());
                System.out.println("最小值：" + accumulator.getMinimum());
                System.out.println("平均值：" + accumulator.getMean());
                System.out.println("方差：" + accumulator.getVariance());
                System.out.println("标准差：" + accumulator.getStdDev());
                System.out.println("分位数：" + Arrays.toString(accumulator.getPercents()));
                System.out.println("分位数值：" + Arrays.toString(accumulator.getPercentiles()));
            }
        }, 2500, 2500, TimeUnit.MILLISECONDS);
        TimeUnit.SECONDS.sleep(Integer.MAX_VALUE);

上面的例子 我们定义了一个DataDistribution来产生数据，DataPublisher绑定DataDistribution用来定时发送数据。

程序输出：

=======时间：Tue Nov 24 22:16:51 CST 2020=======
总个数：20
统计周期：1606227411869
样本数据个数：20
最大值：20.0
最小值：1.0
平均值：10.5
方差：33.25
标准差：5.766281297335398
分位数：[10.0, 50.0, 90.0]
分位数值：[3.0, 11.0, 19.0]
=======时间：Tue Nov 24 22:16:53 CST 2020=======
总个数：19
统计周期：2002
样本数据个数：19
最大值：39.0
最小值：21.0
平均值：30.0
方差：30.0
标准差：5.477225575051661
分位数：[10.0, 50.0, 90.0]
分位数值：[22.9, 30.5, 38.1]

到这里我们就把铺垫工作做好了。netflix-statistics比较简单 但是为了后面我们能很好的理解ServerStats 还是有必要把基础的类讲解一下的。

在这里插入图片描述

# java # ribbon
Logo
永洪数据分析社区

永洪科技，致力于打造全球领先的数据技术厂商，具备从数据应用方案咨询、BI、AIGC智能分析、数字孪生、数据资产、数据治理、数据实施的端到端大数据价值服务能力。

更多推荐

cover

【PostgreSQL数据分析实战：从数据清洗到可视化全流程】4.4 异构数据源整合（CSV/JSON/Excel数据导入）

avatar 永洪数据分析社区
cover

大数据面试问答-数据湖

avatar 永洪数据分析社区
cover

Vue3携手Echarts，打造炫酷数据可视化大屏

avatar 永洪数据分析社区