自己实现一个滑动窗口

基本概念

1. 移动平均值：

   一个移动平均值计算常常用来在事件序列数据中消除短期波动，展示长期的趋势。
   移动平均值的平滑效果通过在计算中考虑到历史值来实现。计算一个移动平均值可以通过少量的状态来进行，对于一个事件序列，我们只需要记录上次发生的时间和上次计算出来的评价值即可。

      diff = currentTime-lastEventTime
      currentAverage = (1.0-alpha)*diff+alpha*lastAverage

上述计算中的alpha的值是一个0~1之间的常量，aplha值决定了一段时间内的平滑水平，alpha越趋于1，历史值对当前的平均值的影响越大，反之亦然

2. 滑动窗口

• 某些情况下，我们需要降低历史值对当前移动平均值的影响，例如当两次事件之间的间隔时间较长时，需要重置平滑作用。如果有一个较小的alpha值，可能不需要这么做，因为平滑效果已经很好。但是，如果aplha值很大时，需要适当地降低平滑效果的影响.
• 考虑下面的例子。
  我们有一个事件(比如说网络错误) 很少发生。偶尔出现小的峰值，通常是设什么问
  题的。所以我们]想平滑这些小的峰值。只有当连续的峰值州现时，我们才需要发出通知。
  如果事件平均一周才发生一次(达不到通知的阈值)，但是某一天一小时内出现了多
  个峰值(超过了通知阈值)，alpha 值较大的平滑效果可能抵消了峰值，导致事件一直无法
  触发。
  为了中和这种影响，我们可以在计算移动平均值时引人滑动窗口的概念。因为我们已
  经保留了上一个事件的时间戳以及当前的平均值，实现一个滑动窗口非常简单，如下面伪
  代码所示:

f(cur rent Time last BventT ime) > s1idingWindowInterval
currentAverage = 0
end if  ....

一个完整的实例代码如下

import java.io.Serializable;

public class EWMA implements Serializable {
    private static final long serialVersionUID = -6408346318181111576L;
    // 和UNIX系统计算负载时使用的标准alpha值相同
    public static final double ONE_MINUTE_ALPHA = 1-Math.exp(-5d / 60d / 1d);
    public static final double FIVE_MINUTE_ALPHA = 1-Math.exp(-5d / 60d / 5d);
    public static final double FIFTEEN_MINUTE_ALPHA = 1-Math.exp(-5d / 60d / 15d);

    public static enum Time {
        MILLISECONDS(1),
        SECONDS(1000),
        MINUTES(SECONDS.getMillis() * 60),
        HOURS(MINUTES.getMillis() * 60),
        DAYS(HOURS.getMillis() * 24),
        WEEKS(DAYS.getMillis() * 7);

        private long millis;

        Time(long millis) {
            this.millis = millis;
        }

        public long getMillis() {
            return millis;
        }
    }

    private long window;  //滑动窗口大小
    private long alphaWindow;
    private long last;  //记录上一次的时间
    private double average;  //移动平均值
    private double alpha = -1D;  //平滑水平
    private boolean sliding = false;  //是否移动

    public EWMA() {
    }

    /**
     * 建立指定时间的滑动窗口
     */
    public EWMA sliding(double count,Time time){
        return this.sliding((long)(time.getMillis()*count));
    }

    private EWMA sliding(long window){
        this.sliding = true;
        this.window = window;
        return this;
    }

    /**
     * 指定alpha值
     * @param alpha
     * @return
     */
    public EWMA withAlpha(double alpha){
        if(!(alpha>0.0D)&&alpha<=1.0D){
            throw new IllegalArgumentException("Alpha must be between 0.0 and 1.0");
        }
        this.alpha = alpha;
        return this;
    }

    /**
     * 作为一个alphaWindow窗口的函数
     * alpha = 【1-Math.exp(-5d / 60d / alphaWindow)】
     * @param alphaWindow
     * @return
     */
    public EWMA withAlphaWindow(long alphaWindow){
        this.alpha = -1;
        this.alphaWindow = alphaWindow;
        return this;
    }


    public EWMA withAlphaWindow(double count,Time time){
        return this.withAlphaWindow((long)(time.getMillis()*count));
    }

    /**
     * 默认使用当前时间更新移动平均值
     */
    public void mark(){
        mark(System.currentTimeMillis());
    }

    /**
     * 更新移动平均值
     * @param time
     */
    public synchronized void mark(long time){
        if(this.sliding){
            //如果发生时间间隔大于窗口，则重置滑动窗口
            if(time-this.last > this.window){
                this.last = 0;
            }
        }
        if(this.last == 0){
            this.average = 0;
            this.last = time;
        }
        // 计算上一次和本次的时间差
        long diff = time-this.last;
        // 计算alpha
        double alpha = this.alpha != -1.0 ? this.alpha : Math.exp(-1.0*((double)diff/this.alphaWindow));
        // 计算当前平均值
        this.average = (1.0-alpha)*diff + alpha*this.average;
        this.last = time;
    }

    /**
     * mark()方法多次调用的平均间隔时间（历史平均水平）
     * @return
     */
    public double getAverage(){
        return this.average;
    }

    /**
     * 按照指定的时间返回平均值
     * @param time
     * @return
     */
    public double getAverageIn(Time time){
        return this.average == 0.0?this.average:this.average/time.getMillis();
    }

    /**
     * 返回特定时间度量内调用mark()的频率
     * @param time
     * @return
     */
    public double getAverageRatePer(Time time){
        return this.average == 0.0?this.average:time.getMillis()/this.average;
    }
}

使用实例

//指定一个1分钟的滑动窗口  
  EWMA ewma = new EWMA().sliding(1.0, EWMA.Time.MINUTES).withAlpha(EWMA.ONE_MINUTE_ALPHA);