固定窗口和滑动窗口算法了解一下

眯眯眼的猫头鹰

发布于 2018-10-31 11:16:41

4.2K0

发布于 2018-10-31 11:16:41

前言

最近在参与一个识别热点数据的需求开发。其中涉及了限流算法相关的内容。所以这里记录一下自己了解的各种限流算法，以及各个限流算法的实现。

限流算法的应用场景非常广泛，比如通过限流来确保下游配置较差的应用不会被上游应用的大量请求击穿，无论是HTTP请求还是RPC请求，从而使得服务保持稳定。限流也同样可以用于客户端，比如当我们需要从微博上爬取数据时，我们需要在请求中携带token从而通过微博的网关验证。但是微博为了防止服务被单个客户端大量访问，往往会在服务端进行限流，比如可能是一个token一个小时只能发起1000次请求。但是爬虫发出的请求通常远远不止这个量级。所以在客户端进行限流可以确保我们的token不会失效或是查封。

限流算法可以从多种角度分类，比如按照处理方式分为两种，一种是在超出限定流量之后会拒绝多余的访问，另一种是超出限定流量之后，只是报警或者是记录日志，访问仍然正常进行。

目前比较常见的限流算法有以下几种：

固定窗口
滑动窗口
令牌桶算法
漏桶算法

本文主要记录一下固定窗口和滑动窗口。令牌桶算法在谷歌的开源guava包中有实现，下次再开一篇文章分享一下。文中错误的地方欢迎指出！如果guava中实现了滑动窗口算法也请告诉我，急需，目前没有找到orz。

固定窗口

这是限流算法中最暴力的一种想法。既然我们希望某个API在一分钟内只能固定被访问N次（可能是出于安全考虑，也可能是出于服务器资源的考虑），那么我们就可以直接统计这一分钟开始对API的访问次数，如果访问次数超过了限定值，则抛弃后续的访问。直到下一分钟开始，再开放对API的访问。

所有的暴力算法的共同点都是容易实现，而固定窗口限流的缺点也同样很明显。假设现在有一个恶意用户在上一分钟的最后一秒和下一分钟的第一秒疯狂的冲击API。按照固定窗口的限流规则，这些请求都能够访问成功，但是在这一秒内，服务将承受超过规定值的访问冲击（这个规定值很可能是服务器能够承受的最大负载），从而导致服务无法稳定提供。而且因为用户在这一秒内耗光了上一分钟和下一分钟的访问定额，从而导致别的用户无法享受正常的服务，对于服务提供方来说是完全不能接收的。

这里自己做了一个简单的实现：

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class FixedWindowRateLimiter implements RateLimiter, Runnable {

    private static final int DEFAULT_ALLOWED_VISIT_PER_SECOND = 5;

    private final int maxVisitPerSecond;

    private AtomicInteger count;

    FixedWindowRateLimiter(){
        this.maxVisitPerSecond = DEFAULT_ALLOWED_VISIT_PER_SECOND;
        this.count = new AtomicInteger();
    }

    FixedWindowRateLimiter(int maxVisitPerSecond) {
        this.maxVisitPerSecond = maxVisitPerSecond;
        this.count = new AtomicInteger();
    }

    @Override
    public boolean isOverLimit() {
        return currentQPS() > maxVisitPerSecond;
    }

    @Override
    public int currentQPS() {
        return count.get();
    }

    @Override
    public boolean visit() {
        count.incrementAndGet();
        System.out.print(isOverLimit());
        return isOverLimit();
    }

    @Override
    public void run() {
        System.out.println(this.currentQPS());
        count.set(0);
    }

    public static void main(String[] args) {
        ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
        FixedWindowRateLimiter rateLimiter = new FixedWindowRateLimiter();
        scheduledExecutorService.scheduleAtFixedRate(rateLimiter, 0, 1, TimeUnit.SECONDS);
        new Thread(new Runnable() {
            @Override
            public void run() {
                while(true) {
                    rateLimiter.visit();
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }).start();

        new Thread(new Runnable() {
            @Override
            public void run() {
                while(true) {
                    rateLimiter.visit();
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }).start();

    }
}

其中RateLimiter是一个通用的接口，后面的其它限流算法也会实现该接口:

public interface RateLimiter {

    boolean isOverLimit();

    int currentQPS();

    boolean visit();
}

也可以不使用多线程的方式实现，更加简单高效：

public class FixedWindowRateLimiterWithoutMultiThread implements RateLimiter {
    private Long lastVisitAt = System.currentTimeMillis();
    private static final int DEFAULT_ALLOWED_VISIT_PER_SECOND = 5;

    private final int maxVisitPerSecond;

    private AtomicInteger count;

    public FixedWindowRateLimiterWithoutMultiThread(int maxVisitPerSecond){
        this.maxVisitPerSecond = maxVisitPerSecond;
        this.count = new AtomicInteger();
    }

    public FixedWindowRateLimiterWithoutMultiThread() {
        this(DEFAULT_ALLOWED_VISIT_PER_SECOND);
    }
    @Override
    public boolean isOverLimit() {
        return count.get() > maxVisitPerSecond;
    }

    @Override
    public int currentQPS() {
        return count.get();
    }

    @Override
    public boolean visit() {
        long now = System.currentTimeMillis();
        synchronized (lastVisitAt) {
            if (now - lastVisitAt > 1000) {
                lastVisitAt = now;
                System.out.println(currentQPS());
                count.set(1);
            }
        }
        count.incrementAndGet();
        return isOverLimit();
    }

    public static void main(String[] args) {
        RateLimiter rateLimiter = new FixedWindowRateLimiterWithoutMultiThread();
        new Thread(new Runnable() {
            @Override
            public void run() {
                while(true) {
                    rateLimiter.visit();
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }).start();

        new Thread(new Runnable() {
            @Override
            public void run() {
                while(true) {
                    rateLimiter.visit();
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }).start();

    }
}

滑动窗口

固定窗口就像是滑动窗口的一个特例。滑动窗口将固定窗口再等分为多个小的窗口，每一次对一个小的窗口进行流量控制。这种方法可以很好的解决之前的临界问题。

这里找的网上一个图，假设我们将1s划分为4个窗口，则每个窗口对应250ms。假设恶意用户还是在上一秒的最后一刻和下一秒的第一刻冲击服务，按照滑动窗口的原理，此时统计上一秒的最后750毫秒和下一秒的前250毫秒，这种方式能够判断出用户的访问依旧超过了1s的访问数量，因此依然会阻拦用户的访问。

使用定时任务实现的滑动窗口代码如下：

public class SlidingWindowRateLimiter implements RateLimiter, Runnable{
    private final long maxVisitPerSecond;

    private static final int DEFAULT_BLOCK = 10;
    private final int block;
    private final AtomicLong[] countPerBlock;

    private AtomicLong count;
    private volatile int index;

    public SlidingWindowRateLimiter(int block, long maxVisitPerSecond) {
        this.block = block;
        this.maxVisitPerSecond = maxVisitPerSecond;
        countPerBlock = new AtomicLong[block];
        for (int i = 0 ; i< block ; i++) {
            countPerBlock[i] = new AtomicLong();
        }
        count = new AtomicLong(0);
    }

    public SlidingWindowRateLimiter() {
        this(DEFAULT_BLOCK, DEFAULT_ALLOWED_VISIT_PER_SECOND);
    }
    @Override
    public boolean isOverLimit() {
        return currentQPS() > maxVisitPerSecond;
    }

    @Override
    public long currentQPS() {
        return count.get();
    }

    @Override
    public boolean visit() {
        countPerBlock[index].incrementAndGet();
        count.incrementAndGet();
        return isOverLimit();
    }

    @Override
    public void run() {
        System.out.println(isOverLimit());
        System.out.println(currentQPS());
        System.out.println("index:" + index);
        index = (index + 1) % block;
        long val = countPerBlock[index].getAndSet(0);
        count.addAndGet(-val);
    }

    public static void main(String[] args) {
        SlidingWindowRateLimiter slidingWindowRateLimiter = new SlidingWindowRateLimiter(10, 1000);
        ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
        scheduledExecutorService.scheduleAtFixedRate(slidingWindowRateLimiter, 100, 100, TimeUnit.MILLISECONDS);

        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    slidingWindowRateLimiter.visit();
                    try {
                        Thread.sleep(10);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }).start();

        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    slidingWindowRateLimiter.visit();
                    try {
                        Thread.sleep(10);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }).start();
    }
}