33.常见限流算法&使用RateLimiter实战限流

上一篇我们说到互联网高并发场景下的服务雪崩问题，同时也稍微介绍了下如何防止服务雪崩，即使用Hystrix做服务熔断、降级、隔离。今天我们来说下高并发场景下如何实现限流。
今天你会学到如下知识：
1. 常见的几种限流算法
1.计数器
2.滑动窗口
3.令牌桶
4.漏桶
2. 使用Guava的RateLimiter实现限流
3. 使用注解+AOP封装RateLimiter

0x01 常见几种限流算法分析

计数器限流

计数器算法是比较简单的一种限流算法，其本质思想就是使用一个计数器，如AtomicInteger。每次来一个请求，计数器+1.如果在规定时间内，计数器的大小超过了限制，就会拒绝请求。代码如下：

public class CounterLimit {
    /**
     * 使用原子类作为计数器
     */
    private static AtomicInteger counter = new AtomicInteger(0);
    /**
     * 限制50个
     */
    private static final int limit = 50;
    /**
     * 一分钟
     */
    private static final int time = 60*1000;
    /**
     * 重置时间
     */
    private long generateTime = System.currentTimeMillis();
    
    /**
     * 限流实现
     * @return true：已被限流   ;false:未被限流
     */
    private boolean Flowlimit() {
        //获取当前时间
        long currentTime = System.currentTimeMillis();
        //判断当前请求时间是否超过了1分钟。如果超过了，需要重置计数器
        if(currentTime>(generateTime + time)) {
            //重置
            counter.set(0);
            generateTime = System.currentTimeMillis();
        }
        //计数器+1
        counter.incrementAndGet();
        //判断是否被限流，即计数器是否超过了限定值
        return counter.get() > limit;
    }
    
    public static void main(String[] args) {
        CounterLimit counterLimit = new CounterLimit(); 
        //使用线程模拟请求
        for(int i=0;i<100;i++) {
            final int tmp = i;
            new Thread(()->{
                boolean flowlimit = counterLimit.Flowlimit();
                if(flowlimit) {
                    System.out.println(tmp+" :你已被限流~");
                }else {
                    System.out.println(tmp+" :可以正常访问~");
                }
            }) .start(); 
        }
    }
}

计数器实现限流的缺点：临界问题
- 临界问题：假设我们的限流策略是1分钟10个请求，那么如果用户在59秒的时候突然发送10个请求，然后再61秒的时候，再发送10个请求，这样在59秒~61秒之间，一共发了20个请求。但是我们规定是1分钟只能接受10个请求，而这里3秒却接受了20个请求，显然不符合我们设定的限流策略。如果利用这种漏洞，可以瞬间压垮我们的应用。那么怎么解决这个问题呢？下面滑动窗口限流算法给你个方案。

滑动窗口限流
1. 顾名思义，滑动窗口，首先是会滑动，并且他会把时间分成一个一个等份小格子(窗口)。如下图(图中的格子因为画的不好，可能并不是等份的)：
2. 那么他是怎么解决临界问题的呢？我们上面说过，他是会滑动的，可以理解有个大格子，把这些小格子包裹，我们这里每一个大格子包裹的就是1分钟。我这里是6个小格子代表1分钟。那么大格子就会包裹6个小格子。并且每10s往右滑动一次。如图：
3. 每当10s过去，又会往右滑动一个小格子。如下：
4. 那么有人要问了，他这样是怎么解决临界问题的呢？莫急莫慌，且听我道来。
  - 每个小格子里会有自己的独立的计数器。也就是说，你0:05秒的时候请求一次，那么在0:00~0:10这个小格子中的计数器就会+1.
  - 限流的时候，会把大格子包裹的小格子的计数器加起来，判断是否超过限制。
  - 继续我们上面说的临界问题，那么假设我们在59s和61s的时候，分别来了10个请求，总共20个请求，但是我们限流的是1分钟10个。会出现什么样的结果呢？
    - 59s的时候，来了10个请求，在0:50~1:00的小格子里计数器+10.也就是10.
    - 61s的时候，来了10个请求，这时候，就会把大格子包裹的小格子中的计数器相加，发现已经超过了10个请求。所以就触发限流。
5. 当1分钟内，格子分的越细，滑动的时候就会越平滑，限流统计越正确。
令牌桶算法
1. 令牌桶，就是一个有固定容量、存放令牌的桶。
2. 桶中的令牌是会按照一个固定速率生成。当桶满了的话，新生成的令牌会丢弃。
3. 每次请求都需要从这个桶里获取一个令牌，有令牌的请求才能被放行，被获取的令牌会从桶中移除。
4. 如果请求量>桶中令牌量，这时候就会触发限流机制，那么没有拿到令牌的请求，要么阻塞，要么就会被丢弃。
5. 大致原理如图：
6. 令牌桶限流是现在用的比较多的。它可以处理一些大量的突发请求，只要令牌桶中有令牌，就可以被执行。
漏桶算法
1. 漏桶算法，也是一个桶，和令牌桶不一样的是这个桶是漏的。(手动滑稽)
2. 漏桶算法的桶也是固定容量的，所有的请求都会先进入桶中，然后他会按照一个固定的速度漏水(放行请求)。
3. 如果桶满了，也就是溢出了，就会触发限流，丢弃请求。
4. 漏桶算法，其流入桶中的速率是不确定的。但是其放行的速度是固定的。所以可以达到限流的目的。
5. 大致原理图：
6. 漏桶算法与令牌桶算法的区别
  1. 漏桶算法是流入速度不固定，但是流出速度是固定的。而令牌桶，是平均流入速率，需要拿到令牌才能放行，可以允许一些突发请求，只需要令牌桶有令牌。

0x02 把码部分

通过Guava的RateLimiter实现令牌桶限流

建立Maven项目

引入Guava的pom依赖

<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
    <version>27.0.1-jre</version>
</dependency>

RateLimiter常用API
- rateLimiter.acquire()：获取令牌，如果桶中没有令牌，则会一直阻塞，直到拿到令牌。返回值为等待的秒数。
- rateLimiter.acquire(n):一次性获取n个令牌。如果桶中令牌不够，也会阻塞。直到拿到n个令牌。返回值为等待的秒数。
- rateLimiter.getRate():获取放令牌的速率
- rateLimiter.setRate():设置放令牌的速率
- rateLimiter.tryAcquire()：获取令牌，成功返回true。如果桶中没有令牌，不会阻塞，直接返回false。
- rateLimiter.tryAcquire(n)：一次性获取n个令牌。如果桶中没有足够的令牌，返回false。
- rateLimiter.tryAcquire(timeout, unit):获取令牌，如果桶中没有令牌，则等待指定timeout，如果还没有，则返回false

把码~

public class RateLimiterTest {
    /**
     * 初始化ReteLimiter。create(10)表示，每秒生成10个令牌到桶中。
     */
    private static RateLimiter rateLimiter = RateLimiter.create(10);
    
    public static void flowlimit() {
        //尝试获取令牌.如果没取到，等待500毫秒再次获取。
        boolean tryAcquire = rateLimiter.tryAcquire(500,TimeUnit.MILLISECONDS);
        if(tryAcquire) {
            System.out.println(Thread.currentThread().getName() + ",获取了令牌，执行服务...");
        }else {
            System.out.println(Thread.currentThread().getName() + ",令牌没有了，待会再来吧...");
        }
    }
    
    public static void main(String[] args) {
        for(int i=0;i<20;i++) {
            new Thread(()->flowlimit()).start();
        }
    }
}

注解+AOP封装RateLimiter

上面只是一个RateLimiter的一个简单例子。那么我们在web开发时，怎么使用RateLimiter限流呢？这里我使用Java的注解和Spring AOP封装下RateLimiter。后面只需要在需要限流的方法上面加上@RateLimiter注解即可对该请求进行限流。

引入Spring依赖，我这里使用SpringBoot快速构建一个web项目，pom依赖如下：

<dependencies>
	<dependency>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-web</artifactId>
	</dependency>
	
	<dependency>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-test</artifactId>
		<scope>test</scope>
	</dependency>
	
	<dependency>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-aop</artifactId>
	</dependency>
	
	<dependency>
	    <groupId>com.google.guava</groupId>
	    <artifactId>guava</artifactId>
	    <version>27.0.1-jre</version>
	</dependency>
</dependencies>

注解类：

@Retention(RUNTIME)
@Target(METHOD)
public @interface ExRateLimiter {
    /**
     * 每秒生成令牌的数量
     * @return
     */
    double rate() default 10;
    /**
     * 使用tryAcquire时如果获取不到锁，等待的毫秒数
     * @return
     */
    long timeout() default 500;
}

AOP类：

@Component
@Aspect
public class RateLimitAop {
    /**
     * 存放url映射和RateLimiter的缓存。
     */
    private static final ConcurrentHashMap<String,RateLimiter> rateLimiterMap = new ConcurrentHashMap<>();
    
    /**
     * AOP扫包，扫描top.zyling.api下面的所有类
     */
    @Pointcut(value="execution( * top.zyzling.api..*.*(**))")
    public void point() {
        
    }
    
    /**
     * 使用环绕通知。可以在没拿到令牌时，使用proceedingJoinPoint中断方法的运行，即不调用目标方法
     * @param proceedingJoinPoint
     * @throws Throwable 
     */
    @Around(value = "point()")
    public  Object around(ProceedingJoinPoint proceedingJoinPoint) throws Throwable{
        //获取方法上是否标有@RateLimiter注解
        MethodSignature methodSignature = (MethodSignature) proceedingJoinPoint.getSignature();
        Method method = methodSignature.getMethod();
        ExRateLimiter exRateLimiter = method.getDeclaredAnnotation(ExRateLimiter.class);
        
        if(exRateLimiter==null) {
            //没有标此注解，直接放行
            return proceedingJoinPoint.proceed();
        }
        
        //到这里说明方法上已经标有@RateLimiter注解。获取配置的属性
        double rate = exRateLimiter.rate();
        long timeout = exRateLimiter.timeout();
        RateLimiter rateLimiter = getRateLimiter(rate);
        //获取令牌
        boolean tryAcquire = rateLimiter.tryAcquire(timeout, TimeUnit.MILLISECONDS);
        if(tryAcquire) {
            //获取到了令牌，放行
            return proceedingJoinPoint.proceed();
        }
        //获取不到令牌，则返回一个提示。不执行目标方法
        getResponse().getWriter().write("未获取到令牌，不能执行方法~");
        return null;
    }
    
    /**
     * 获取RateLimiter
     * @param rate
     * @param timeout
     * @return
     */
    private RateLimiter getRateLimiter(double rate) {
        //获取当前的Request对象
        HttpServletRequest request = getRequest();
        //获取请求的uri
        String uri = request.getRequestURI();
        //根据uri从缓存中获取对应的RateLimiter.如果获取不到，就新建个，put到缓存中
        return rateLimiterMap.computeIfAbsent(uri,k->RateLimiter.create(rate));
    }
    
    /**
     * 获取request
     * @return
     */
    private HttpServletRequest getRequest() {
        ServletRequestAttributes requestAttributes = (ServletRequestAttributes)RequestContextHolder.getRequestAttributes();
        return requestAttributes.getRequest();
    }
    
    /**
     * 获取response
     * @return
     */
    private HttpServletResponse getResponse() {
        ServletRequestAttributes requestAttributes = (ServletRequestAttributes)RequestContextHolder.getRequestAttributes();
        HttpServletResponse response = requestAttributes.getResponse();
        response.setContentType("text/html;charset=utf-8");
        return response;
    }
}

测试API类：

@RestController
@RequestMapping("/user")
public class UserController {
    @GetMapping("/{userId}")
    //rate=0.1表示1秒生成0.1个令牌。也就是需要10s才会生成1个令牌
    @ExRateLimiter(rate=0.1,timeout=0)
    public String getUserInfo(@PathVariable("userId") String userId) {
        return "这是学号为【"+userId+"】的信息。";
    }
}

测试结果：
- 启动后，马上访问。结果如下：
- 随后继续访问，结果如下：

0x03 总结

到此，该篇已经完结了。谢谢观看。我们来总结下今天的内容；
- 在最开始的时候，我们介绍了几种常见的限流算法：
  - 计数器：原理很简单，来一个请求，计数器+1，到了限制就限流。但是有临界问题。
  - 滑动窗口：每隔一段时间滑动一次，可以解决临界问题。
  - 令牌桶：每个请求执行时，都需要获取令牌。有令牌才能执行。
  - 漏桶算法：按照固定速率流出，但是请求流入速率不固定，但是又因为桶的容量是固定的。所以桶满了(溢出)，溢出的请求将会被丢弃。
- 接下来我们使用Guava的RateLimiter来实现了令牌桶限流。
- 在最后，我们使用注解+AOP封装了下RateLimiter。
参考链接
- 三种常见的限流算法

高并发相关