最近测试team在测试过程中反馈部分接口需要做一定的限流措施,刚好我也回顾了下限流相关的算法。常见限流相关的算法有四种:计数器算法, 滑动窗口算法, 漏桶算法, 令牌桶算法

1.计数器算法(固定窗口)

 计数器算法是使用计数器在周期内累加访问次数,当达到设定的阈值时就会触发限流策略。下一个周期开始时,清零重新开始计数。此算法在单机和分布式环境下实现都非常简单,可以使用Redis的incr原子自增和线程安全即可以实现

 这个算法常用于QPS限流和统计访问总量,对于秒级以上周期来说会存在非常严重的问题,那就是临界问题,如下图:

 假设我们设置的限流策略时1分钟限制计数100,在第一个周期最后5秒和第二个周期的开始5秒,分别计数都是88,即在10秒时间内计数达到了176次,已经远远超过之前设置的阈值,由此可见,计数器算法(固定窗口)限流方式对于周期比较长的限流存在很大弊端。

 Java 实现计数器(固定窗口):

package com.brian.limit;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

import lombok.extern.slf4j.Slf4j;

/**
 * 固定窗口
 */
@Slf4j
public class FixWindow {

    private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(5);

    private final int limit = 100;

    private AtomicInteger currentCircleRequestCount = new AtomicInteger(0);

    private AtomicInteger timeCircle = new AtomicInteger(0);

    private void doFixWindow() {
        scheduledExecutorService.scheduleWithFixedDelay(() -> {
            log.info(" 当前时间窗口,第 {} 秒 ", timeCircle.get());
            if(timeCircle.get() >= 60) {
                timeCircle.set(0);
                currentCircleRequestCount.set(0);
                log.info(" =====进入新的时间窗口===== ");
            }
            if(currentCircleRequestCount.get() > limit) {
                log.info("触发限流策略,当前窗口累计请求数 : {}", currentCircleRequestCount);
            } else {
                final int requestCount = (int) ((Math.random() * 5) + 1);
                log.info("当前发出的 ==requestCount== : {}", requestCount);
                currentCircleRequestCount.addAndGet(requestCount);
            }
           timeCircle.incrementAndGet();
        }, 0, 1, TimeUnit.SECONDS);
    }

    public static void main(String[] args) {
        new FixWindow().doFixWindow();
    }
    
}

2.滑动窗口算法

 滑动窗口算法是将时间周期拆分成N个小的时间周期,分别记录小周期里面的访问次数,并且根据时间的滑动删除过期的小周期。如下图,假设时间周期为1分钟,将1分钟再分为2个小周期,统计每个小周期的访问数量,则可以看到,第一个时间周期内,访问数量为92,第二个时间周期内,访问数量为104,超过100的访问则被限流掉了。

 

 由此可见,当滑动窗口的格子划分的越多,那么滑动窗口的滚动就越平滑,限流的统计就会越精确。此算法可以很好的解决固定窗口算法的临界问题。

  Java实现滑动窗口:

package com.brian.limit;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import lombok.extern.slf4j.Slf4j;

/**
 * 滑动窗口
 * 
 * 60s限流100次请求
 */
@Slf4j
public class RollingWindow {

    private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(5);

    // 窗口跨度时间60s
    private int timeWindow = 60;

    // 限流100个请求
    private final int limit = 100;

    // 当前窗口请求数
    private AtomicInteger currentWindowRequestCount = new AtomicInteger(0);

    // 时间片段滚动次数
    private AtomicInteger timeCircle = new AtomicInteger(0);

    // 触发了限流策略后等待的时间
    private AtomicInteger waitTime = new AtomicInteger(0);

    // 在下一个窗口时,需要减去的请求数
    private int expiredRequest = 0;

    // 时间片段为5秒,每5秒统计下过去60秒的请求次数
    private final int slidingTime = 5;

    private ArrayBlockingQueue<Integer> slidingTimeValues = new ArrayBlockingQueue<>(11);

    public void rollingWindow() {
        scheduledExecutorService.scheduleWithFixedDelay(() -> {

            if (waitTime.get() > 0) {
                waitTime.compareAndExchange(waitTime.get(), waitTime.get() - slidingTime);
                log.info("=====当前滑动窗口===== 限流等待下一个时间窗口倒计时: {}s", waitTime.get());
                if (currentWindowRequestCount.get() > 0) {
                    currentWindowRequestCount.set(0);
                }
            } else {
                final int requestCount = (int) ((Math.random() * 10) + 7);
                if (timeCircle.get() < 12) {
                    timeCircle.incrementAndGet();
                }
                
            log.info("当前时间片段5秒内的请求数: {} ", requestCount);
            currentWindowRequestCount.addAndGet(requestCount);
            log.info("=====当前滑动窗口===== {}s 内请求数: {} ", timeCircle.get()*slidingTime , currentWindowRequestCount.get());

            if(!slidingTimeValues.offer(requestCount)){
                expiredRequest =  slidingTimeValues.poll();
                slidingTimeValues.offer(requestCount);
            } 

            if(currentWindowRequestCount.get() > limit) {
                // 触发限流
                log.info("=====当前滑动窗口===== 请求数超过100, 触发限流,等待下一个时间窗口 ");
                waitTime.set(timeWindow);
                timeCircle.set(0);
                slidingTimeValues.clear();
            } else {
                // 没有触发限流,滑动下一个窗口需要,移除相应的:在下一个窗口时,需要减去的请求数
                log.info("=====当前滑动窗口===== 请求数 <100, 未触发限流,当前窗口请求总数: {},即将过期的请求数:{}"
                        ,currentWindowRequestCount.get(), expiredRequest);
                currentWindowRequestCount.compareAndExchange(currentWindowRequestCount.get(), currentWindowRequestCount.get() - expiredRequest);
            }
        }   
        }, 5, 5, TimeUnit.SECONDS);
    }

    public static void main(String[] args) {
        new RollingWindow().rollingWindow();
    }
    

}

计数器(固定窗口)和滑动窗口区别:

计数器算法是最简单的算法,可以看成是滑动窗口的低精度实现。滑动窗口由于需要存储多份的计数器(每一个格子存一份),所以滑动窗口在实现上需要更多的存储空间。也就是说,如果滑动窗口的精度越高,需要的存储空间就越大。

3.漏桶算法

 漏桶算法是访问请求到达时直接放入漏桶,如当前容量已达到上限(限流值),则进行丢弃(触发限流策略)。漏桶以固定的速率进行释放访问请求(即请求通过),直到漏桶为空。

 Java实现漏桶:

package com.brian.limit;

import java.util.concurrent.*;

import lombok.extern.slf4j.Slf4j;

/**
 * 漏桶算法
 */
@Slf4j
public class LeakyBucket {
    private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(5);

    // 桶容量
    public  int capacity = 1000;
    
    // 当前桶中请求数
    public int curretRequest = 0;

    // 每秒恒定处理的请求数
    private final int handleRequest = 100;

    public void doLimit() {
        scheduledExecutorService.scheduleWithFixedDelay(() -> {
            final int requestCount = (int) ((Math.random() * 200) + 50);
            if(capacity > requestCount){
                capacity -= requestCount;
                log.info("<><>当前1秒内的请求数:{}, 桶的容量:{}", requestCount, capacity);
                if(capacity <=0) {
                    log.info(" =====触发限流策略===== ");
                } else {
                    capacity += handleRequest;
                    log.info("<><><><>当前1秒内处理请求数:{}, 桶的容量:{}", handleRequest, capacity);
                }
            } else {
                log.info("<><><><>当前请求数:{}, 桶的容量:{},丢弃的请求数:{}", requestCount, capacity,requestCount-capacity);
                if(capacity <= requestCount) {
                    capacity = 0;
                }
                capacity += handleRequest;
                log.info("<><><><>当前1秒内处理请求数:{}, 桶的容量:{}", handleRequest, capacity);
            }
        }, 0, 1, TimeUnit.SECONDS);
    }

    public static void main(String[] args) {
        new LeakyBucket().doLimit();
    }
}

 漏桶算法有个缺点:如果桶的容量过大,突发请求时也会对后面请求的接口造成很大的压力。

4.令牌桶算法

 令牌桶算法是程序以恒定的速度向令牌桶中增加令牌,令牌桶满了之后会丢弃新进入的令牌,当请求到达时向令牌桶请求令牌,如获取到令牌则通过请求,否则触发限流策略。

 

 Java实现令牌桶:

package com.brian.limit;

import java.util.concurrent.*;

import lombok.extern.slf4j.Slf4j;
/**
 * 令牌桶算法
 */
@Slf4j
public class TokenBucket {
    private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(5);

    // 桶容量
    public  int capacity = 1000;
    
    // 当前桶中请求数
    public int curretToken = 0;

    // 恒定的速率放入令牌
    private final int tokenCount = 200;

    public void doLimit() {
        scheduledExecutorService.scheduleWithFixedDelay(() -> {
            
            new Thread( () -> {
                if(curretToken >= capacity) {
                    log.info(" =====桶中的令牌已经满了===== ");
                    curretToken = capacity;
                } else {
                    if((curretToken+tokenCount) >= capacity){
                      log.info(" 当前桶中的令牌数:{},新进入的令牌将被丢弃的数: {}",curretToken,(curretToken+tokenCount-capacity));
                      curretToken = capacity;
                  } else {
                      curretToken += tokenCount;
                  }
                }
            }).start();

            new Thread( () -> {
                final int requestCount = (int) ((Math.random() * 200) + 50);
                if(requestCount >= curretToken){
                    log.info(" 当前请求数:{},桶中令牌数: {},将被丢弃的请求数:{}",requestCount,curretToken,(requestCount - curretToken));
                    curretToken = 0;
                } else {
                    log.info(" 当前请求数:{},桶中令牌数: {}",requestCount,curretToken);
                    curretToken -= requestCount;
                }
            }).start();
        }, 0, 500, TimeUnit.MILLISECONDS);
    }

    public static void main(String[] args) {
        new TokenBucket().doLimit();
    }
    
}

漏桶算法和令牌桶算法区别:

令牌桶可以用来保护自己,主要用来对调用者频率进行限流,为的是让自己不被打垮。所以如果自己本身有处理能力的时候,如果流量突发(实际消费能力强于配置的流量限制),那么实际处理速率可以超过配置的限制。而漏桶算法,这是用来保护他人,也就是保护他所调用的系统。主要场景是,当调用的第三方系统本身没有保护机制,或者有流量限制的时候,我们的调用速度不能超过他的限制,由于我们不能更改第三方系统,所以只有在主调方控制。这个时候,即使流量突发,也必须舍弃。因为消费能力是第三方决定的。
总结起来:如果要让自己的系统不被打垮,用令牌桶。如果保证被别人的系统不被打垮,用漏桶算法

 

参考博客:https://blog.csdn.net/weixin_41846320/article/details/95941361

     https://www.cnblogs.com/xuwc/p/9123078.html

原文链接:https://www.cnblogs.com/hlkawa/p/13111003.html

最后修改日期:2020年6月14日