flink內(nèi)部計(jì)算指標(biāo)的95線-99線等的實(shí)現(xiàn)

15年在某電商從0設(shè)計(jì)了一個(gè)通用的API監(jiān)控系統(tǒng)，當(dāng)時(shí)只是計(jì)算了成功率+平均耗時(shí)，沒(méi)有算75,90,95,99,999,9999線，這次單位需要，所以促使我去思考這個(gè)問(wèn)題，問(wèn)了單位CAT維護(hù)人員，大致了解了計(jì)算方式，跟我在18年7月份在單位內(nèi)網(wǎng)BBS發(fā)表的文章思路是一致的，所以就直接寫(xiě)了下面的代碼

PercentageCalculation.java

package com.ymm.computation.udf.define;import org.apache.flink.table.functions.AggregateFunction;import org.slf4j.Logger;import org.slf4j.LoggerFactory;//批量計(jì)算95line類(lèi)似的數(shù)據(jù)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;//!只針對(duì)時(shí)間來(lái)計(jì)算95線等,其它參數(shù)不要使用本類(lèi)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;
        }        //沒(méi)法命中,則根據(jù)大小來(lái)定
        if (value <= halfMidValue) {            if (mid - 1 < 0) {//沒(méi)路可走的邊界條件
                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分法來(lái)計(jì)算
        return positionByTwoDivision(SCALE, 0, length - 1, val);
    }    public void accumulate(Object value) {        //轉(zhuǎn)換為long值,int值夠用了
        Long longValue = (Long) value;        int intValue = longValue.intValue();        //找到下標(biāo)
        int index = positionInValueArray(intValue);
        countContainer[index]++;
    }    //確保在[1,MAX]范圍內(nèi),
    //自然順序
    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;        //開(kāi)始遍歷每一個(gè)元素,從后往前算
        int scanned = 0;        int left = PERCENT_COUNT;        for (int index = length - 1; index >= 0; index--) {            //當(dāng)前沒(méi)有值,無(wú)論如何也不會(huì)成為備選
            if (0 == countContainer[index]) {                continue;
            }            //當(dāng)前有值
            scanned += countContainer[index];            //逐個(gè)判斷
            //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;
            }
        }        //所有的值都算好了
        //拿出來(lái)時(shí)先reset一下
        StringBuilder stringBuilder = STR_BUILDER_ThreadLocal.get();
        stringBuilder.delete(0, stringBuilder.length());        //開(kāi)始掛各種數(shù)據(jù),測(cè)試表明每秒幾百萬(wàn)次執(zhí)行
        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]);
        }
    }
}

歡迎提出 優(yōu)化建議，比如對(duì)GC更友好的優(yōu)化建議！

這個(gè)函數(shù)的瓶頸在于stringbuilder.