flink内部计算指标的95线-99线等的实现

15年在某电商从0设计了一个通用的API监控系统，当时只是计算了成功率+平均耗时，没有算75,90,95,99,999,9999线，这次单位需要，所以促使我去思考这个问题，问了单位CAT维护人员，大致了解了计算方式，跟我在18年7月份在单位内网BBS发表的文章思路是一致的，所以就直接写了下面的代码

PercentageCalculation.java

package com.ymm.computation.udf.define;

import org.apache.flink.table.functions.AggregateFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

//批量计算95line类似的数据
public class PercentageCalculation extends AggregateFunction<Object, PercentageAccumulator> {

    /**  */
    private static final long   serialVersionUID = 4009559061130131166L;
    private static final Logger LOG              = LoggerFactory
        .getLogger(PercentageCalculation.class);
    //private static BlockingQueue<PercentageAggregatorContainer> GLOBAL_QUEUE     = new LinkedBlockingQueue<PercentageAggregatorContainer>();

    @Override
    public PercentageAccumulator createAccumulator() {
        return new PercentageAccumulator();
    }

    public void accumulate(PercentageAccumulator accumulator, Object value) {
        accumulator.accumulate(value);
    }

    @Override
    public Object getValue(PercentageAccumulator accumulator) {
        return accumulator.getValue();
    }

    public void resetAccumulator(PercentageAccumulator acc) {
        acc = null;//help GC
    }

}

PercentageAccumulator.java

package com.ymm.computation.udf.define;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

//!只针对时间来计算95线等,其它参数不要使用本类
public class PercentageAccumulator {

    private static final Logger LOG            = LoggerFactory
        .getLogger(PercentageAccumulator.class);

    public final static double  PERCENT_50     = 0.5;

    public final static double  PERCENT_75     = 0.25;

    public final static double  PERCENT_90     = 0.1;

    public final static double  PERCENT_95     = 0.05;

    public final static double  PERCENT_99     = 0.01;

    public final static double  PERCENT_999    = 0.001;

    public final static double  PERCENT_9999   = 0.0001;

    public final static int     PERCENT_COUNT  = 7;

    private final static int[]  SCALE          = {             //
                                                   1,          //
                                                   2,          //
                                                   4,          //
                                                   8,          //
                                                   16,         //
                                                   32,         //
                                                   64,         //
                                                   128,        //
                                                   256,        //
                                                   512,        //
                                                   1024,       //
                                                   2048,       //
                                                   4096,       //
                                                   8192,       //
                                                   16384,      //
                                                   32768,      //
                                                   65536       //
    };

    private int[]               countContainer = {             //
                                                   0,          //<=1
                                                   0,          //<=2
                                                   0,          //<=4
                                                   0,          //<=8
                                                   0,          //<=16
                                                   0,          //<=32
                                                   0,          //<=64
                                                   0,          //<=128
                                                   0,          //<=256
                                                   0,          //<=512
                                                   0,          //<=1024
                                                   0,          //<=2048
                                                   0,          //<=4096
                                                   0,          //<=8192
                                                   0,          //<=16384
                                                   0,          //<=32768
                                                   0           //<=65536
    };

    private int positionByTwoDivision(int[] array, int begin, int end, int value) {
        int mid = (begin + end) >> 1;
        int midValue = array[mid];
        int halfMidValue = midValue >> 1;
        //判断是否可以命中mid
        if (value > halfMidValue && value <= midValue) {
            return mid;
        }
        //没法命中,则根据大小来定
        if (value <= halfMidValue) {
            if (mid - 1 < 0) {//没路可走的边界条件
                return 0;
            }
            return positionByTwoDivision(array, begin, mid - 1, value);
        } else {
            return positionByTwoDivision(array, mid + 1, end, value);
        }

    }

    public int positionInValueArray(int val) {
        int length = SCALE.length;
        //如果大于最大值|小于等于最小值
        if (val >= SCALE[length - 1]) {
            return length - 1;
        } else if (val <= SCALE[0]) {
            return 0;
        }
        //采用2分法来计算
        return positionByTwoDivision(SCALE, 0, length - 1, val);
    }

    public void accumulate(Object value) {
        //转换为long值,int值够用了
        Long longValue = (Long) value;
        int intValue = longValue.intValue();
        //找到下标
        int index = positionInValueArray(intValue);
        countContainer[index]++;
    }

    //确保在[1,MAX]范围内,
    //自然顺序
    private int adjust(int input, int max) {
        if (input <= 1) {
            return 1;
        } else if (input >= max) {
            return max;
        } else {
            return input;
        }
    }

    private static final ThreadLocal<StringBuilder> STR_BUILDER_ThreadLocal = new ThreadLocal<StringBuilder>() {
                                                                                public StringBuilder initialValue() {
                                                                                    return new StringBuilder();
                                                                                }
                                                                            };
    private static final String                     SEPARATOR               = ":";

    public String getValue() {

        //total
        int total = 0;
        int length = countContainer.length;
        for (int index = 0; index < length; index++) {
            total += countContainer[index];
        }

        //如果total为0的异常情况
        //注意是自然序---[1,total]
        int percent_9999_pos = adjust((int) (total * PERCENT_9999), total);
        boolean found_9999 = false;
        int percent_9999_value = Integer.MAX_VALUE;
        //999
        int percent_999_pos = adjust((int) (total * PERCENT_999), total);
        boolean found_999 = false;
        int percent_999_value = Integer.MAX_VALUE;
        //99
        int percent_99_pos = adjust((int) (total * PERCENT_99), total);
        boolean found_99 = false;
        int percent_99_value = Integer.MAX_VALUE;
        //95
        int percent_95_pos = adjust((int) (total * PERCENT_95), total);
        boolean found_95 = false;
        int percent_95_value = Integer.MAX_VALUE;
        //90
        int percent_90_pos = adjust((int) (total * PERCENT_90), total);
        boolean found_90 = false;
        int percent_90_value = Integer.MAX_VALUE;
        //75
        int percent_75_pos = adjust((int) (total * PERCENT_75), total);
        boolean found_75 = false;
        int percent_75_value = Integer.MAX_VALUE;
        //50
        int percent_50_pos = adjust((int) (total * PERCENT_50), total);
        boolean found_50 = false;
        int percent_50_value = Integer.MAX_VALUE;

        //开始遍历每一个元素,从后往前算
        int scanned = 0;
        int left = PERCENT_COUNT;
        for (int index = length - 1; index >= 0; index--) {
            //当前没有值,无论如何也不会成为备选
            if (0 == countContainer[index]) {
                continue;
            }
            //当前有值
            scanned += countContainer[index];
            //逐个判断
            //9999线
            if (false == found_9999 && scanned >= percent_9999_pos) {
                percent_9999_value = SCALE[index];
                found_9999 = true;
                left--;
            }
            //999线
            if (false == found_999 && scanned >= percent_999_pos) {
                percent_999_value = SCALE[index];
                found_999 = true;
                left--;
            }
            //99线
            if (false == found_99 && scanned >= percent_99_pos) {
                percent_99_value = SCALE[index];
                found_99 = true;
                left--;
            }
            //95线
            if (false == found_95 && scanned >= percent_95_pos) {
                percent_95_value = SCALE[index];
                found_95 = true;
                left--;
            }
            //90线
            if (false == found_90 && scanned >= percent_90_pos) {
                percent_90_value = SCALE[index];
                found_90 = true;
                left--;
            }
            //75线
            if (false == found_75 && scanned >= percent_75_pos) {
                percent_75_value = SCALE[index];
                found_75 = true;
                left--;
            }
            //50线
            if (false == found_50 && scanned >= percent_50_pos) {
                percent_50_value = SCALE[index];
                found_50 = true;
                left--;
            }
            //全部都找到了就break
            if (0 == left) {
                break;
            }
        }

        //所有的值都算好了
        //拿出来时先reset一下
        StringBuilder stringBuilder = STR_BUILDER_ThreadLocal.get();
        stringBuilder.delete(0, stringBuilder.length());
        //开始挂各种数据,测试表明每秒几百万次执行
        stringBuilder.append(percent_50_value);
        stringBuilder.append(SEPARATOR);
        stringBuilder.append(percent_75_value);
        stringBuilder.append(SEPARATOR);
        stringBuilder.append(percent_90_value);
        stringBuilder.append(SEPARATOR);
        stringBuilder.append(percent_95_value);
        stringBuilder.append(SEPARATOR);
        stringBuilder.append(percent_99_value);
        stringBuilder.append(SEPARATOR);
        stringBuilder.append(percent_999_value);
        stringBuilder.append(SEPARATOR);
        stringBuilder.append(percent_9999_value);
        //return
        return stringBuilder.toString();
    }

    public void print() {
        for (int index = 0; index < this.countContainer.length; index++) {
            System.out.println(index + "->" + this.countContainer[index]);
        }
    }
}

欢迎提出 优化建议，比如对GC更友好的优化建议！

这个函数的瓶颈在于stringbuilder.