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.
