常用限流算法及自定义限流框架

限流算法

1、计数器

在一段时间内计数，与阈值进行比较，到了临界时间点计数器清零。该算法优点为实现简单，缺点是无法应对突发流量激增

public class CounterLimiter {
    private static long timeStamp = System.currentTimeMillis();
    // 限制为1s内限制在100个请求
    private static long limitCount = 10;
    // 时间间隔（毫秒）
    private static long interval = 1000;
    // 请求数
    private static AtomicLong reqCount = new AtomicLong(0);

    public static boolean grant() {
        long now = System.currentTimeMillis();
        if (now < timeStamp + interval) {
            if (reqCount.incrementAndGet() < limitCount) {
                return true;
            } else {
                return false;
            }
        } else {
            timeStamp = System.currentTimeMillis();
            reqCount = new AtomicLong(0);
            return false;
        }
    }

    public static void main(String[] args) throws IOException, InterruptedException {
        for (int i = 0; i < 20; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    if (grant()) {
                        System.out.println("执行业务逻辑");
                    } else {
                        System.out.println("限流");
                    }
                }
            }).start();
        }
    }

}

2、滑动窗口

将时间划分为多个区间，在每个区间有一次请求则将区间计数器加一，维持一个时间区间，占据多个区间，每经过一个时间区间则抛弃最老的一个区间同时纳入一个最新的区间，如果当前窗口区间内的请求总数大于限制总数则限流。该算法有点避免了固定窗口带来的突发流量激增，缺点为时间区间精度越高所需空间容量就越大。

public class SlidingWindowLimiter {
    /**
     * 循环队列，就是装多个窗口用，该数量是windowSize的2倍
     */
    private AtomicInteger[] timeSlices;
    /**
     * 队列的总长度
     */
    private int timeSliceSize;
    /**
     * 每个时间片的时长，以毫秒为单位
     */
    private int timeMillisPerSlice;
    /**
     * 共有多少个时间片（即窗口长度）
     */
    private int windowSize;
    /**
     * 在一个完整窗口期内允许通过的最大阈值
     */
    private int threshold;
    /**
     * 该滑窗的起始创建时间，也就是第一个数据
     */
    private long beginTimestamp;
    /**
     * 最后一个数据的时间戳
     */
    private long lastAddTimestamp;


    public SlidingWindowLimiter(int timeMillisPerSlice, int windowSize, int threshold) {
        this.timeMillisPerSlice = timeMillisPerSlice;
        this.windowSize = windowSize;
        this.threshold = threshold;
        // 保证存储在至少两个window
        this.timeSliceSize = windowSize * 2;

        reset();
    }

    //通过修改每个时间片的时间，窗口数量，阈值，来进行测试。
    public static void main(String[] args) {
        //1秒一个时间片，窗口共5个
        SlidingWindowLimiter window = new SlidingWindowLimiter(100, 4, 8);
        for (int i = 0; i < 100; i++) {
            // 生成范围内的随机数
            Random random = new Random(); // 0,1,2,3,4
            System.out.println(window.addCount(random.nextInt(5)));

            window.print();
            System.out.println("--------------------------");
            try {
                Thread.sleep(102);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 初始化队列，由于此初始化会申请一些内容空间，为了节省空间，延迟初始化
     */
    private void reset() {
        beginTimestamp = System.currentTimeMillis();

        if (timeSlices != null) {
            return;
        }
        //窗口个数
        AtomicInteger[] localTimeSlices = new AtomicInteger[timeSliceSize];
        for (int i = 0; i < timeSliceSize; i++) {
            localTimeSlices[i] = new AtomicInteger(0);
        }
        timeSlices = localTimeSlices;
    }

    private void print() {
        for (AtomicInteger integer : timeSlices) {
            System.out.print(integer + "-");
        }
    }

    /**
     * 计算当前所在的时间片的位置
     */
    private int locationIndex() {
        long now = System.currentTimeMillis();
        //如果当前的key已经超出一整个时间片了，那么就直接初始化就行了，不用去计算了
        if (now - lastAddTimestamp > timeMillisPerSlice * windowSize) {
            reset();
        }

        return (int) (((now - beginTimestamp) / timeMillisPerSlice) % timeSliceSize);
    }

    /**
     * 增加count个数量
     */
    public boolean addCount(int count) {
        //当前自己所在的位置，是哪个小时间窗
        int index = locationIndex();
        System.out.println("index:" + index);
        //然后清空自己前面windowSize到2*windowSize之间的数据格的数据
        //譬如1秒分4个窗口，那么数组共计8个窗口
        //当前index为5时，就清空6、7、8、1。然后把2、3、4、5的加起来就是该窗口内的总和
        clearFromIndex(index);

        int sum = 0;
        // 在当前时间片里继续+1
        sum += timeSlices[index].addAndGet(count);
        //加上前面几个时间片
        for (int i = 1; i < windowSize; i++) {
            sum += timeSlices[(index - i + timeSliceSize) % timeSliceSize].get();
        }
        System.out.println(sum + "---" + threshold);

        lastAddTimestamp = System.currentTimeMillis();

        return sum >= threshold;
    }

    private void clearFromIndex(int index) {
        for (int i = 1; i <= windowSize; i++) {
            int j = index + i;
            if (j >= windowSize * 2) {
                j -= windowSize * 2;
            }
            timeSlices[j].set(0);
        }
    }
}

3、漏桶

漏桶算法多使用队列实现，请求会缓存到队列中，服务提供方则按照固定的速率处理请求，过多的请求放在队列中或者直接拒绝。漏桶算法实现了削峰填谷，缺点是面对突发流量时即使服务器没有任何负载也必须等待一段时间请求才能被执行。

public class LeakyBucketLimiter {

    // 时间刻度
    private static long timeStamp = System.currentTimeMillis();
    // 桶大小
    private static int bucketSize = 10;
    // 每ms流出的请求
    private static int rate = 1;
    // 当前的水量
    private static long count = 0;

    public static boolean grant() {
        long now = System.currentTimeMillis();
        // 计算出水的数量
        long out = (now - timeStamp) * rate;
        // 先执行漏水，计算剩余水量
        count = Math.max(0, count - out);
        timeStamp = now;
        if ((count + 1) < bucketSize) {
            // 先执行漏水，计算剩余水量
            count++;
            return true;
        } else {
            // 水满，拒绝加水
            return false;
        }
    }

    public static void main(String[] args) {
        for (int i = 0; i < 100; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    if (grant()) {
                        System.out.println("执行业务逻辑");
                    } else {
                        System.out.println("限流");
                    }
                }
            }).start();
        }
    }

}

4、令牌桶

令牌以固定的速率生成并放入到令牌桶中，如果令牌桶满则丢弃，当请求到达时，先从令牌桶中取令牌，取到则可以被执行，如果令牌桶空了则丢弃。令牌桶算法即能够将请求均匀的分发到时间区间内，同时也能满足服务器能够承受一定程度上的突发流量限流，是目前一种广为使用的限流算法。guava已有现成的实现

public class GuavaRateLimiter {

    public static ConcurrentHashMap<String, RateLimiter> resourceRateLimiter = new ConcurrentHashMap<>();

    static {
        createrResourceLimiter("order", 5000);
    }

    private static void createrResourceLimiter(String resource, int qps) {
        if (resourceRateLimiter.contains(resource)) {
            resourceRateLimiter.get(resource).setRate(qps);
        } else {
            //permitsPerSecond表示每秒钟新增的令牌数，warmupPeriod表示从冷启动速率过渡到平均速率所需要的时间间隔
            RateLimiter rateLimiter = RateLimiter.create(qps, 2l, TimeUnit.MILLISECONDS);
            resourceRateLimiter.putIfAbsent(resource, rateLimiter);
        }
    }

    public static void main(String[] args) {
        for (int i = 0; i < 20; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    if (resourceRateLimiter.get("order").tryAcquire()) {
                        System.out.println("执行业务逻辑");
                    } else {
                        System.out.println("限流");
                    }
                }
            }).start();
        }
    }
}

隔离技术

1、线程池隔离

通过线程池来控制实际工作的线程数量，通过线程复用减少内存开销

public class ThreadPoolTest {

    public static void main(String[] args) {

        // 创建线程池，为了更好的明白运行流程，增加了一些额外的代码
//        BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(2);
        BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>();
//        BlockingQueue<Runnable> queue = new PriorityBlockingQueue<Runnable>();
//        BlockingQueue<Runnable> queue = new SynchronousQueue<Runnable>();

        // AbortPolicy/CallerRunsPolicy/DiscardOldestPolicy/DiscardPolicy
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2, 4, 5, TimeUnit.SECONDS,
                queue, new ThreadPoolExecutor.AbortPolicy());

        // 向线程池里面扔任务
        for (int i = 0; i < 10; i++) {
            System.out.println("当前线程池大小[" + threadPool.getPoolSize() + "],当前队列大小[" + queue.size() + "]");

            threadPool.execute(new MyThread ("Thread" + i));
        }
        // 关闭线程池
        threadPool.shutdown();
    }

    static class MyThread implements Runnable {
        private String name;

        public AllenThread(String name) {
            this.name = name;
        }

        @Override
        public void run() {
            // 做点事情
            try {
                Thread.sleep(1000);

                System.out.println(name + " finished job!");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

2、信号量隔离

Semaphore是一个并发工具类，通过内部虚拟了一组许可来控制可同时并发执行的线程数，每次线程执行操作先需先获取许可，执行完释放许可，若无可用的许可则线程阻塞等待。

public class SemaphoreTest {
    public static void main(String[] args) {
        Runnable customer = new Runnable() {
            final Semaphore availableWindow = new Semaphore(5, true);
            int count = 1;

            @Override
            public void run() {
                int time = (int) (Math.random() * 10 + 3);
                int num = count++;
                try {
                    availableWindow.acquire();
                    System.out.println("正在为第【" + num + "】个客户办理业务，需要时间：" + time + "s！");
                    Thread.sleep(time * 1000);
                    if (availableWindow.hasQueuedThreads()) {
                        System.out.println("第【" + num + "】个客户已办理完业务，有请下一位！");
                    } else {
                        System.out.println("第【" + num + "】个客户已办理完业务，没有客户了，休息中！");
                    }
                    availableWindow.release();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };

        for (int i = 1; i < 10; i++) {
            new Thread(customer).start();
        }
    }
}

手动实现一个限流框架

自定义注解

@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.METHOD})
@Documented
public @interface MyRateLimiter {
    //以固定数值往令牌桶添加令牌
    double permitsPerSecond();

    //获取令牌最大等待时间
    long timeout();

    // 单位(例:分钟/秒/毫秒) 默认:毫秒
    TimeUnit timeunit() default TimeUnit.MILLISECONDS;

    // 无法获取令牌返回提示信息 默认值可以自行修改
    String msg() default "您的操作太频繁了，请稍后再试";
}

切面类

@Aspect
@Component
public class RateLimiterAspect {
    private Logger logger = LoggerFactory.getLogger(RateLimiterAspect.class);

    /**
     * 使用url做为key,存放令牌桶 防止每次重新创建令牌桶
     */
    private Map<String, RateLimiter> limitMap = Maps.newConcurrentMap();

    @Pointcut("@annotation(com.example.limiter.aop.MyRateLimiter)")
    public void allenRateLimiter() {
    }

    @Around("allenRateLimiter()")
    public Object around(ProceedingJoinPoint joinPoint) throws Throwable {
        // 获取request,response
        HttpServletRequest request = ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getRequest();
        HttpServletResponse response = ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getResponse();
        // 或者url(存在map集合的key)
        String url = request.getRequestURI();
        // 获取自定义注解
        MyRateLimiter rateLimiter = getAllenRateLimiter(joinPoint);
        if (rateLimiter != null) {
            RateLimiter limiter = null;
            // 判断map集合中是否有创建有创建好的令牌桶
            if (!limitMap.containsKey(url)) {
                // 创建令牌桶
                limiter = RateLimiter.create(rateLimiter.permitsPerSecond());
                limitMap.put(url, limiter);
                logger.info("<<=================  请求{},创建令牌桶,容量{} 成功!!!", url, rateLimiter.permitsPerSecond());
            }
            limiter = limitMap.get(url);
            // 获取令牌
            boolean acquire = limiter.tryAcquire(rateLimiter.timeout(), rateLimiter.timeunit());

            if (!acquire) {
                responseResult(response, 500, rateLimiter.msg());
                return null;
            }
        }
        return joinPoint.proceed();
    }

    /**
     * 获取注解对象
     *
     * @param joinPoint 对象
     * @return ten LogAnnotation
     */
    private MyRateLimiter getAllenRateLimiter(final JoinPoint joinPoint) {
        Method[] methods = joinPoint.getTarget().getClass().getDeclaredMethods();
        String name = joinPoint.getSignature().getName();
        if (!StringUtils.isEmpty(name)) {
            for (Method method : methods) {
                MyRateLimiter annotation = method.getAnnotation(MyRateLimiter.class);
                if (!Objects.isNull(annotation) && name.equals(method.getName())) {
                    return annotation;
                }
            }
        }
        return null;
    }

    /**
     * 自定义响应结果
     *
     * @param response 响应
     * @param code     响应码
     * @param message  响应信息
     */
    private void responseResult(HttpServletResponse response, Integer code, String message) {
        response.resetBuffer();
        response.setHeader("Access-Control-Allow-Origin", "*");
        response.setHeader("Access-Control-Allow-Credentials", "true");
        response.setContentType("application/json");
        response.setCharacterEncoding("UTF-8");
        PrintWriter writer = null;
        try {
            writer = response.getWriter();
            writer.println("{"code":" + code + " ,"message" :"" + message + ""}");
            response.flushBuffer();
        } catch (IOException e) {
            logger.error(" 输入响应出错 e = {}", e.getMessage(), e);
        } finally {
            if (writer != null) {
                writer.flush();
                writer.close();
            }
        }
    }
}

测试类

@RestController
@RequestMapping("limiter")
public class MyLimiterController {

    @GetMapping("index")
    @MyRateLimiter(permitsPerSecond=2,timeout=1000)
    public String indexLimiter() {
        return "success";
    }

}

分布式限流实现

时间窗口方式

JedisLuaTimeWindowLimiter

public class JedisLuaTimeWindowLimiter {

    private String luaScript;

    private String key;

    private String limit;

    private String expire;

    public JedisLuaTimeWindowLimiter(String key, String limit, String expire, String scriptFile) {
        super();
        this.key = key;
        this.limit = limit;
        this.expire = expire;
        try {
            luaScript = Files.asCharSource(new ClassPathResource(scriptFile).getFile(), Charset.defaultCharset())
                    .read();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public boolean acqure() {
        Jedis jedis = new Jedis("localhost", 6379);
        return (Long) jedis.eval(luaScript, 1, key, limit, expire) == 1L;
    }
}

timeWindowLimit.lua

local key = KEYS[1] 			--限流KEY（一秒一个）
local limit = tonumber(ARGV[1]) --限流大小
local exprie = ARGV[2] 			--过期时间
-- 获取当前计数值
local current = tonumber(redis.call('get', key) or "0")
if current + 1 > limit then --如果超出限流大小
    return 0
else
	current = tonumber(redis.call("INCRBY", key, "1")) --请求数+1
	if current == 1 then  --第一次访问需要设置过期时间
    	redis.call("expire", key,exprie) --设置过期时间
    end
end
return 1  --返回1代表不限流

令牌桶实现

获取令牌实现
JedisRateLimiter

public class JedisRateLimiter {
	private String luaScript;

	private String key;

	public JedisRateLimiter(String scriptFile, String key) {
		super();
		this.key = key;
		try {
			luaScript = Files.asCharSource(new ClassPathResource(scriptFile).getFile(), Charset.defaultCharset())
					.read();
		} catch (IOException e) {
			e.printStackTrace();
		}

	}

	public boolean acquire() {
		try (Jedis jedis = new Jedis("localhost", 6379);) {
			return (Long) jedis.eval(luaScript, 1, key) == 1L;
		}
	}
}

rateLimit.lua

local key = KEYS[1] 			--限流KEY
-- 获取当前可用令牌数
local current = tonumber(redis.call('get', key) or "0")
if current <= 0 then --没有令牌了
    return 0
else
	redis.call("DECRBY", key, "1") --令牌数-1
end
return 1  --返回1代表不限流

添加令牌实现
JedisRateLimiterSeter

public class JedisRateLimiterSeter implements AutoCloseable {
	private String luaScript;

	private Timer timer;

	private final Jedis jedis = new Jedis("localhost", 6379);

	public JedisRateLimiterSeter(String scriptFile, String key, String limit) {
		super();

		try {
			luaScript = Files.asCharSource(new ClassPathResource(scriptFile).getFile(), Charset.defaultCharset())
					.read();
		} catch (IOException e) {
			e.printStackTrace();
		}

		timer = new Timer();

		// 放入令牌的时间间隔
		long period = 1000L / Long.valueOf(limit);

		// 通过定时器，定时放入令牌
		timer.scheduleAtFixedRate(new TimerTask() {
			@Override
			public void run() {

				System.out.println(
						System.currentTimeMillis() + " 放入令牌：" + ((Long) jedis.eval(luaScript, 1, key, limit) == 1L));
			}

		}, period, period);
	}

	@Override
	public void close() throws Exception {
		this.jedis.close();
		this.timer.cancel();
	}
	
}

rateLimitSet.lua

local key = KEYS[1] 			--限流KEY
local limit = tonumber(ARGV[1]) --容量
-- 获取当前令牌数
local current = tonumber(redis.call('get', key) or "0")
if current + 1 > limit then --如果超出容量
    return 0
else
	redis.call("INCRBY", key, "1") --令牌数+1
end
return 1  --返回1代表不限流