解决方案-限流器 | HoleLin's Blog

参考文献

限流的目的

高并发系统中有三把利器用来保护系统:缓存,降级和限流.限流的目的是为了保护系统不被大量请求冲垮,限制系统的输入和输出流量已达到保护系统的目的.在电商的秒杀活动中,限流是必不可少的一个环节.
常见的限流算法有:令牌桶、漏桶.计数器也可以进行限流实现.

令牌桶

令牌桶算法是一个存放固定容量令牌的桶,按照固定速率往桶里添加令牌.可以控制流量也可以控制并发量,假如我们想要控制 API 网关的并发量最高为 1000,可以创建一个令牌桶,以固定的速度往桶里添加令牌,超出了 1000 则不添加.
当一个请求到达之后就从桶中获取一个令牌,如果能获取到令牌就可以继续往下请求,获取不到就说明令牌不够,并发量达到了最高,请求就被拦截.

漏桶

漏桶是一个固定容量的桶,按照固定的速率流出,可以以任意的速率流入到漏桶中,超出了漏桶的容量就被丢弃,总容量是不变的.但是输出的速率是固定的,无论你上面的水流入的多快,下面的出口只有这么大,就像水坝开闸放水一样

常见的限流方式

限制总并发数(如数据库连接池、线程池)
限制瞬时并发数(nginx的limit_req_conn模块,用来限制瞬时并发连接数)
限制时间窗口内的平均速率(如Guava的RateLimiter、nginx的limit_req_zone模块,限制每秒的平均速率
其他的还有限制远程接口调用速率、限制MQ的消费速率.
另外还可以根据网络连接数、网络流量、CPU或内存负载等来限流.

限流算法

固定时间窗口(计数器)

原理:对一段固定时间窗口内的请求进行计数,如果请求数超过了阈值,则拒绝该请求；如果没有达到设定的阈值,则接受该请求,且计数加1.当时间窗口结束时,重置计数器为0.
优点:实现简单,容易理解
缺点:
- 一段时间内服务不可用
- 限流策略过于粗略,不够平滑,无法应对两个时间窗口临界时间内的突发流量.
Java内部也可以通过原子类计数器AtomicInteger、Semaphore信号量来做简单的限流.

// 限流的个数
 private int maxCount = 10;
 // 指定的时间内
 private long interval = 60;
 // 原子类计数器
 private AtomicInteger atomicInteger = new AtomicInteger(0);
 // 起始时间
 private long startTime = System.currentTimeMillis();

 public boolean limit(int maxCount, int interval) {
     atomicInteger.addAndGet(1);
     if (atomicInteger.get() == 1) {
         startTime = System.currentTimeMillis();
         atomicInteger.addAndGet(1);
         return true;
     }
     // 超过了间隔时间,直接重新开始计数
     if (System.currentTimeMillis() - startTime > interval * 1000) {
         startTime = System.currentTimeMillis();
         atomicInteger.set(1);
         return true;
     }
     // 还在间隔时间内,check有没有超过限流的个数
     if (atomicInteger.get() > maxCount) {
         return false;
     }
     return true;
 }

滑动时间窗口

滑动窗口算法在固定窗口的基础上,将一个计时窗口分成了若干个小窗口,然后每个小窗口维护一个独立的计数器.
优点:准确性更高.
缺点:没有从根本上解决临界问题.基于时间窗口的限流算法,不管是固定时间窗口还是滑动时间窗口,只能在选定的时间粒度上限流,对选定时间粒度内的更加细粒度的访问频率不做限制.

package com.holelin.sundry.test;

import java.util.concurrent.atomic.AtomicInteger;

public class SlidingWindow {

    private AtomicInteger[] timeSlices;
    /* 队列的总长度  */
    private final int timeSliceSize;
    /* 每个时间片的时长 */
    private final long timeMillisPerSlice;
    /* 窗口长度 */
    private final int windowSize;

    /* 当前所使用的时间片位置 */
    private AtomicInteger cursor = new AtomicInteger(0);

    public 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;
        }
    }

    public SlidingWindow(int windowSize, Time timeSlice) {
        this.timeMillisPerSlice = timeSlice.millis;
        this.windowSize = windowSize;
        // 保证存储在至少两个window
        this.timeSliceSize = windowSize * 2 + 1;

        init();
    }

    /**
     * 初始化
     */
    private void init() {
        AtomicInteger[] localTimeSlices = new AtomicInteger[timeSliceSize];
        for (int i = 0; i < timeSliceSize; i++) {
            localTimeSlices[i] = new AtomicInteger(0);
        }
        timeSlices = localTimeSlices;
    }

    private int locationIndex() {
        long time = System.currentTimeMillis();
        return (int) ((time / timeMillisPerSlice) % timeSliceSize);
    }

    /**
     * <p>对时间片计数+1,并返回窗口中所有的计数总和
     * <p>该方法只要调用就一定会对某个时间片进行+1
     *
     * @return
     */
    public int incrementAndSum() {
        int index = locationIndex();
        int sum = 0;
        // cursor等于index,返回true
        // cursor不等于index,返回false,并会将cursor设置为index
        int oldCursor = cursor.getAndSet(index);
        if (oldCursor == index) {
            // 在当前时间片里继续+1
            sum += timeSlices[index].incrementAndGet();
        } else {
            //轮到新的时间片,置0,可能有其它线程也置了该值,容许
            timeSlices[index].set(0);
            // 清零,访问量不大时会有时间片跳跃的情况
            clearBetween(oldCursor, index);

            sum += timeSlices[index].incrementAndGet();
        }

        for (int i = 1; i < windowSize; i++) {
            sum += timeSlices[(index - i + timeSliceSize) % timeSliceSize].get();
        }
        return sum;
    }

    /**
     * 判断是否允许进行访问,未超过阈值的话才会对某个时间片+1
     *
     * @param threshold
     * @return
     */
    public boolean allow(int threshold) {
        int index = locationIndex();
        int sum = 0;
        int oldCursor = cursor.getAndSet(index);
        if (oldCursor != index) {
            timeSlices[index].set(0);
            clearBetween(oldCursor, index);
        }
        for (int i = 0; i < windowSize; i++) {
            sum += timeSlices[(index - i + timeSliceSize) % timeSliceSize].get();
        }

        // 阈值判断
        if (sum < threshold) {
            // 未超过阈值才+1
            timeSlices[index].incrementAndGet();
            return true;
        }
        return false;
    }

    /**
     * <p>将fromIndex~toIndex之间的时间片计数都清零
     * <p>极端情况下,当循环队列已经走了超过1个timeSliceSize以上,这里的清零并不能如期望的进行
     *
     * @param fromIndex 不包含
     * @param toIndex   不包含
     */
    private void clearBetween(int fromIndex, int toIndex) {
        for (int index = (fromIndex + 1) % timeSliceSize; index != toIndex; index = (index + 1) % timeSliceSize) {
            timeSlices[index].set(0);
        }
    }

    public static void main(String[] args) {
        SlidingWindow window = new SlidingWindow(5, Time.MILLISECONDS);
        for (int i = 0; i < 10; i++) {
            System.out.println(window.allow(3));
        }
    }
}

漏桶算法

思路:水(请求)先进入到漏桶里,漏桶以一定速度向外出水.当水流入速度过大,桶会直接溢出.即请求进入一个固定容量的Queue,若Queue满,则拒绝新的请求,可以阻塞,也可以抛异常.
这种模型其实非常类似MQ的思想,利用漏桶削峰填谷,使得Queue的下游具有一个稳定流量.
实现:生产者和消费者.
优点:从出口处限制请求速率,并不存在上面计数器法的临界问题,请求曲线始终是平滑的.
缺点:对请求的过滤太精准了,不允许任何的突发流量.比如我们限制每秒下单 10 万次,那第10 万零 1 次的请求,就会被拒绝.大部分业务场景下,虽然限流了,但还是希望系统允许一定的突发流量,这时候就需要令牌桶算法.

令牌桶算法

令牌桶算法的原理也比较简单,我们可以理解成医院的挂号看病,只有拿到号以后才可以进行诊病.
系统会维护一个令牌(token)桶,以一个恒定的速度往桶里放入令牌(token),这时如果有请求进来想要被处理,则需要先从桶里获取一个令牌(token),当桶里没有令牌(token)可取时,则该请求将被拒绝服务.令牌桶算法通过控制桶的容量、发放令牌的速率,来达到对请求的限制.
令牌桶算法并不能实际的控制速率.比如,10秒往桶里放入10000个令牌桶,即10秒内只能处理10000个请求,那么qps就是1000.但这种模型可以出现1秒内把10000个令牌全部消费完,即qps为10000.所以令牌桶算法实际是限制的平均流速.具体控制的粒度以放令牌的间隔和每次的量来决定.若想要把流速控制的更加稳定,就要缩短间隔时间.
优点:允许一定程度流量突发,但不会超过设置阈值,对用户友好同时有效保护系统.
缺点:请求异步处理,无法同步返回