限流算法原理和实现

Posted 2021-09-25 Java攻城师

限流原理与实战

接口限流的算法主要有3种，分别是计数器限流、漏桶算法和令牌桶算法

计数器限流原理

原理：在一个时间窗口（间隔）内，所处理的请求的最大数量是有限制的，对超过限制的部分请求不做处理。

代码实现

@Slf4j
public class CountLimiter {
  
    private static long startTime = System.currentTimeMillis();
      //时间间隔
    private static long interval = 1000;
      //时间间隔内最大处理请求数
    private static long maxCount = 2;
      //计数器
    private static AtomicInteger accumulator = new AtomicInteger();

    //在1秒内，只允许2个请求接入，如若查过时间片，则初始化参数进入新的一轮时间片
    private static long tryAcquire(long taskId, int turn){
        long nowTime = System.currentTimeMillis();
          //在时间段内，且数量小于等于最大允许请求值，则返回数量
        if (nowTime < startTime + interval){
            int count = accumulator.incrementAndGet();
            if (count <= maxCount){
                return count;
            }else {
                return -count;
            }
        }else {
        //不为一个时间段，则重置计数器和开始时间
            synchronized (CountLimiter.class){
                log.info("新时间区到了，taskId{}, turn{}..", taskId, turn);
                if (nowTime > startTime + interval){
                    accumulator.set(0);
                    startTime = nowTime;
                }
            }
            return 0;
        }
    }

    private ExecutorService pool = Executors.newFixedThreadPool(10);


    @Test
    public void testLimit(){
        AtomicInteger limited = new AtomicInteger(0);
        final int threads = 2;
        final int turns = 20;
        CountDownLatch countDownLatch = new CountDownLatch(threads);
        long start = System.currentTimeMillis();
        for (int i = 0; i < threads;i++){
            pool.submit(() -> {
                try {
                    for (int j = 0; j < turns; j++) {
                        long taskId = Thread.currentThread().getId();
                        long index = tryAcquire(taskId, j);
                        if (index <= 0){
                            limited.getAndIncrement();
                        }
                        Thread.sleep(200);
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                countDownLatch.countDown();
            });
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        float time = (System.currentTimeMillis() - start) /1000F;
        log.info("限制次数为：" + limited.get() + "，通过次数为:"+(threads*turns - limited.get()));

        log.info("限制比例为："+(float)limited.get()/((float) threads*turns));
        log.info("运行时长："+time);
    }
}

复制代码

漏桶限流原理和Java参考实现

漏桶限流的基本原理

1）水通过进水口（对应客户端请求）以任意速率流入漏桶。 2）漏桶的容量是固定的，出水（放行）速率也是固定的。 3）漏桶容量是不变的，如果处理速度太慢，桶内水量会超出桶的容量，后面流入的水就会溢出，表示请求拒绝。

 private static long lastOutTime = System.currentTimeMillis();

    //流出速率每秒2个
    private static int rate = 2;

    //剩余水的量
    private static long water = 0;

    /**
     * false:没有被限制
     * true:被限流
     * @param taskId
     * @param turns
     * @return
     */
    public synchronized static boolean tryAcquire(long taskId, int turns){
        long now = System.currentTimeMillis();
        long pastTime = now - lastOutTime;
        long outWater = pastTime * rate/ 1000;
        water = water -outWater;
        log.info("water {} pastTime {} outWater {}",water ,pastTime, outWater);

        if (water < 0){
            water = 0;
        }
        if (water <= 1){
            lastOutTime = now;
            water ++ ;
            return false;
        }else {
            return true;
        }
    }
复制代码

令牌桶限流原理和Java参考实现

令牌桶限流大致的规则如下：

（1）进水口按照某个速度向桶中放入令牌。 （2）令牌的容量是固定的，但是放行的速度是不固定的，只要桶中还有剩余令牌，一旦请求过来就能申请成功，然后放行。 （3）如果令牌的发放速度慢于请求到来的速度，桶内就无令牌可领，请求就会被拒绝。

@Slf4j
public class TokenBucketLimiter {
    //上一次令牌发放的时间
    public long lastTime = System.currentTimeMillis();
    //桶的容量
    public int capacity = 2;
    //令牌生成速度个/秒
    public int rate = 2;
    //当前令牌的数量
    public int tokens;

    //返回值说明
    /**
     * false:没有被限制
     * true:被限流
     * @param taskId
     * @param turns
     * @return
     */
    public synchronized boolean tryAcquire(long taskId, int applyCount){
        long now = System.currentTimeMillis();
        //时间间隔
        long gap = now - lastTime;
        //当前令牌数
        tokens = Math.min(capacity, (int)(tokens+gap*rate/1000));
        log.info("tokens {} capacity {} gap {}",tokens ,capacity, gap);
        if (tokens < applyCount){
            log.info("被限流了.. {} ,applyCount:{}",taskId,applyCount);
            return true;
        }else {
            tokens -= applyCount;
            lastTime = now;
            log.info("剩余令牌.." + tokens);
            return false;
        }
    }

    private ExecutorService pool = Executors.newFixedThreadPool(10);

    @Test
    public void testLimit(){
        AtomicInteger limited = new AtomicInteger(0);
        final int threads = 2;
        final int turns = 20;
        CountDownLatch countDownLatch = new CountDownLatch(threads);
        long start = System.currentTimeMillis();
        for (int i = 0; i < threads;i++){
            pool.submit(() -> {
                try {
                    for (int j = 0; j < turns; j++) {
                        long taskId = Thread.currentThread().getId();
                        boolean isLimited = tryAcquire(taskId, 1);
                        if (isLimited){
                            limited.getAndIncrement();
                        }
                        Thread.sleep(200);
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                countDownLatch.countDown();
            });
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        float time = (System.currentTimeMillis() - start) /1000F;
        log.info("限制次数为：" + limited.get() + "，通过次数为:"+(threads*turns - limited.get()));

        log.info("限制比例为："+(float)limited.get()/((float) threads*turns));
        log.info("运行时长："+time);

    }

}

参考：《2020最新Java基础精讲视频教程和学习路线！》
链接：https://juejin.cn/post/693747...