限流介绍

​ 限流算法基本上有三种:漏桶、令牌桶和信号量这三种,本文是关于漏桶和令牌桶的限流的实现方式.

限流注意的点是多线程下限流算法的修改并发问题

一、令牌桶

1.话不多说,上代码

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
public class TokenBucket {
    // 上一次令牌发放时间
    public static long lastTime = 0;
    // 桶的容量
    public static int capacity = 1;
    // 令牌生成速度 /s
    public static int rate = 1;
    // 当前令牌数量
    public static AtomicLong tokens = new AtomicLong(0);


    //返回值说明：
    // false 没有被限制到
    // true 被限流
    public static synchronized boolean isLimited(int applyCount) {
        long now = System.currentTimeMillis();
        //时间间隔,单位为 ms
        long gap = now - lastTime;
        //场景三：当前请求和上次请求，在同一个时间区间
        //当前时间，在时间区间之内
        //以时间区间为计算维度，同一个区间，没有必要重复去计算令牌数量
        if (lastTime != 0 && gap < 1000) {
            if (tokens.get() < applyCount) {
                // 若拿不到令牌,则拒绝
                return true;
            } else {
                // 还有令牌，领取令牌
                tokens.getAndAdd(-applyCount);
                return false;
            }
        }
        log.info("overGap gap = " + gap);
        if (lastTime == 0) {
            gap = 1000;
        }
        //计算时间段内的令牌数
        int reverse_permits = (int) (gap * rate / 1000);
        int all_permits = Math.toIntExact(tokens.get() + reverse_permits);
        // 当前令牌数
        tokens.set(Math.min(capacity, all_permits));
        log.info("tokens {} capacity {} gap {} ", tokens, capacity, gap);
        lastTime = now;
        if (tokens.get() < applyCount) {
            // 若拿不到令牌,则拒绝
            return true;
        } else {
            // 还有令牌，领取令牌
            tokens.getAndAdd(-applyCount);
            return false;
        }
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        long start = System.currentTimeMillis();
        for (int i = 0; i < 2; i++) {
            Thread.sleep(200);
            CompletableFuture.runAsync(() -> {
                int x = 0;
                for (int j = 0; j < 5; j++) {
                    System.out.println("Request " + x++ + ": " + (isLimited(1) ? "Rejected" : "Accepted"));
                }
            });
        }
        Thread.sleep(1500);
    }
}

​

二、漏桶

1.话不多说,上代码

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.atomic.AtomicLong;

@Slf4j
public class LeakBucketLimiter {
    // 计算的起始时间
    private static long lastOutTime = System.currentTimeMillis();

    // 时间区间的时间间隔 ms
    private static long interval = 1000;

    // 流出速率 每秒 2 次
    private static int leakRate = 1;

    // 桶的容量
    private static int capacity = 1;

    //剩余的水量
    private static AtomicLong waterInBucket = new AtomicLong(0);

    //返回值说明：
    // false 没有被限制到
    // true 被限流
    public static synchronized boolean isLimit() {
        // 如果是空桶，就当前时间作为漏出的时间
        if (waterInBucket.get() == 0) {
            System.out.println("汉字666");
            lastOutTime = System.currentTimeMillis();
            waterInBucket.addAndGet(1);
            return false;
        }
        //场景三：当前请求和上次请求，在同一个时间区间
        long nowTime = System.currentTimeMillis();
        //当前时间，在时间区间之内
        //漏水以时间区间为计算维度，同一个区间，没有必要重复去计算漏水
        if (nowTime < lastOutTime + interval) {
            // 尝试加水,并且水还未满 ，放行
            if ((waterInBucket.get()) < capacity) {
                waterInBucket.addAndGet(1);
                return false;
            } else {
                // 水满，拒绝加水， 限流
                return true;
            }
        }
        //场景二： 桶里边有水
        //当前时间，在时间区间之外
        // 计算漏水，以时间的区间为维度的
        int waterLeaked = ((int) ((System.currentTimeMillis() - lastOutTime) / 1000)) * leakRate;
        // 计算剩余水量
        int waterLeft = Math.toIntExact(waterInBucket.get() - waterLeaked);
        //校正数据
        waterLeft = Math.max(0, waterLeft);
        waterInBucket.set(waterLeft);
        // 重新更新leakTimeStamp
        lastOutTime = System.currentTimeMillis();
        // 尝试加水,并且水还未满 ，放行
        if ((waterInBucket.get()) < capacity) {
            waterInBucket.addAndGet(1);
            return false;
        } else {
            // 水满，拒绝加水， 限流
            return true;
        }
    }

    public static void main(String[] args) throws InterruptedException {
        long start = System.currentTimeMillis();
        for (int i = 0; i < 3; i++) {
            int x = 0;
            for (int j = 0; j < 5; j++) {
                System.out.println((System.currentTimeMillis()-start)+" Request " + x++ + ": " + (isLimit() ? "Rejected" : "Accepted"));
                Thread.sleep(200);
            }
        }
    }

}

三、写在最后

​ 1.在spring全局过滤器中使用限流算法可以有效的限制单位时间的访问应用的流量

​ 2.这个demo只是针对单个应用的所有请求的限流,如果要针对到每个人或者每个session的话,实现思路有:可以为每个用户创建一个限流器,然后保存到本地缓存中(可能有更好的实现方式)

​ 3.如果是分布式的应用集群,可以将用户的登录信息保存到redis当中,使用redis+lua脚本的方式或者在nginx+lua的方式进行限流操作(没做过,应该有坑)