SpringBoot+Redis+Lua分佈式限流的實現

Posted on by

Redis支持LUA腳本的主要優勢

LUA腳本的融合將使Redis數據庫產生更多的使用場景，迸發更多新的優勢：

  • 高效性：減少網絡開銷及時延，多次redis服務器網絡請求的操作，使用LUA腳本可以用一個請求完成
  • 數據可靠性：Redis會將整個腳本作為一個整體執行，中間不會被其他命令插入。
  • 復用性：LUA腳本執行後會永久存儲在Redis服務器端，其他客戶端可以直接復用
  • 可嵌入性：可嵌入JAVA，C#等多種編程語言，支持不同操作系統跨平臺交互
  • 簡單強大：小巧輕便，資源占用率低，支持過程化和對象化的編程語言

自己也是第一次在工作中使用lua這種語言，記錄一下

創建Lua文件req_ratelimit.lua

local key = KEYS[1]   --限流KEY
local limitCount = tonumber(ARGV[1])       --限流大小
local limitTime = tonumber(ARGV[2])        --限流時間
local current = redis.call('get', key);
if current then
    if current + 1 > limitCount then --如果超出限流大小
        return 0
    else
        redis.call("INCRBY", key,"1")
        return current + 1
    end
else
    redis.call("set", key,"1")
    redis.call("expire", key,limitTime)
    return 1
end

自定義註解RateLimiter

package com.shinedata.ann;
 
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
 
 
@Target({ElementType.TYPE, ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
public @interface RateLimiter {
 
    /**
     * 限流唯一標識
     * @return
     */
    String key() default "rate.limit:";
 
    /**
     * 限流時間
     * @return
     */
    int time() default 1;
 
    /**
     * 限流次數
     * @return
     */
    int count() default 100;
 
    /**
     *是否限制IP,默認 否
     * @return
     */
    boolean restrictionsIp() default false;
}

定義切面RateLimiterAspect

package com.shinedata.aop;
 
import com.shinedata.ann.RateLimiter;
import com.shinedata.config.redis.RedisUtils;
import com.shinedata.exception.RateLimiterException;
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.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Component;
import org.springframework.web.context.request.RequestContextHolder;
import org.springframework.web.context.request.ServletRequestAttributes;
 
import javax.annotation.PostConstruct;
import javax.servlet.http.HttpServletRequest;
import java.io.Serializable;
import java.lang.reflect.Method;
import java.util.Collections;
import java.util.List;
 
/**
 * @ClassName RateLimiterAspect
 * @Author yupanpan
 * @Date 2020/5/6 13:46
 */
@Aspect
@Component
public class RateLimiterAspect {
 
    private final Logger logger	= LoggerFactory.getLogger(this.getClass());
 
    private static ThreadLocal<String> ipThreadLocal=new ThreadLocal();
 
    private DefaultRedisScript<Number> redisScript;
 
    @PostConstruct
    public void init(){
        redisScript = new DefaultRedisScript<Number>();
        redisScript.setResultType(Number.class);
        redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("redis/req_ratelimit.lua")));
    }
 
    @Around("@annotation(com.shinedata.ann.RateLimiter)")
    public Object interceptor(ProceedingJoinPoint joinPoint) throws Throwable {
        try {
            MethodSignature signature = (MethodSignature) joinPoint.getSignature();
            Method method = signature.getMethod();
            Class<?> targetClass = method.getDeclaringClass();
            RateLimiter rateLimit = method.getAnnotation(RateLimiter.class);
 
            if (rateLimit != null) {
                HttpServletRequest request = ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getRequest();
                boolean restrictionsIp = rateLimit.restrictionsIp();
                if(restrictionsIp){
                    ipThreadLocal.set(getIpAddr(request));
                }
 
                StringBuffer stringBuffer = new StringBuffer();
                stringBuffer.append(rateLimit.key());
                if(StringUtils.isNotBlank(ipThreadLocal.get())){
                    stringBuffer.append(ipThreadLocal.get()).append("-");
                }
                stringBuffer.append("-").append(targetClass.getName()).append("- ").append(method.getName());
 
                List<String> keys = Collections.singletonList(stringBuffer.toString());
 
                Number number = RedisUtils.execute(redisScript, keys, rateLimit.count(), rateLimit.time());
 
                if (number != null && number.intValue() != 0 && number.intValue() <= rateLimit.count()) {
                    logger.info("限流時間段內訪問第：{} 次", number.toString());
                    return joinPoint.proceed();
                }else {
                    logger.error("已經到設置限流次數,當前次數:{}",number.toString());
                    throw new RateLimiterException("服務器繁忙,請稍後再試");
                }
            } else {
                return joinPoint.proceed();
            }
        }finally {
            ipThreadLocal.remove();
        }
    }
 
    public static String getIpAddr(HttpServletRequest request) {
        String ipAddress = null;
        try {
            ipAddress = request.getHeader("x-forwarded-for");
            if (ipAddress == null || ipAddress.length() == 0 || "unknown".equalsIgnoreCase(ipAddress)) {
                ipAddress = request.getHeader("Proxy-Client-IP");
            }
            if (ipAddress == null || ipAddress.length() == 0 || "unknown".equalsIgnoreCase(ipAddress)) {
                ipAddress = request.getHeader("WL-Proxy-Client-IP");
            }
            if (ipAddress == null || ipAddress.length() == 0 || "unknown".equalsIgnoreCase(ipAddress)) {
                ipAddress = request.getRemoteAddr();
            }
            // 對於通過多個代理的情況，第一個IP為客戶端真實IP,多個IP按照','分割
            if (ipAddress != null && ipAddress.length() > 15) {
                // "***.***.***.***".length()= 15
                if (ipAddress.indexOf(",") > 0) {
                    ipAddress = ipAddress.substring(0, ipAddress.indexOf(","));
                }
            }
        } catch (Exception e) {
            ipAddress = "";
        }
        return ipAddress;
    }
}

Spring data redis提供瞭DefaultRedisScript來使用lua和redis進行交互，具體的詳情網上很多文章,這裡使用ThreadLocal是因為IP存在可變的，保證自己的線程的IP不會被其他線程所修改，切記要最後清理ThreadLocal,防止內存泄漏

RedisUtils工具類(方法太多,隻展示execute方法)

package com.shinedata.config.redis;
 
import org.checkerframework.checker.units.qual.K;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
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.util.CollectionUtils;
 
import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
 
/**
 * @ClassName RedisUtils
 * @Author yupanpan
 * @Date 2019/11/20 13:38
 */
@Component
public class RedisUtils {
 
    @Autowired
    @Qualifier("redisTemplate")
    private RedisTemplate<String, Object> redisTemplate;
 
    private static RedisUtils redisUtils;
    
    @PostConstruct
    public void init() {
        redisUtils = this;
        redisUtils.redisTemplate = this.redisTemplate;
    }
 
    public static Number execute(DefaultRedisScript<Number> script, List keys, Object... args) {
        return redisUtils.redisTemplate.execute(script, keys,args);
    }
}

自己配置的RedisTemplate

package com.shinedata.config.redis;
 
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import redis.clients.jedis.JedisPoolConfig;
 
/**
 * @ClassName RedisConfig
 * @Author yupanpan
 * @Date 2019/11/20 13:26
 */
@Configuration
public class RedisConfig extends RedisProperties{
 
    protected Logger log = LogManager.getLogger(RedisConfig.class);
 
    /**
     * JedisPoolConfig 連接池
     * @return
     */
    @Bean("jedisPoolConfig")
    public JedisPoolConfig jedisPoolConfig() {
        JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
        // 最大空閑數
        jedisPoolConfig.setMaxIdle(500);
        jedisPoolConfig.setMinIdle(100);
        // 連接池的最大數據庫連接數
        jedisPoolConfig.setMaxTotal(6000);
        // 最大建立連接等待時間
        jedisPoolConfig.setMaxWaitMillis(5000);
        // 逐出連接的最小空閑時間 默認1800000毫秒(30分鐘)
        jedisPoolConfig.setMinEvictableIdleTimeMillis(100);
        // 每次逐出檢查時 逐出的最大數目 如果為負數就是 : 1/abs(n), 默認3
//        jedisPoolConfig.setNumTestsPerEvictionRun(numTestsPerEvictionRun);
        // 逐出掃描的時間間隔(毫秒) 如果為負數,則不運行逐出線程, 默認-1
        jedisPoolConfig.setTimeBetweenEvictionRunsMillis(600);
        // 是否在從池中取出連接前進行檢驗,如果檢驗失敗,則從池中去除連接並嘗試取出另一個
        jedisPoolConfig.setTestOnBorrow(true);
        // 在空閑時檢查有效性, 默認false
        jedisPoolConfig.setTestWhileIdle(false);
        return jedisPoolConfig;
    }
 
    /**
     * JedisConnectionFactory
     * @param jedisPoolConfig
     */
    @Bean("jedisConnectionFactory")
    public JedisConnectionFactory jedisConnectionFactory(@Qualifier("jedisPoolConfig")JedisPoolConfig jedisPoolConfig) {
        JedisConnectionFactory JedisConnectionFactory = new JedisConnectionFactory(jedisPoolConfig);
        // 連接池
        JedisConnectionFactory.setPoolConfig(jedisPoolConfig);
        // IP地址
        JedisConnectionFactory.setHostName(redisHost);
        // 端口號
        JedisConnectionFactory.setPort(redisPort);
        // 如果Redis設置有密碼
         JedisConnectionFactory.setPassword(redisPassword);
        // 客戶端超時時間單位是毫秒
        JedisConnectionFactory.setTimeout(10000);
        return JedisConnectionFactory;
    }
 
    /**
     * 實例化 RedisTemplate 對象代替原有的RedisTemplate<String, String>
     * @return
     */
    @Bean("redisTemplate")
    public RedisTemplate<String, Object> functionDomainRedisTemplate(@Qualifier("jedisConnectionFactory") RedisConnectionFactory redisConnectionFactory) {
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        initDomainRedisTemplate(redisTemplate, redisConnectionFactory);
        return redisTemplate;
    }
 
    /**
     * 設置數據存入 redis 的序列化方式
     * @param redisTemplate
     * @param factory
     */
    private void initDomainRedisTemplate(RedisTemplate<String, Object> redisTemplate, RedisConnectionFactory factory) {
        // 如果不配置Serializer，那麼存儲的時候缺省使用String，比如如果用User類型存儲，那麼會提示錯誤User can't cast
        // to String！
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());
        redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());
        // 開啟事務/true必須手動釋放連接，false會自動釋放連接 如果調用方有用@Transactional做事務控制，可以開啟事務,Spring會處理連接問題
        redisTemplate.setEnableTransactionSupport(false);
        redisTemplate.setConnectionFactory(factory);
    }
}

全局Controller異常處理GlobalExceptionHandler

package com.shinedata.exception;
 
import com.fasterxml.jackson.databind.JsonMappingException;
import com.shinedata.util.ResultData;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.HttpStatus;
import org.springframework.web.bind.annotation.ExceptionHandler;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.bind.annotation.RestControllerAdvice;
 
@RestControllerAdvice
public class GlobalExceptionHandler {
 
    private Logger logger = LoggerFactory.getLogger(GlobalExceptionHandler.class);
 
    @ExceptionHandler(value = RateLimiterException.class)
    @ResponseStatus(HttpStatus.OK)
    public ResultData runtimeExceptionHandler(RateLimiterException e) {
        logger.error("系統錯誤:", e);
        return ResultData.getResultError(StringUtils.isNotBlank(e.getMessage()) ? e.getMessage() : "處理失敗");
    }
 
    @ExceptionHandler(value = Exception.class)
    @ResponseStatus(HttpStatus.OK)
    public ResultData runtimeExceptionHandler(RuntimeException e) {
        Throwable cause = e.getCause();
        logger.error("系統錯誤:", e);
        logger.error(e.getMessage());
        if (cause instanceof JsonMappingException) {
            return ResultData.getResultError("參數錯誤");
        }
        return ResultData.getResultError(StringUtils.isNotBlank(e.getMessage()) ? e.getMessage() : "處理失敗");
    }
 
}

使用就很簡單瞭，一個註解搞定

補充:優化瞭lua為

local key = KEYS[1]
local limitCount = tonumber(ARGV[1])
local limitTime = tonumber(ARGV[2])
local current = redis.call('get', key);
if current then
    redis.call("INCRBY", key,"1")
    return current + 1
else
    redis.call("set", key,"1")
    redis.call("expire", key,limitTime)
    return 1
end

到此這篇關於SpringBoot+Redis+Lua分佈式限流的實現的文章就介紹到這瞭,更多相關SpringBoot Redis Lua分佈式限流內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！

推薦閱讀：