限流之redis 有序集合（zset）实现滑动窗口_redistemplate zadd

限流

需求：同一用户1分钟内登录失败次数超过3次，页面添加验证码登录验证，即限流的思想。

常见的限流算法：固定窗口计数器；滑动窗口计数器；漏桶；令牌桶。本篇选择的滑动窗口计数器

redis 有序集合（zset）特性

Redis 有序集合(sorted set)和集合(set)一样都是元素的集合，不允许重复的元素，但不同的是每个元素都会关联一个 double 类型的分数(score)。redis 正是通过分数来为集合中的元素进行从小到大的排序。有序集合的元素是唯一的,但分数(score)可以重复。

源码可对比java的LinkedHashMap和HashMap，都是通过多维护成员变量使无序的集合变成有序的。区别是LinkedHashMap内部是多维护了2个成员变量Entry<K,V> before, after用于双向链表的连接，redis zset是多维护了一个score变量完成顺序的排列。

排序：有序集合中的每个元素都关联着一个分数（score），通过分数来对元素进行排序。这使得有序集合在需要按照特定顺序访问元素的场景中非常有用。

唯一性：有序集合中的每个元素都是唯一的，不会存在重复的元素。

快速插入和删除：有序集合支持快速的插入和删除操作，时间复杂度为 O(log(N))，其中 N 是有序集合中元素的数量。

高效的范围操作：有序集合支持根据分数范围进行查询操作。可以按照分数从小到大或从大到小的顺序，获取指定范围内的元素。

成员访问：可以根据成员（元素）进行访问，例如获取指定成员的分数或排名。

底层实现：有序集合使用跳跃表（Skip List）和哈希表（Hash Table）两种数据结构来实现。跳跃表提供了快速的有序访问能力，而哈希表则提供了快速的成员访问能力。

支持事务和持久化：有序集合和其他 Redis 数据类型一样，可以通过事务（Transaction）和持久化（Persistence）功能来保证数据的一致性和可靠性。

有序集合在很多场景下都非常有用，如排行榜、计数器、范围查找等。它的灵活性和高性能使得 Redis 成为了许多应用程序的首选数据库之一。

滑动窗口算法

滑动窗口算法思想就是记录一个滑动的时间窗口内的操作次数，操作次数超过阈值则进行限流。

java代码实现

key-用户的登录名，value-使用zset。

zset：score-当前登录时间戳，value-也使用当前登录时间戳。

zset的value(元素)是唯一不可重复的，每个元素都有一个关联的分数。如果插入一个已经存在的元素，则会更新该元素的分数。value这里使用时间戳已满足我的需求，但value应该使用可唯一标识数据的值，例如雪花算法生成的唯一标识。

一、RedisTemplate

import java.util.List;
import javax.annotation.Resource;
 
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Component;
 
/**
 * 滑动窗口计数
 *
 * @author yangzihe
 * @date 2023/8/8
 */
@Component
@Slf4j
public class SlidingWindowCounter {
 
    @Resource
    private RedisTemplate<String, Object> redisTemplate;
 
    /**
     * 数据统计-判断数量是否超过最大限定值
     *
     * @param key        redis key
     * @param windowTime 窗口时间，单位：秒
     * @param maxNum     最大数量
     *
     * @return true-超过 false-未超过
     */
    public boolean countOver(String key, int windowTime, long maxNum) {
        // 窗口结束时间
        long windowEndTime = System.currentTimeMillis();
        // 窗口开始时间
        long windowStartTime = windowEndTime - windowTime * 1000L;
 
        // 按score统计key的value中的有效数量
        Long count = redisTemplate.opsForZSet().count(key, windowStartTime, windowEndTime);
        if (count == null) {
            return false;
        }
        return count > maxNum;
    }
 
    /**
     * 数据上报-滑动窗口计数增长
     *
     * @param key        redis key
     * @param windowTime 窗口时间，单位：秒
     */
    public void increment(String key, Integer windowTime) {
        // 窗口结束时间
        long windowEndTime = System.currentTimeMillis();
        // 窗口开始时间
        long windowStartTime = windowEndTime - windowTime * 1000L;
 
        RedisCallback<List<Object>> pipelineCallback = (connection) -> {
            RedisSerializer<String> serializer = redisTemplate.getStringSerializer();
 
            // 在管道中执行多个 ZSet 命令
            connection.openPipeline();
 
            // 添加当前时间 value=当前时间戳 score=当前时间戳
            connection.zAdd(serializer.serialize(key), windowEndTime, serializer.serialize(String.valueOf(windowEndTime)));
            // 清除窗口过期成员
            connection.zRemRangeByScore(serializer.serialize(key), 0, windowStartTime);
            // 设置key过期时间
            connection.expire(serializer.serialize(key), windowTime);
 
            return connection.closePipeline();
        };
 
        // 执行管道操作
        redisTemplate.executePipelined(pipelineCallback);
    }
 
    /**
     * 数据统计、数据上报同步处理，判断数量是否超过最大限定值
     *
     * @param key        redis key
     * @param windowTime 窗口时间，单位：秒
     * @param maxNum     最大数量
     *
     * @return true-超过 false-未超过
     */
    public boolean countAndIncrement(String key, int windowTime, long maxNum) {
        // 窗口结束时间
        long windowEndTime = System.currentTimeMillis();
        // 窗口开始时间
        long windowStartTime = windowEndTime - windowTime * 1000L;
 
        RedisCallback<List<Object>> pipelineCallback = (connection) -> {
            RedisSerializer<String> serializer = redisTemplate.getStringSerializer();
 
            // 在管道中执行多个 ZSet 命令
            connection.openPipeline();
 
            // 添加当前时间 value=当前时间戳 score=当前时间戳
            connection.zAdd(serializer.serialize(key), windowEndTime, serializer.serialize(String.valueOf(windowEndTime)));
            // 按score统计key的value中的有效数量
            connection.zCount(serializer.serialize(key), windowStartTime, windowEndTime);
            // 清除窗口过期成员
            connection.zRemRangeByScore(serializer.serialize(key), 0, windowStartTime);
            // 设置key过期时间
            connection.expire(serializer.serialize(key), windowTime);
 
            return connection.closePipeline();
        };
 
        // 执行管道操作 阻塞api，直到所有命令都被发送并得到结果
        List<Object> results = redisTemplate.executePipelined(pipelineCallback);
        Long count = (Long) results.get(1);
        if (count == null) {
            return false;
        }
        return count > maxNum;
    }
 
}

二、RedissonClient

    /**
     * 统计请求次数
     *
     * @param windowTime 窗口时间，单位：秒
     * @param key        redis key
     *
     * @return 请求次数
     */
    public static Integer count(int windowTime, String key) {
        // 窗口结束时间
        long windowEndTime = System.currentTimeMillis();
        // 窗口开始时间
        long windowStartTime = windowEndTime - windowTime * 1000L;
 
        try {
            // 创建 RBatch 实例，批量执行命令
            RBatch batch = redissonClient.createBatch();
 
            // 添加元素 score=当前时间戳 value=请求序列号，唯一不可重复
            batch.getScoredSortedSet(key).addAsync(windowEndTime, UUID.randomUUID().toString());
            // 统计数据
            batch.getScoredSortedSet(key).countAsync(windowStartTime, true, windowEndTime, true);
            // 清除窗口过期成员
            batch.getScoredSortedSet(key).removeRangeByScoreAsync(0, true, windowStartTime, false);
            // 设置key过期时间
            batch.getScoredSortedSet(key).expireAsync(Duration.ofSeconds(windowTime));
 
            // 执行管道命令
            BatchResult<?> batchResult = batch.execute();
 
            // 返回统计数量
            List<?> responses = batchResult.getResponses();
            return (Integer) responses.get(1);
        } catch (Exception e) {
            log.error("统计请求次数异常！", e);
            return null;
        }
    }