Redis 除了做緩存，還能干很多很多事情：分布式鎖、限流、處理請求接口冪等性。。。太多太多了～

今天想和小伙伴們聊聊用 Redis 處理接口限流，這也是最近的 TienChin 項目涉及到這個知識點了，我就拎出來和大家聊聊這個話題，后面視頻也會講。

1. 準備工作

首先我們創建一個 Spring Boot 工程，引入 Web 和 Redis 依賴，同時考慮到接口限流一般是通過注解來標記，而注解是通過 AOP 來解析的，所以我們還需要加上 AOP 的依賴，最終的依賴如下：

<dependency>
    <groupId>org.springframework.bootgroupId>
    <artifactId>spring-boot-starter-data-redisartifactId>
dependency>
<dependency>
    <groupId>org.springframework.bootgroupId>
    <artifactId>spring-boot-starter-webartifactId>
dependency>
<dependency>
    <groupId>org.springframework.bootgroupId>
    <artifactId>spring-boot-starter-aopartifactId>
dependency>

然后提前準備好一個 Redis 實例，這里我們項目配置好之后，直接配置一下 Redis 的基本信息即可，如下：

spring.redis.host=localhost
spring.redis.port=6379
spring.redis.password=123

好啦，準備工作就算是到位了。

2. 限流注解

接下來我們創建一個限流注解，我們將限流分為兩種情況：

1. 針對當前接口的全局性限流，例如該接口可以在 1 分鐘內訪問 100 次。
2. 針對某一個 IP 地址的限流，例如某個 IP 地址可以在 1 分鐘內訪問 100 次。

針對這兩種情況，我們創建一個枚舉類：

public enum LimitType {
    /**
     * 默認策略全局限流
     */
    DEFAULT,
    /**
     * 根據請求者IP進行限流
     */
    IP
}

接下來我們來創建限流注解：

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface RateLimiter {
    /**
     * 限流key
     */
    String key() default "rate_limit:";

    /**
     * 限流時間,單位秒
     */
    int time() default 60;

    /**
     * 限流次數
     */
    int count() default 100;

    /**
     * 限流類型
     */
    LimitType limitType() default LimitType.DEFAULT;
}

第一個參數限流的 key，這個僅僅是一個前綴，將來完整的 key 是這個前綴再加上接口方法的完整路徑，共同組成限流 key，這個 key 將被存入到 Redis 中。

另外三個參數好理解，我就不多說了。

好了，將來哪個接口需要限流，就在哪個接口上添加 @RateLimiter 注解，然后配置相關參數即可。

3. 定制 RedisTemplate

小伙伴們知道，在 Spring Boot 中，我們其實更習慣使用 Spring Data Redis 來操作 Redis，不過默認的 RedisTemplate 有一個小坑，就是序列化用的是 JdkSerializationRedisSerializer，不知道小伙伴們有沒有注意過，直接用這個序列化工具將來存到 Redis 上的 key 和 value 都會莫名其妙多一些前綴，這就導致你用命令讀取的時候可能會出錯。

例如存儲的時候，key 是 name，value 是 javaboy，但是當你在命令行操作的時候，get name 卻獲取不到你想要的數據，原因就是存到 redis 之后 name 前面多了一些字符，此時只能繼續使用 RedisTemplate 將之讀取出來。

我們用 Redis 做限流會用到 Lua 腳本，使用 Lua 腳本的時候，就會出現上面說的這種情況，所以我們需要修改 RedisTemplate 的序列化方案。

可能有小伙伴會說為什么不用 StringRedisTemplate 呢？StringRedisTemplate 確實不存在上面所說的問題，但是它能夠存儲的數據類型不夠豐富，所以這里不考慮。

修改 RedisTemplate 序列化方案，代碼如下：

@Configuration
public class RedisConfig {

    @Bean
    public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory connectionFactory) {
        RedisTemplate<Object, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(connectionFactory);
        // 使用Jackson2JsonRedisSerialize 替換默認序列化(默認采用的是JDK序列化)
        Jackson2JsonRedisSerializer<Object> 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.setKeySerializer(jackson2JsonRedisSerializer);
        redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);
        redisTemplate.setHashKeySerializer(jackson2JsonRedisSerializer);
        redisTemplate.setHashValueSerializer(jackson2JsonRedisSerializer);
        return redisTemplate;
    }
}

這個其實也沒啥好說的，key 和 value 我們都使用 Spring Boot 中默認的 jackson 序列化方式來解決。

4. 開發 Lua 腳本

這個其實我在之前 vhr 那一套視頻中講過，Redis 中的一些原子操作我們可以借助 Lua 腳本來實現，想要調用 Lua 腳本，我們有兩種不同的思路：

1. 在 Redis 服務端定義好 Lua 腳本，然后計算出來一個散列值，在 Java 代碼中，通過這個散列值鎖定要執行哪個 Lua 腳本。
2. 直接在 Java 代碼中將 Lua 腳本定義好，然后發送到 Redis 服務端去執行。

Spring Data Redis 中也提供了操作 Lua 腳本的接口，還是比較方便的，所以我們這里就采用第二種方案。

我們在 resources 目錄下新建 lua 文件夾專門用來存放 lua 腳本，腳本內容如下：

local key = KEYS[1]
local count = tonumber(ARGV[1])
local time = tonumber(ARGV[2])
local current = redis.call('get', key)
if current and tonumber(current) > count then
    return tonumber(current)
end
current = redis.call('incr', key)
if tonumber(current) == 1 then
    redis.call('expire', key, time)
end
return tonumber(current)

這個腳本其實不難，大概瞅一眼就知道干啥用的。 KEYS 和 ARGV 都是一會調用時候傳進來的參數，tonumber 就是把字符串轉為數字，redis.call 就是執行具體的 redis 指令，具體流程是這樣：

1. 首先獲取到傳進來的 key 以及 限流的 count 和時間 time。
2. 通過 get 獲取到這個 key 對應的值，這個值就是當前時間窗內這個接口可以訪問多少次。
3. 如果是第一次訪問，此時拿到的結果為 nil，否則拿到的結果應該是一個數字，所以接下來就判斷，如果拿到的結果是一個數字，并且這個數字還大于 count，那就說明已經超過流量限制了，那么直接返回查詢的結果即可。
4. 如果拿到的結果為 nil，說明是第一次訪問，此時就給當前 key 自增 1，然后設置一個過期時間。
5. 最后把自增 1 后的值返回就可以了。

其實這段 Lua 腳本很好理解。

接下來我們在一個 Bean 中來加載這段 Lua 腳本，如下：

@Bean
public DefaultRedisScript<Long> limitScript() {
    DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
    redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("lua/limit.lua")));
    redisScript.setResultType(Long.class);
    return redisScript;
}

可以啦，我們的 Lua 腳本現在就準備好了。

5. 注解解析

接下來我們就需要自定義切面，來解析這個注解了，我們來看看切面的定義：

@Aspect
@Component
public class RateLimiterAspect {
    private static final Logger log = LoggerFactory.getLogger(RateLimiterAspect.class);

    @Autowired
    private RedisTemplate<Object, Object> redisTemplate;

    @Autowired
    private RedisScript<Long> limitScript;

    @Before("@annotation(rateLimiter)")
    public void doBefore(JoinPoint point, RateLimiter rateLimiter) throws Throwable {
        String key = rateLimiter.key();
        int time = rateLimiter.time();
        int count = rateLimiter.count();

        String combineKey = getCombineKey(rateLimiter, point);
        List<Object> keys = Collections.singletonList(combineKey);
        try {
            Long number = redisTemplate.execute(limitScript, keys, count, time);
            if (number==null || number.intValue() > count) {
                throw new ServiceException("訪問過于頻繁，請稍候再試");
            }
            log.info("限制請求'{}',當前請求'{}',緩存key'{}'", count, number.intValue(), key);
        } catch (ServiceException e) {
            throw e;
        } catch (Exception e) {
            throw new RuntimeException("服務器限流異常，請稍候再試");
        }
    }

    public String getCombineKey(RateLimiter rateLimiter, JoinPoint point) {
        StringBuffer stringBuffer = new StringBuffer(rateLimiter.key());
        if (rateLimiter.limitType() == LimitType.IP) {
            stringBuffer.append(IpUtils.getIpAddr(((ServletRequestAttributes) RequestContextHolder.currentRequestAttributes()).getRequest())).append("-");
        }
        MethodSignature signature = (MethodSignature) point.getSignature();
        Method method = signature.getMethod();
        Class<?> targetClass = method.getDeclaringClass();
        stringBuffer.append(targetClass.getName()).append("-").append(method.getName());
        return stringBuffer.toString();
    }
}

這個切面就是攔截所有加了 @RateLimiter 注解的方法，在前置通知中對注解進行處理。

1. 首先獲取到注解中的 key、time 以及 count 三個參數。
2. 獲取一個組合的 key，所謂的組合的 key，就是在注解的 key 屬性基礎上，再加上方法的完整路徑，如果是 IP 模式的話，就再加上 IP 地址。以 IP 模式為例，最終生成的 key 類似這樣：rate_limit:127.0.0.1-org.javaboy.ratelimiter.controller.HelloController-hello（如果不是 IP 模式，那么生成的 key 中就不包含 IP 地址）。
3. 將生成的 key 放到集合中。
4. 通過 redisTemplate.execute 方法取執行一個 Lua 腳本，第一個參數是腳本所封裝的對象，第二個參數是 key，對應了腳本中的 KEYS，后面是可變長度的參數，對應了腳本中的 ARGV。
5. 將 Lua 腳本執行的結果與 count 進行比較，如果大于 count，就說明過載了，拋異常就行了。

好了，大功告成了。

6. 接口測試

接下來我們就進行接口的一個簡單測試，如下：

@RestController
public class HelloController {
    @GetMapping("/hello")
    @RateLimiter(time = 5,count = 3,limitType = LimitType.IP)
    public String hello() {
        return "hello>>>"+new Date();
    }
}

每一個 IP 地址，在 5 秒內只能訪問 3 次。

這個自己手動刷新瀏覽器都能測試出來。

7. 全局異常處理

由于過載的時候是拋異常出來，所以我們還需要一個全局異常處理器，如下：

@RestControllerAdvice
public class GlobalException {
    @ExceptionHandler(ServiceException.class)
    public Map<String,Object> serviceException(ServiceException e) {
        HashMap<String, Object> map = new HashMap<>();
        map.put("status", 500);
        map.put("message", e.getMessage());
        return map;
    }
}

這是一個小 demo，我就不去定義實體類了，直接用 Map 來返回 JSON 了。

好啦，大功告成。

最后我們看看過載時的測試效果：

好啦，這就是我們使用 Redis 做限流的方式。

松哥最近正在錄制 TienChin 項目視頻，本文的內容也是 TienChin 項目視頻的一部分～TienChin 項目采用 Spring Boot+Vue3 技術棧，里邊會涉及到各種好玩的技術，小伙伴們來和松哥一起做一個完成率超 90% 的項目。