Redis分布式限流組件設計與使用實例

更新時間：2021年08月12日 10:05:06 作者：程序員小強

本文主要講解基于自定義注解+Aop+反射+Redis+Lua表達式實現(xiàn)的限流設計方案。實現(xiàn)的限流設計與實際使用。具有一定的參考價值，感興趣的小伙伴們可以參考一下

1.背景

在互聯(lián)網開發(fā)中經常遇到需要限流的場景一般分為兩種

業(yè)務場景需要（比如：5分鐘內發(fā)送驗證碼不超過xxx次）;
對流量大的功能流量削峰;

一般我們衡量系統(tǒng)處理能力的指標是每秒的QPS或者TPS，假設系統(tǒng)每秒的流量閾值是2000，
理論上第2001個請求進來時，那么這個請求就需要被限流。

本文演示項目使用的是 SpringBoot 項目，項目構建以及其他配置，這里不做演示。文末附限流Demo源碼

2.Redis計數(shù)器限流設計

本文演示項目使用的是 SpringBoot 項目，這里僅挑選了重點實現(xiàn)代碼展示，
項目構建以及其他配置，這里不做演示，詳細配置請參考源碼demo工程。

2.1Lua腳本

Lua 是一種輕量小巧的腳本語言可以理解為就是一組命令。
使用Redis的計數(shù)器達到限流的效果，表面上Redis自帶命令多個組合也可以支持了，那為什么還要用Lua呢？
因為要保證原子性，這也是使用redis+Lua表達式原因，一組命令要么全成功，要么全失敗。
相比Redis事務，Lua腳本的優(yōu)點：

減少網絡開銷：多個請求通過腳本一次發(fā)送，減少網絡延遲
原子操作：將腳本作為一個整體執(zhí)行，中間不會插入其他命令，無需使用事務
復用：客戶端發(fā)送的腳本永久存在redis中，其他客戶端可以復用腳本
可嵌入性：可嵌入JAVA，C#等多種編程語言，支持不同操作系統(tǒng)跨平臺交互

實現(xiàn)限流Lua腳本示例

# 定義計數(shù)變量
local count
# 獲取調用腳本時傳入的第一個key值（用作限流的 key）
count = redis.call('get',KEYS[1])
# 限流最大值比較，若超過最大值，則直接返回
if count and tonumber(count) > tonumber(ARGV[1]) then
return count;
end
# incr 命令 執(zhí)行計算器累加
count = redis.call('incr',KEYS[1])
# 從第一次調用開始限流,并設置失效時間
if tonumber(count) == 1 then
redis.call('expire',KEYS[1],ARGV[2])
end
return count;

參數(shù)說明

KEYS[1] - redis的Key
ARGV[1] - 限流次數(shù)
ARGV[2] - 失效時間

2.2自定義注解

支持范圍：任意接口

/**
 * 描述: 限流注解
 *
 * @author 程序員小強
 **/
@Target({ElementType.TYPE, ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
public @interface RateLimit {

    /**
     * 限流唯一標示 key
     * 若同時使用 keyFiled 則當前 key作為前綴
     */
    String key();

    /**
     * 限流時間-單位:秒數(shù)
     * 默認 60s
     */
    int time() default 60;

    /**
     * 限流次數(shù)
     * 失效時間段內最大放行次數(shù)
     */
    int count();

    /**
     * 可作為限流key-參數(shù)類中屬性名，動態(tài)值
     * 示例：phone、userId 等
     */
    String keyField() default "";

    /**
     * 超過最大訪問次數(shù)后的，提示內容
     */
    String msg() default "over the max request times please try again";

}

屬性介紹

key - 必填，限流key唯一標識，redis存儲key
time -過期時間，單位秒，默認60s
count - 必填，失效時間段內最大放行次數(shù)
keyField - 動態(tài)限流key,比如參數(shù)是一個自定義的類，里面有屬性userId 等?？梢允褂胟eyField=“userId”,

這樣生成的key為參數(shù)中userId的值。一般與key屬性組合使用。不支持java基本類型參數(shù)，
僅支持參數(shù)是一個對象的接口。

msg - 超過限流的提示內容

示例:

@RateLimit(key = "limit-phone-key", time = 300, count = 10, keyField = "phone", msg = "5分鐘內，驗證碼最多發(fā)送10次")

含義 - 5分鐘內根據(jù)手機號限流10次
RedisKey- limit-phone-key:后面拼接的是參數(shù)中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.執(zhí)行 lua 表達式
     * 2.通過 lua 表達式實現(xiàn)-限流計數(shù)器
     *
     * @param redisKey
     * @param time           超時時間-秒數(shù)
     * @param rateLimitCount 限流次數(shù)
     */
    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] -- 參數(shù)key
     * ARGV[1]-- 失效時間段內最大放行次數(shù)
     * ARGV[2]-- 失效時間|秒
     */
    private String buildLuaScript() {
        StringBuilder luaBuilder = new StringBuilder();
        //定義變量
        luaBuilder.append("local count");
        //獲取調用腳本時傳入的第一個key值（用作限流的 key）
        luaBuilder.append("\ncount = redis.call('get',KEYS[1])");
        // 獲取調用腳本時傳入的第一個參數(shù)值（限流大小）-- 調用不超過最大值，則直接返回
        luaBuilder.append("\nif count and tonumber(count) > tonumber(ARGV[1]) then");
        luaBuilder.append("\nreturn count;");
        luaBuilder.append("\nend");
        //執(zhí)行計算器自增
        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限流切面實現(xiàn)

/**
 * 描述:限流切面實現(xiàn)
 *
 * @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);

        //限流組件-通過計數(shù)方式限流
        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()) {
            //未超過限流次數(shù)-執(zhí)行業(yè)務方法
            return joinPoint.proceed();
        } else {
            //超過限流次數(shù)
            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 則從參數(shù)中獲取該字段的值拼接到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();
        }

        //處理自定義-參數(shù)名-動態(tài)屬性key
        StringBuilder rateLimitKeyBuilder = new StringBuilder(rateLimit.key());
        for (Object obj : joinPoint.getArgs()) {
            if (null == obj) {
                continue;
            }
            //過濾基本類型參數(shù)
            if (ReflectionUtil.isBaseType(obj.getClass())) {
                continue;
            }
            //屬性值
            Object fieldValue = ReflectionUtil.getFieldByClazz(fieldName, obj);
            if (null != fieldValue) {
                rateLimitKeyBuilder.append(":").append(fieldValue.toString());
                break;
            }
        }
        return rateLimitKeyBuilder.toString();
    }
}

由于演示項目中做了統(tǒng)一異常處理
在限流切面這里未做異常捕獲，若超過最大限流次數(shù)會拋出自定義限流異常?？梢愿鶕?jù)業(yè)務自行處理。

/**
 * 反射工具
 *
 * @author 程序員小強
 */
public class ReflectionUtil {

    private static final Logger logger = LoggerFactory.getLogger(ReflectionUtil.class);

    /**
     * 根據(jù)屬性名獲取屬性元素，
     * 包括各種安全范圍和所有父類
     *
     * @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 {
                    //子類中查詢不到屬性-繼續(xù)向父類查
                    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;
    }

    /**
     * 判斷對象屬性是否是基本數(shù)據(jù)類型,包括是否包括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方法限流示例

  /**
   * 計數(shù)器
   * 演示 demo 為了方便計數(shù)
   */
  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動態(tài)入參限流示例

3.2.1場景一：5分鐘內，方法最多訪問10次，根據(jù)入參手機號限流

入參類

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<>();


  /**
   * 根據(jù)手機號限流-限制驗證碼發(fā)送次數(shù)
   * <p>
   * 示例：5分鐘內，驗證碼最多發(fā)送10次
   */
  @RequestMapping("/limitByPhone")
  @RateLimit(key = "limit-phone-key", time = 300, count = 10, keyField = "phone", msg = "5分鐘內，驗證碼最多發(fā)送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場景二：根據(jù)訂單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<>();

  /**
   * 根據(jù)訂單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.其它擴展

根據(jù)ip限流

在key中拼接IP即可；

5.源碼地址

傳送門

到此這篇關于Redis分布式限流組件設計與使用實例的文章就介紹到這了,更多相關Redis分布式限流內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

您可能感興趣的文章:

分割超大Redis數(shù)據(jù)庫例子
這篇文章主要介紹了分割超大Redis數(shù)據(jù)庫例子,本文講解了分割的需求、分割的思路及分割實例,需要的朋友可以參考下
2015-03-03
淺談Redis緩存有哪些淘汰策略
redis用做緩存是一種非常常見的手段,然而由于內存大小的限制，會導致redis在內存空間滿了以后需要處理繼續(xù)存入的數(shù)據(jù)，所以就需要淘汰策略，本文就詳細的介紹一下
2021-08-08
redis使用不當導致應用卡死bug的過程解析
本文主要記一次找因redis使用不當導致應用卡死bug的過程，文中通過示例代碼介紹的非常詳細，需要的朋友們下面隨著小編來一起學習學習吧
2021-07-07
redis?setIfAbsent返回null的問題及解決
這篇文章主要介紹了redis?setIfAbsent返回null的問題及解決方案，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教
2022-11-11
詳解Redis實現(xiàn)分布式鎖的原理
分布式鎖,即分布式系統(tǒng)中的鎖,在單體應用中我們通過鎖解決的是控制共享資源訪問的問題,而分布式鎖,就是解決了分布式系統(tǒng)中控制共享資源訪問的問題,本文講給大家詳細介紹一下Redis實現(xiàn)分布式鎖的原理,需要的朋友可以參考下
2023-09-09
Redis中redis-shake實現(xiàn)數(shù)據(jù)遷移同步
在業(yè)務環(huán)境中使用Redis進行跨區(qū)域數(shù)據(jù)遷移和同步的挑戰(zhàn),文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧
2025-11-11
Redis @type坑的解決
新建一個對象存入redis中,對象中會出現(xiàn)一個字段@type,本文主要介紹了Redis @type坑的解決,具有一定的參考價值,感興趣的可以了解一下
2024-06-06
redis中RedissonLock如何實現(xiàn)等待鎖的
本文主要介紹了redis中RedissonLock如何實現(xiàn)等待鎖的，文中通過示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下
2021-11-11
詳解Redis單線程的正確理解
這篇文章主要介紹了詳解Redis單線程的正確理解，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧
2021-05-05
Redis中set類型實現(xiàn)交集并集差集
本文主要介紹了Redis中set類型實現(xiàn)交集并集差集，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧
2023-06-06