Redis分布式限流组件设计与使用实例

本文主要讲解基于 自定义注解+aop+反射+redis+lua表达式 实现的限流设计方案。实现的限流设计与实际使用。

1.背景

在互联网开发中经常遇到需要限流的场景一般分为两种

• 业务场景需要（比如：5分钟内发送验证码不超过xxx次）;
• 对流量大的功能流量削峰;

一般我们衡量系统处理能力的指标是每秒的qps或者tps，假设系统每秒的流量阈值是2000，
理论上第2001个请求进来时，那么这个请求就需要被限流。

本文演示项目使用的是 springboot 项目，项目构建以及其他配置，这里不做演示。文末附限流demo源码

2.redis计数器限流设计

本文演示项目使用的是 springboot 项目，这里仅挑选了重点实现代码展示，
项目构建以及其他配置，这里不做演示，详细配置请参考源码demo工程。

2.1lua脚本

lua 是一种轻量小巧的脚本语言可以理解为就是一组命令。
使用redis的计数器达到限流的效果，表面上redis自带命令多个组合也可以支持了，那为什么还要用lua呢？
因为要保证原子性，这也是使用redis+lua表达式原因，一组命令要么全成功，要么全失败。
相比redis事务，lua脚本的优点：

• 减少网络开销：多个请求通过脚本一次发送，减少网络延迟
• 原子操作：将脚本作为一个整体执行，中间不会插入其他命令，无需使用事务
• 复用：客户端发送的脚本永久存在redis中，其他客户端可以复用脚本
• 可嵌入性：可嵌入java，c#等多种编程语言，支持不同操作系统跨平台交互

实现限流lua脚本示例

# 定义计数变量
local count
# 获取调用脚本时传入的第一个key值（用作限流的 key）
count = redis.call('get',keys[1])
# 限流最大值比较，若超过最大值，则直接返回
if count and tonumber(count) > tonumber(argv[1]) then
return count;
end
# incr 命令 执行计算器累加
count = redis.call('incr',keys[1])
# 从第一次调用开始限流,并设置失效时间
if tonumber(count) == 1 then
redis.call('expire',keys[1],argv[2])
end
return count;

参数说明

• keys[1] - redis的key
• argv[1] - 限流次数
• argv[2] - 失效时间

2.2自定义注解

支持范围：任意接口

/**
 * 描述: 限流注解
 *
 * @author 程序员小强
 **/
@target({elementtype.type, elementtype.method})
@retention(retentionpolicy.runtime)
public @interface ratelimit {

    /**
     * 限流唯一标示 key
     * 若同时使用 keyfiled 则当前 key作为前缀
     */
    string key();

    /**
     * 限流时间-单位:秒数
     * 默认 60s
     */
    int time() default 60;

    /**
     * 限流次数
     * 失效时间段内最大放行次数
     */
    int count();

    /**
     * 可作为限流key-参数类中属性名，动态值
     * 示例：phone、userid 等
     */
    string keyfield() default "";

    /**
     * 超过最大访问次数后的，提示内容
     */
    string msg() default "over the max request times please try again";

}

属性介绍

• key - 必填，限流key唯一标识，redis存储key
• time -过期时间，单位 秒，默认60s
• count - 必填，失效时间段内最大放行次数
• keyfield - 动态限流key,比如参数是一个自定义的类，里面有属性userid 等。可以使用keyfield=“userid”,
  

这样生成的key为参数中userid的值。一般与key属性组合使用。不支持java基本类型参数，
仅支持参数是一个对象的接口。

msg - 超过限流的提示内容

示例:

@ratelimit(key = "limit-phone-key", time = 300, count = 10, keyfield = "phone", msg = "5分钟内，验证码最多发送10次")

含义 - 5分钟内根据手机号限流10次
rediskey- limit-phone-key:后面拼接的是参数中phone的值。

2.3限流组件

这里用的是jedis客户端，配置就不列在这里的，详见源码，文末附源码地址

/**
 * redis限流组件
 *
 * @author 程序员小强
 */
@component
public class redisratelimitcomponent {
    private static final logger logger = loggerfactory.getlogger(redisratelimitcomponent.class);

    private jedispool jedispool;

    @autowired
    public redisratelimitcomponent(jedispool jedispool) {
        this.jedispool = jedispool;
    }

    /**
     * 限流方法
     * 1.执行 lua 表达式
     * 2.通过 lua 表达式实现-限流计数器
     *
     * @param rediskey
     * @param time           超时时间-秒数
     * @param ratelimitcount 限流次数
     */
    public long ratelimit(string rediskey, integer time, integer ratelimitcount) {
        jedis jedis = null;
        try {
            jedis = jedispool.getresource();
            object obj = jedis.evalsha(jedis.scriptload(this.buildluascript()), collections.singletonlist(rediskey),
                    arrays.aslist(string.valueof(ratelimitcount), string.valueof(time)));
            return long.valueof(obj.tostring());
        } catch (jedisexception ex) {
            logger.error("[ executelua ] >> messages:{}", ex.getmessage(), ex);
            throw new ratelimitexception("[ redisratelimitcomponent ] >> jedis run lua script exception" + ex.getmessage());
        } finally {
            if (jedis != null) {
                if (jedis.isconnected()) {
                    jedis.close();
                }
            }
        }
    }

    /**
     * 构建lua 表达式
     * keys[1] -- 参数key
     * argv[1]-- 失效时间段内最大放行次数
     * argv[2]-- 失效时间|秒
     */
    private string buildluascript() {
        stringbuilder luabuilder = new stringbuilder();
        //定义变量
        luabuilder.append("local count");
        //获取调用脚本时传入的第一个key值（用作限流的 key）
        luabuilder.append("\ncount = redis.call('get',keys[1])");
        // 获取调用脚本时传入的第一个参数值（限流大小）-- 调用不超过最大值，则直接返回
        luabuilder.append("\nif count and tonumber(count) > tonumber(argv[1]) then");
        luabuilder.append("\nreturn count;");
        luabuilder.append("\nend");
        //执行计算器自增
        luabuilder.append("\ncount = redis.call('incr',keys[1])");
        //从第一次调用开始限流
        luabuilder.append("\nif tonumber(count) == 1 then");
        //设置过期时间
        luabuilder.append("\nredis.call('expire',keys[1],argv[2])");
        luabuilder.append("\nend");
        luabuilder.append("\nreturn count;");
        return luabuilder.tostring();
    }
}

2.4限流切面实现

/**
 * 描述:限流切面实现
 *
 * @author 程序员小强
 **/
@aspect
@configuration
public class ratelimitaspect {
    private static final logger logger = loggerfactory.getlogger(ratelimitaspect.class);

    private redisratelimitcomponent redisratelimitcomponent;

    @autowired
    public ratelimitaspect(redisratelimitcomponent redisratelimitcomponent) {
        this.redisratelimitcomponent = redisratelimitcomponent;
    }

    /**
     * 匹配所有使用以下注解的方法
     *
     * @see ratelimit
     */
    @pointcut("@annotation(com.example.ratelimit.annotation.ratelimit)")
    public void pointcut() {
    }

    @around("pointcut()&&@annotation(ratelimit)")
    public object logaround(proceedingjoinpoint joinpoint, ratelimit ratelimit) throws throwable {
        methodsignature signature = (methodsignature) joinpoint.getsignature();
        string methodname = signature.getmethod().getname();

        //组装限流key
        string ratelimitkey = this.getratelimitkey(joinpoint, ratelimit);

        //限流组件-通过计数方式限流
        long count = redisratelimitcomponent.ratelimit(ratelimitkey, ratelimit.time(), ratelimit.count());
        logger.debug("[ ratelimit ] method={},ratelimitkey={},count={}", methodname, ratelimitkey, count);

        if (null != count && count.intvalue() <= ratelimit.count()) {
            //未超过限流次数-执行业务方法
            return joinpoint.proceed();
        } else {
            //超过限流次数
            logger.info("[ ratelimit ] >> over the max request times method={},ratelimitkey={},currentcount={},ratelimitcount={}",
                    methodname, ratelimitkey, count, ratelimit.count());
            throw new ratelimitexception(ratelimit.msg());
        }
    }

    /**
     * 获取限流key
     * 默认取 ratelimit > key 属性值 
     * 若设置了 keyfield 则从参数中获取该字段的值拼接到key中
     * 示例：user_phone_login_max_times:13235777777
     *
     * @param joinpoint
     * @param ratelimit
     */
    private string getratelimitkey(proceedingjoinpoint joinpoint, ratelimit ratelimit) {
        string fieldname = ratelimit.keyfield();
        if ("".equals(fieldname)) {
            return ratelimit.key();
        }

        //处理自定义-参数名-动态属性key
        stringbuilder ratelimitkeybuilder = new stringbuilder(ratelimit.key());
        for (object obj : joinpoint.getargs()) {
            if (null == obj) {
                continue;
            }
            //过滤基本类型参数
            if (reflectionutil.isbasetype(obj.getclass())) {
                continue;
            }
            //属性值
            object fieldvalue = reflectionutil.getfieldbyclazz(fieldname, obj);
            if (null != fieldvalue) {
                ratelimitkeybuilder.append(":").append(fieldvalue.tostring());
                break;
            }
        }
        return ratelimitkeybuilder.tostring();
    }
}

由于演示项目中做了统一异常处理
在限流切面这里未做异常捕获，若超过最大限流次数会抛出自定义限流异常。可以根据业务自行处理。

/**
 * 反射工具
 *
 * @author 程序员小强
 */
public class reflectionutil {

    private static final logger logger = loggerfactory.getlogger(reflectionutil.class);

    /**
     * 根据属性名获取属性元素，
     * 包括各种安全范围和所有父类
     *
     * @param fieldname
     * @param object
     * @return
     */
    public static object getfieldbyclazz(string fieldname, object object) {
        field field = null;
        class<?> clazz = object.getclass();
        try {
            for (; clazz != object.class; clazz = clazz.getsuperclass()) {
                try {
                    //子类中查询不到属性-继续向父类查
                    field = clazz.getdeclaredfield(fieldname);
                } catch (nosuchfieldexception ignored) {
                }
            }
            if (null == field) {
                return null;
            }
            field.setaccessible(true);
            return field.get(object);
        } catch (exception e) {
            //通过反射获取 属性值失败
            logger.error("[ reflectionutil ] >> [getfieldbyclazz] fieldname:{} ", fieldname, e);
        }
        return null;
    }

    /**
     * 判断对象属性是否是基本数据类型,包括是否包括string | bigdecimal
     *
     * @param clazz
     * @return
     */
    public static boolean isbasetype(class clazz) {
        if (null == clazz) {
            return false;
        }
        //基本类型
        if (clazz.isprimitive()) {
            return true;
        }
        //string
        if (clazz.equals(string.class)) {
            return true;
        }
        //integer
        if (clazz.equals(integer.class)) {
            return true;
        }
        //boolean
        if (clazz.equals(boolean.class)) {
            return true;
        }
        //bigdecimal
        if (clazz.equals(bigdecimal.class)) {
            return true;
        }
        //byte
        if (clazz.equals(byte.class)) {
            return true;
        }
        //long
        if (clazz.equals(long.class)) {
            return true;
        }
        //double
        if (clazz.equals(double.class)) {
            return true;
        }
        //float
        if (clazz.equals(float.class)) {
            return true;
        }
        //character
        if (clazz.equals(character.class)) {
            return true;
        }
        //short
        return clazz.equals(short.class);
    }
}

3.测试一下

基本属性已经配置好了，写个接口测试一下。

3.1方法限流示例

  /**
   * 计数器
   * 演示 demo 为了方便计数
   */
  private static final atomicinteger counter = new atomicinteger();    

  /**
   * 普通限流
   * <p>
   * 30 秒中，可以访问10次
   */
  @requestmapping("/limittest")
  @ratelimit(key = "limit-test-key", time = 30, count = 10)
  public response limittest() {
      map<string, object> datamap = new hashmap<>();
      datamap.put("date", dateformatutils.format(new date(), "yyyy-mm-dd hh:mm:ss.sss"));
      datamap.put("times", counter.incrementandget());
      return response.success(datamap);
  }

3.2动态入参限流示例

3.2.1场景一：5分钟内，方法最多访问10次，根据入参手机号限流

入参类

public class userphonecaptcharateparam implements serializable {

    private static final long serialversionuid = -1l;

    private string phone;
    //省略 get/set
}

  private static final map<string, atomicinteger> count_phone_map = new hashmap<>();


  /**
   * 根据手机号限流-限制验证码发送次数
   * <p>
   * 示例：5分钟内，验证码最多发送10次
   */
  @requestmapping("/limitbyphone")
  @ratelimit(key = "limit-phone-key", time = 300, count = 10, keyfield = "phone", msg = "5分钟内，验证码最多发送10次")
  public response limitbyphone(userphonecaptcharateparam param) {
      map<string, object> datamap = new hashmap<>();
      datamap.put("date", dateformatutils.format(new date(), "yyyy-mm-dd hh:mm:ss.sss"));
      if (count_phone_map.containskey(param.getphone())) {
          count_phone_map.get(param.getphone()).incrementandget();
      } else {
          count_phone_map.put(param.getphone(), new atomicinteger(1));
      }
      datamap.put("times", count_phone_map.get(param.getphone()).intvalue());
      datamap.put("reqparam", param);
      return response.success(datamap);
  }

3.2.2场景二：根据订单id限流

入参类

@data
public class orderrateparam implements serializable {

    private static final long serialversionuid = -1l;

    private string orderid;
    //省略 get\set
}

  private static final map<string, atomicinteger> count_order_map = new hashmap<>();

  /**
   * 根据订单id限流示例
   * <p>
   * 300 秒中，可以访问10次
   */
  @requestmapping("/limitbyorderid")
  @ratelimit(key = "limit-order-key", time = 300, count = 10, keyfield = "orderid", msg = "订单飞走了，请稍后再试!")
  public response limitbyorderid(orderrateparam param) {
      map<string, object> datamap = new hashmap<>();
      datamap.put("date", dateformatutils.format(new date(), "yyyy-mm-dd hh:mm:ss.sss"));
      if (count_order_map.containskey(param.getorderid())) {
          count_order_map.get(param.getorderid()).incrementandget();
      } else {
          count_order_map.put(param.getorderid(), new atomicinteger(1));
      }
      datamap.put("times", count_order_map.get(param.getorderid()).intvalue());
      datamap.put("reqparam", param);
      return response.success(datamap);
  }

4.其它扩展

根据ip限流

在key中拼接ip即可；

5.源码地址

传送门

到此这篇关于redis分布式限流组件设计与使用实例的文章就介绍到这了,更多相关redis分布式限流内容请搜索以前的文章或继续浏览下面的相关文章希望大家以后多多支持！