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 {    /**  */    private static final long   serialVersionUID = 4009559061130131166L;    private static final Logger LOG              = LoggerFactory        .getLogger(PercentageCalculation.class);    //private static BlockingQueue GLOBAL_QUEUE     = new LinkedBlockingQueue();    @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 STR_BUILDER_ThreadLocal = new ThreadLocal() {                                                                                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.