一、常用的限流算法

1.计数器方式（传统计数器缺点：临界问题可能违背定义固定速率原则）

2.令牌桶方式

令牌桶算法的原理是系统会以一个恒定的速度往桶里放入令牌，而如果请求需要被处理，则需要先从桶里获取一个令牌，当桶里没有令牌可取时，则拒绝服务。从原理上看，令牌桶算法和漏桶算法是相反的，一个“进水”，一个是“漏水”。

RateLimiter是guava提供的基于令牌桶算法的实现类，可以非常简单的完成限流特技，并且根据系统的实际情况来调整生成token的速率。 RateLimiter 是单机（单进程）的限流，是JVM级别的的限流，所有的令牌生成都是在内存中，在分布式环境下不能直接这么用。

3、漏桶算法

如上图所示，我们假设系统是一个漏桶，当请求到达时，就是往漏桶里“加水”，而当请求被处理掉，就是水从漏桶的底部漏出。水漏出的速度是固定的，当“加水”太快，桶就会溢出，也就是“拒绝请求”。从而使得桶里的水的体积不可能超出桶的容量。

令牌桶算法与漏桶算法的区别：

令牌桶里面装载的是令牌，然后让令牌去关联到数据发送，常规漏桶里面装载的是数据，令牌桶允许用户的正常的持续突发量（Bc），就是一次就将桶里的令牌全部用尽的方式来支持续突发，而常规的漏桶则不允许用户任何突发行。

二、限流实现

基于 redis 的分布式限流

单机版中我们了解到 AtomicInteger、RateLimiter、Semaphore 这几种解决方案，但它们也仅仅是单机的解决手段，在集群环境下就透心凉了，后面又讲述了 Nginx 的限流手段，可它又属于网关层面的策略之一，并不能解决所有问题。例如供短信接口，你无法保证消费方是否会做好限流控制，所以自己在应用层实现限流还是很有必要的。

导入依赖

在 pom.xml 中添加上 starter-web、starter-aop、starter-data-redis 的依赖即可，习惯了使用 commons-lang3 和 guava 中的一些工具包…

<dependencies>
    <!-- 默认就内嵌了Tomcat 容器，如需要更换容器也极其简单-->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-aop</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>
    <dependency>
        <groupId>com.google.guava</groupId>
        <artifactId>guava</artifactId>
        <version>21.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.commons</groupId>
        <artifactId>commons-lang3</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
    </dependency>
</dependencies>

属性配置

在 application.yml 资源文件中添加 redis 相关的配置项

spring:
  redis:
    timeout: 5000
    jedis:
      pool:
        max-active: 20
        max-idle: 8
        min-idle: 0
        max-wait: 0
    port: 6379 
    host: 192.168.89.70 
    database: 4

Limit 注解

创建一个 Limit 注解，不多说注释都给各位写齐全了….

import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Inherited;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

// 限流
@Target({ElementType.METHOD, ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
public @interface Limit {

    /**
     * 资源的名称
     * @return
     */
    String name() default "";
    
    /**
     * 资源的key
     *
     * @return
     */
    String key() default "";

    /**
     * Key的prefix
     *
     * @return
     */
    String prefix() default "";
    
    /**
     * 给定的时间段
     * 单位秒
     *
     * @return
     */
    int period();

    /**
     * 最多的访问限制次数
     *
     * @return
     */
    int count();
    
    /**
     * 类型
     *
     * @return
     */
    LimitType limitType() default LimitType.CUSTOMER;
}

// 限制的类型
public enum LimitType {

    /**
     * 自定义key
     */
    CUSTOMER,
    /**
     * 根据请求者IP
     */
    IP;
    
}

RedisTemplate

默认情况下 spring-boot-data-redis 为我们提供了StringRedisTemplate 但是满足不了其它类型的转换，所以还是得自己去定义其它类型的模板…

@Configuration
public class RedisConfig {

    @Resource
    RedisConnectionFactory factory;

    @Bean
    public RedisTemplate<String, Object> redisTemplate() {
        //设置序列化
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(om);
        // 配置redisTemplate
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(factory);
        RedisSerializer stringSerializer = new StringRedisSerializer();
        // key序列化
        redisTemplate.setKeySerializer(stringSerializer);
        // value序列化
        //redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);
        // Hash key序列化
        redisTemplate.setHashKeySerializer(stringSerializer);
        // Hash value序列化
        //redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }

    @Bean(name = "redisTemp")
    public RedisTemplate<String, String> redisTemp() {
        //设置序列化
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(om);
        // 配置redisTemplate
        RedisTemplate<String, String> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(factory);
        RedisSerializer stringSerializer = new StringRedisSerializer();
        // key序列化
        redisTemplate.setKeySerializer(stringSerializer);
        // value序列化
        redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);
        // Hash key序列化
        redisTemplate.setHashKeySerializer(stringSerializer);
        // Hash value序列化
        redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }

    @Bean
    public HashOperations<String, String, Object> hashOperations(RedisTemplate<String, Object> redisTemplate) {
        return redisTemplate.opsForHash();
    }

    @Bean
    public ValueOperations<String, Object> valueOperations(RedisTemplate<String, Object> redisTemplate) {
        return redisTemplate.opsForValue();
    }

    @Bean
    public ListOperations<String, Object> listOperations(RedisTemplate<String, Object> redisTemplate) {
        return redisTemplate.opsForList();
    }

    @Bean
    public SetOperations<String, Object> setOperations(RedisTemplate<String, Object> redisTemplate) {
        return redisTemplate.opsForSet();
    }

    @Bean
    public ZSetOperations<String, Object> zSetOperations(RedisTemplate<String, Object> redisTemplate) {
        return redisTemplate.opsForZSet();
    }

Limit 拦截器（AOP）

熟悉 Redis 的朋友都知道它是线程安全的，我们利用它的特性可以实现分布式锁、分布式限流等组件，官方虽然没有提供相应的API，但却提供了支持 Lua 脚本的功能，我们可以通过编写 Lua 脚本实现自己的API，同时他是满足原子性的….
下面核心就是调用 execute 方法传入我们的 Lua 脚本内容，然后通过返回值判断是否超出我们预期的范围，超出则给出错误提示。

import com.google.common.collect.ImmutableList;
import com.lanswon.limit.annotation.Limit;
import com.lanswon.limit.annotation.LimitType;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.stereotype.Component;
import org.springframework.web.context.request.RequestContextHolder;
import org.springframework.web.context.request.ServletRequestAttributes;

import javax.annotation.Resource;
import javax.servlet.http.HttpServletRequest;
import java.lang.reflect.Method;

@Aspect
@Slf4j
@Component
public class LimitInterceptor {

    @Resource(name = "redisTemp")
    private RedisTemplate<String,String> redisTemplate;

    private final String REDIS_SCRIPT = buildLuaScript();

    @Around("execution(public * *(..)) && @annotation(com.lanswon.limit.annotation.Limit)")
    public Object interceptor(ProceedingJoinPoint pjp){

        MethodSignature signature = (MethodSignature) pjp.getSignature();
        Method method = signature.getMethod();
        Limit limitAnno = method.getAnnotation(Limit.class);
        LimitType limitType = limitAnno.limitType();
        String name = limitAnno.name();
        String key = null;
        int limitPeriod = limitAnno.period();
        int limitCount = limitAnno.count();
        switch (limitType) {
            case IP:
                key = getIpAddress();
                break;
            case CUSTOMER:
                key = limitAnno.key();
                break;
            default:
                break;
        }

        ImmutableList<String> keys = ImmutableList.of(StringUtils.join(limitAnno.prefix(), key));
        try {
            RedisScript<Number> redisScript = new DefaultRedisScript<Number>(REDIS_SCRIPT, Number.class);
            Number count = redisTemplate.execute(redisScript, keys, limitCount, limitPeriod);
            log.info("Access try count is {} for name={} and key = {}", count, name, key);
            if(count != null && count.intValue() <= limitCount) {
                return pjp.proceed();
            } else {
                throw new RuntimeException("当前服务繁忙,请稍后再试");
            }
        } catch (Throwable e) {
            if (e instanceof RuntimeException) {
                throw new RuntimeException(e.getLocalizedMessage());
            }
            throw new RuntimeException("server exception");
        }

    }



    /**
     * 限流 脚本
     *
     * @return lua脚本
     */
    private String buildLuaScript() {
        StringBuilder lua = new StringBuilder();
        lua.append("local c")
                .append("\nc = redis.call('get', KEYS[1])")
                // 调用不超过最大值，则直接返回
                .append("\nif c and tonumber(c) > tonumber(ARGV[1]) then")
                .append("\nreturn c;")
                .append("\nend")
                // 执行计算器自加
                .append("\nc = redis.call('incr', KEYS[1])")
                .append("\nif tonumber(c) == 1 then")
                // 从第一次调用开始限流，设置对应键值的过期
                .append("\nredis.call('expire', KEYS[1], ARGV[2])")
                .append("\nend")
                .append("\nreturn c;");
        return lua.toString();
    }

    private static final String UNKNOWN = "unknown";

    public String getIpAddress() {
        HttpServletRequest request = ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getRequest();
        String ip = request.getHeader("x-forwarded-for");
        if (ip == null || ip.length() == 0 || UNKNOWN.equalsIgnoreCase(ip)) {
            ip = request.getHeader("Proxy-Client-IP");
        }
        if (ip == null || ip.length() == 0 || UNKNOWN.equalsIgnoreCase(ip)) {
            ip = request.getHeader("WL-Proxy-Client-IP");
        }
        if (ip == null || ip.length() == 0 || UNKNOWN.equalsIgnoreCase(ip)) {
            ip = request.getRemoteAddr();
        }
        return ip;
    }
}

控制层

在接口上添加 @Limit() 注解，如下代码会在 Redis 中生成过期时间为 100s 的 key = test 的记录

    @GetMapping("/oauth/checkToken")
    @Limit(key = "test", period = 10, count = 10,name = "TOKEN效验")
    public Map<String ,Object> checkToken(HttpServletRequest request, HttpServletResponse response){
        String token = request.getParameter("token");
        String uri = "http://127.0.0.1:"+port+"/oauth/check_token?token="+token;
        log.debug("授权服务ID=[{}]",uri);
        Map<String,Object> result=new HashMap<>();
        try{
        result = restTemplate.getForObject(uri,Map.class);
        result.put("code",HttpStatus.OK);
        log.debug("result=[{}]",result.toString());
        }catch (Exception ex){
           result.put("code",HttpStatus.UNAUTHORIZED);
           result.put("message","token非法");
           return result;
        }
        log.debug("Token验证结果=[{}]",result.get("active"));
        if(!(Boolean) result.get("active")){
            result.put("code",HttpStatus.UNAUTHORIZED);
            result.put("message","token失效");
            //TODO token失效加入处理队列

        }
        return result;
    }


    @ExceptionHandler
    @ResponseBody
    public SimpleResponse handleException(Exception ex, HttpServletRequest request) throws UnknownHostException {
        log.error("exception=[{}]",ex.getClass());
        log.error("===系统出现异常[{}]===",ex.getMessage());
        ex.printStackTrace();
        SimpleResponse simpleResponse = SimpleResponse
                .builder()
                .status(Integer.valueOf(HttpStatus.INTERNAL_SERVER_ERROR.value()))
                .build();
        if(ex instanceof Exception){
            simpleResponse.setMsg("系统出现异常=["+ex.getMessage()+"]");
        }
        return simpleResponse;
    }

测试

完成准备事项后，启动启动类自行测试即可

注：上面的方式是使用计数器的限流方式，无法处理临界的时候，大量请求的的情况。要解决这个问题，可以使用redis中列表类型的键来记录最近N次访问的时间，一旦键中的元素超过N个，就判断时间最早的元素距现在的时间是否小于M秒。如果是则表示用户最近M秒的访问次数超过了N次；如果不是就将现在的时间加入列表中同时把最早的元素删除（可以通过脚本功能避免竞态条件）。由于需要记录每次访问的时间，所以当要限制“M时间最多访问N次”时，如果“N”的数值较大，此方法会占用较多的存储空间，实际使用时还需要开发者自己去权衡。

下面的解决思路的实现如下：

/**
     * 限流 脚本（处理临界时间大量请求的情况）
     *
     * @return lua脚本
     */
    private String buildLuaScript2() {
        StringBuilder lua = new StringBuilder();
        lua.append("local listLen, time")
           .append("\nlistLen = redis.call('LLEN', KEYS[1])")
           // 不超过最大值，则直接写入时间
           .append("\nif listLen and tonumber(listLen) < tonumber(ARGV[1]) then")
                .append("\nlocal a = redis.call('TIME');")
                .append("\nredis.call('LPUSH', KEYS[1], a[1]*1000000+a[2])")
           .append("\nelse")
               // 取出现存的最早的那个时间，和当前时间比较，看是小于时间间隔
               .append("\ntime = redis.call('LINDEX', KEYS[1], -1)")
               .append("\nlocal a = redis.call('TIME');")
               .append("\nif a[1]*1000000+a[2] - time < tonumber(ARGV[2])*1000000 then")
                   // 访问频率超过了限制，返回0表示失败
                   .append("\nreturn 0;")
               .append("\nelse")                   
                   .append("\nredis.call('LPUSH', KEYS[1], a[1]*1000000+a[2])")
                   .append("\nredis.call('LTRIM', KEYS[1], 0, tonumber(ARGV[1])-1)")
               .append("\nend")   
           .append("\nend")
           .append("\nreturn 1;");
        return lua.toString();
    }

调用的地方的也相应修改如下：

if(count != null && count.intValue() == 1) {
    return pjp.proceed();
} else {
    throw new RuntimeException("You have been dragged into the blacklist");
}

补充，最近执行 buildLuaScript2(）中的lua脚本，报错

Write commands not allowed after non deterministic commands.

这个错误的原因大家可以参见这篇文章，大致原因跟redis集群的重放和备份策略有关，相当于我调用TIME操作，会在主从各执行一次，得到的结果肯定会存在差异，这个差异就给最终逻辑正确性带来了不确定性。在redis 4.0之后引入了redis.replicate_commands()来放开限制。

于是，在 buildLuaScript2 的 lua 脚本最前面加上 “redis.replicate_commands();”，错误得以解决。

Q.E.D.

GuBlog

SpringBoot 2.x 分布式限流(不依赖Sentinel和Hystrix)

一、常用的限流算法

1.计数器方式（传统计数器缺点：临界问题可能违背定义固定速率原则）

2.令牌桶方式

3、漏桶算法

令牌桶算法与漏桶算法的区别：

二、限流实现

基于 redis 的分布式限流

导入依赖

属性配置

Limit 注解

RedisTemplate

Limit 拦截器（AOP）

控制层

测试

Rocket MQ集群搭建

fastDFS文件服务器安装

爱调味品的大哥

GuBlog

一、常用的限流算法

1.计数器方式（传统计数器缺点：临界问题 可能违背定义固定速率原则）

2.令牌桶方式

3、漏桶算法

令牌桶算法与漏桶算法的区别：

二、限流实现

基于 redis 的分布式限流

导入依赖

属性配置

Limit 注解

RedisTemplate

Limit 拦截器（AOP）

控制层

测试

Rocket MQ集群搭建

fastDFS文件服务器安装

爱调味品的大哥

GuBlog

1.计数器方式（传统计数器缺点：临界问题可能违背定义固定速率原则）