【工程化】限流

摘要：限流，是对流量控制。基于时间的滑动窗口，参照于滑动窗口，将单位时间看做是一个窗口，将窗口中的每个格子设定为指定时间间隔，为格子总数，那么单位时间就是。很明显格子划分的越多，滑动窗口的滑动就越平滑，限流统计就越精确。

限流，在一些我们已知的场景有：

1）在Tcp协议中，Flow Control，

流量控制以动态调整发送空间大小(滑动窗口)的形式来反映接收端接收消息的能力，反馈给发送端以调整发送速度，避免发送速度过快导致的丢包或者过慢降低整体性能。

在Tcp协议中，通过在首部设置window size的值来控制窗口大小。

2) 在Web sever中，使用nginx对请求访问进行限流，基于nginx扩展模块，

ngx_http_limit_conn_module，可以针对Ip或Domain来限制连接数。

ngx_http_limit_req_module，可以通过自定义键值来限制请求处理的频率。

3) 在现在一些主流的开放平台，都有针对API调用的限制，比如淘宝开放平台针对商品查询接口的限制次数，百度地图针对开发者““地点检索”API是有指定限额的。这些都是针对API的限流。

4) 秒杀系统中，由于库存量一般是很少的，对应只有少部分的用户才能秒杀成功，因此我们要限制绝大部分用户流量。

5） 各种框架使用时的数量限制，如数据连接池最大连接数、线程池最大线程数、zk最大连接等等

以上是限流的常见场景。限流，是对流量控制。

Flow Control，是在数据通信中，流量控制是管理两个节点之间的数据传输速率的过程，以防止快速发送者压倒慢速接收者。
它为接收机提供了一种控制传输速度的机制，使接收节点不会被来自发送节点的数据淹没。

Rate Limiting，在计算机网络中，网络接口控制器用限流来控制发送和接收的流量速率，以防止Dos攻击。

可以看出，限流主要是控制发送和接受双方的流量速率，保证工作正常进行。

为什么要限流

限流，为了保护系统在应对流量高峰时，系统能够依然以可控的处理速率对外提供服务，而不至于奔溃或变为不可服务的。
这也从侧面体现了系统服务的稳定性，如果是SOA服务的话，也体现了服务设计原则的自治性。

计数器 Counter

基于时间滑动窗口Timed Sliding Window

漏桶Leaky bucket

令牌桶Token bucket

注：以下算法只做算法演示，肯定有很多细节未考虑，包括在多线程下行为是否正确等

计数器是最简单的一种，针对资源设置访问最大总量（上限）Max，以及定义一个计数器Counter，每当需要对资源访问时，Counter++，当Counter小于Max,访问可以通过，否则不可用。

一般这个场景在项目中比较常见，比如我们使用Semphore的acquire、release来控制多线程对资源的许可，比如Jedis Pool的对象池borrow、return。

基于单位时间的计数器

限制指定时间内的请求数量，比如1秒内最大的请求量为2个

demo如下：

public class PerTimeUnitCounterFlowControl {
    private static final long INTERVAL = 5 * 1000;//时间间隔ms
    private long timestamp;
    private int counter;
    private int limit;
    private long interval;

    public PerTimeUnitCounterFlowControl(long interval,int limit) {
        this.interval = interval <= 0? INTERVAL:interval;
        this.timestamp = SystemClock.now();
        this.limit = limit;
    }

    /**
     *
     * @return
     */
    public boolean acquire(){
        long now = SystemClock.now();
        if (now < timestamp + interval){
            counter++;
            return counter <= limit;
        }
        timestamp = now;
        counter = 1;
        return true;
    }
}

该算法的缺陷是，在时间节点重置的时隙里可能被突发请求超限。

Timed Sliding Window，参照于Tcp滑动窗口，将单位时间T看做是一个窗口，将窗口中的每个格子设定为指定时间间隔Duration，Window Size为格子总数 buckets，那么单位时间就是buckets * Duration。每个格子有自己独立的计数器。当时间每过去Duration时候，窗口就会向右滑动一个格子。

如下：

每当有请求过来时，都会落在指定格子里，然后获取当前窗口的所有计数器之和，以此来触发是否限流。

很明显格子划分的越多，滑动窗口的滑动就越平滑，限流统计就越精确。

demo如下：

public class TimedSlidingWindowFlowControl {
    private static final int LIMIT = 5;
    private long duration;//每个格子的时长
    private int bucketSize;//总格子数
    private final long windowTime;
    private final ScheduledExecutorService scheduledExecutor;
    private long startedTimestamp;
    private volatile int head;//指向第一个格子
    private AtomicInteger[] buckets;

    public TimedSlidingWindowFlowControl(long duration, int bucketSize) {
        this.duration = duration;
        this.bucketSize = bucketSize;
        this.scheduledExecutor = Executors.newSingleThreadScheduledExecutor();
        this.windowTime = duration * bucketSize;
        buckets = new AtomicInteger[bucketSize];
    }

    /**
     * 初始化
     */
    protected void init(){
        startedTimestamp = SystemClock.now();
        for(int i = 0; i < bucketSize;i++){
            buckets[i] = new AtomicInteger(0);
        }
        head = 0;//指向第一个格子
        scheduledExecutor.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                timeRolling();
            }
        },duration/2,duration, TimeUnit.MILLISECONDS);
    }

    /**
     * 获取许可
     * @return
     */
    private boolean acquire(){
        long now = SystemClock.now();
        long timestampDiff = now - startedTimestamp;
        long mask = timestampDiff % (windowTime);
        //相对于head的位置
        int idx = getBucketIndex(mask);
        if(idx == -1){
            throw new IllegalStateException("illegalState");
        }
        buckets[idx].incrementAndGet();
        int count = getCurrentCount();
        System.out.println("当前count:" + count);
        if(count <= LIMIT){
            return true;
        }
        return false;
    }

    /**
     * 查找当前的位置
     * @param mask
     * @return
     */
    private int getBucketIndex(long mask){
        int cursor = head;
        int stopIndex = cursor;
        if(mask <= duration){
            return cursor;
        }
        long d = duration;
        while (true){
            cursor = (cursor + 1) % bucketSize;
            if(cursor == stopIndex){
                return -1;
            }
            d = d + duration;
            if(mask <= d){
                return cursor;
            }
        }
    }

    /**
     * 获取当前计数
     * @return  int
     */
    private int getCurrentCount(){
        return Arrays.stream(buckets).mapToInt(buckets -> buckets.get()).sum();
    }

    /**
     * 时间滚动
     */
    private void timeRolling(){
        //每次格子移动都会更改head
        int last = head;
        head = (head + 1) % bucketSize;
        System.out.println("时间向前移动一格：" + head);
        buckets[last].set(0);//reset
    }

    /**
     * 关闭
     */
    protected void shutdown() throws InterruptedException {
        scheduledExecutor.shutdown();
        scheduledExecutor.awaitTermination(5,TimeUnit.SECONDS);
    }
}

上面的demo可能有些细节未考虑，基本思路是使用定时任务模拟时钟滚动，循环复用计数器桶，使用head指针始终指向第一个桶，统计所有的桶计数的和 判断是否触发限流。
实际上考虑“使用定时任务模拟时钟滚动”，这种方式有一些缺点：会浪费Cpu资源，而且还依赖时钟。可以考虑采用类似Guava的RateLimiter的延迟计算机制。
另更多关于滑动窗口计数可以参考Storm的RollingCountBolt和Hystrix的Metrics实现。

漏桶算法，即Leaky bucket，是一种很常用的限流算法，可以用来实现流量整形（Traffic Shaping）和流量控制（Traffic Policing）。

以下是wikipedia对Leaky bucket的算法描述：

[A] fixed capacity bucket, associated with each virtual connection or user, leaks at a fixed rate.
If the bucket is empty, it stops leaking.
For a packet to conform, it has to be possible to add a specific amount of water to the bucket: The specific amount added by a conforming packet can be the same for all packets, or can be proportional to the length of the packet.
If this amount of water would cause the bucket to exceed its capacity then the packet does not conform and the water in the bucket is left unchanged.

翻译过来的意思为：

一个固定容量的漏桶，按照常量固定速率流出水滴；

如果桶是空的，则不需流出水滴；

可以以任意速率流入水滴到漏桶；

如果流入水滴超出了桶的容量，则流入的水滴溢出了（被丢弃），而漏桶容量是不变的。

这个可以使用基于生产者-消费者共享阻塞队列实现。

demo如下：

public class LeakyBucketFlowControl{
    private int capacity;
    private LinkedBlockingQueue bucket;
    private int flowOutNum;//以恒定的速率流出
    private int flowOutTimeUnit;//
    private static final int VALUE = 1;
    private Thread thread;
    private volatile boolean stop = false;

    public LeakyBucketFlowControl(int capacity, int flowOutNum, int flowOutTimeUnit) {
        this.capacity = capacity;
        this.flowOutNum = flowOutNum;
        this.flowOutTimeUnit = flowOutTimeUnit;
        this.bucket = new LinkedBlockingQueue<>(capacity);
        this.thread = new Thread(new Worker());
    }

    /**
     * init
     */
    public void init(){
        thread.start();
    }

    /**
     * 获取许可
     * @return
     */
    protected boolean acquire(){
        boolean of = bucket.offer(VALUE);
        return of;
    }

    /**
     * shutdown
     */
    public void shutdown(){
        stop = true;
        System.out.println("当前漏桶的容量：" + bucket.size());
    }

    /**
     * 内部worker
     */
    class Worker implements Runnable{

        @Override
        public void run() {
            while (!Thread.currentThread().isInterrupted() && !stop){
                try {
                    TimeUnit.MILLISECONDS.sleep(flowOutTimeUnit);
                    for(int i = 1;i <= flowOutNum;i++){
                        bucket.take();
                    }
                    System.out.println("漏桶流出容量为：" + bucket.size());
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }
}

一般情况下，漏桶对于流量整形有用， 无论什么样的请求速率进来，漏桶总是以恒定的速率执行，但对于突发传输有一定限制，除非当前漏桶已经为空。

关于令牌token的使用场景比较多了，比如Auth的access token，计算机网络中轮转访问MAC协议中的Token passing等。

现在是关于token在限流中的算法描述：

[A] token is added to the bucket every  1/r seconds.
The bucket can hold at the most  b tokens. If a token arrives when the bucket is full, it is discarded.
When a packet (network layer PDU) of n bytes arrives, n tokens are removed from the bucket, and the packet is sent to the network.
If fewer than n tokens are available, no tokens are removed from the bucket, and the packet is considered to be non-conformant.

令牌将按照固定的速率被放入令牌桶中。比如每秒放10个。

桶中最多存放b个令牌，当桶满时，新添加的令牌被丢弃或拒绝。

当一个n个字节大小的数据包到达，将从桶中删除n个令牌，接着数据包被发送到网络上。

如果桶中的令牌不足n个，则不会删除令牌，且该数据包将被限流（要么丢弃，要么缓冲区等待）。

算法实现直接使用Guava的RateLimiter

public class RateLimiterTester {

    public static void main(String[] args) {
        RateLimiter limiter = RateLimiter.create(2);//发令牌的间隔时间约500ms
        double x = limiter.acquire(5) * 1000;
        System.out.println(x + "....");
        for (int i = 1;i <= 5;i++){
            double y = limiter.acquire()  * 1000;
            System.out.println(y);
        }
    }
}

输出
0.0....
2497.7299999999996
491.842
495.838
497.392
498.442

令牌桶算法可以应对突发流量，RateLimiter提供了SmoothBursty和SmoothWarmingUp两种需求。具体区别和实现可以自行查看下文档或网上找下相关分析。

推荐关于Hystrix一篇好文。