当前位置： 首页 > news >正文

基于Redis实现限流

news 2025/3/14 18:46:37

限流尽可能在满足需求的情况下越简单越好！

分布式限流是指在分布式系统中对请求进行限制，以防止系统过载或滥用资源。以下是常见的分布式限流策略及其实现方式：

1、基于 Redis 的固定窗口限流

原理：

  • 设定一个时间窗口（如 1 秒）
  • 使用 Redis 维护一个计数器，存储当前窗口的请求数
  • 当请求到来时，INCR 计数器，如果超过阈值则拒绝
  • 过期后自动删除键，进入下一个窗口

优缺点： ✅ 简单易实现
❌ 在窗口交界处可能会出现短时间的突发流量（"临界突增"）

public class RedisRateLimiter {private final StringRedisTemplate redisTemplate;// 命令前缀private final String key;private final int rate;private final int window;public RedisRateLimiter(StringRedisTemplate redisTemplate, String key, int rate,int window) {this.redisTemplate = redisTemplate;this.key = key;this.rate = rate;Assert.isTrue(window > 0 && window <= 60,"窗口只支持分钟内");this.window = window;}// 检查并获取令牌public boolean acquire() {String currentKey = key + "_" + (DateUtil.currentSeconds() / window);Long currentCount = redisTemplate.opsForValue().increment(currentKey);redisTemplate.expire(currentKey, window, TimeUnit.SECONDS);if (currentCount > rate){return false;}return true;}public void acquireSleep() {int count = 0;while (!acquire()){ThreadUtil.sleep(1,TimeUnit.SECONDS);count++;log.info("RedisRateLimiter[{}] try acquire sleep {}",key,count);}}public boolean acquireSleep(int waitSecond) {int count = 0;while (!acquire()){if (count >= waitSecond){return false;}ThreadUtil.sleep(1,TimeUnit.SECONDS);count++;log.info("RedisRateLimiter[{}] try acquire sleep {}",key,count);}return true;}}

使用案例：

下面这个任务是实时请求评论和子评论接口，但是两个接口每分钟不能超过100，所以我们使用限流限制10秒不超过18即可也能满足需求。

public class ScCommentRealTimeSyncTask  {private RedisRateLimiter rateLimiter;@PostConstructpublic void init(){rateLimiter = newRedisRateLimiter(stringRedisTemplate,KAOLA_COMMENT_RATE_KEY,16,10);}@Scheduled(fixedDelay = 3000)public void task(){// 请求接口1rateLimiter.acquireSleep();request1();//请求接口2rateLimiter.acquireSleep();request2();}}

2. 基于 Redis 的滑动窗口限流

原理：

  • 维护一个基于时间的列表（ZSET，有序集合）
  • 每次请求时，记录当前时间戳到 ZSET
  • 删除超出窗口时间范围的请求
  • 统计 ZSET 中当前窗口内的请求数，超出阈值则拒绝

优缺点： ✅ 解决了固定窗口的临界突增问题
❌ 存储和计算成本比固定窗口稍高

原理说明

  • 利用 Redis 的有序集合（ZSet），以请求的时间戳作为 score，每个请求入队一个唯一的 member（例如时间戳+UUID）。
  • 每次请求时，先移除时间窗口外的记录（score 小于当前时间减去窗口长度）。
  • 统计当前窗口内的请求数量，若数量超过设定阈值，则拒绝请求。
@Slf4j
public class RedisSlidingWindowRateLimiter {private final StringRedisTemplate redisTemplate;private final String key;private final int rate;private final int window; // 窗口长度，单位秒public RedisSlidingWindowRateLimiter(StringRedisTemplate redisTemplate, String key, int rate, int window) {this.redisTemplate = redisTemplate;this.key = key;this.rate = rate;// 限制窗口长度在 1 分钟以内Assert.isTrue(window > 0 && window <= 60, "窗口只支持一分钟内");this.window = window;}// 检查并获取令牌public boolean acquire() {long now = System.currentTimeMillis();// 计算窗口起始时间（单位毫秒）long windowStart = now - window * 1000;// 移除过期的请求记录redisTemplate.opsForZSet().removeRangeByScore(key, 0, windowStart);// 添加当前请求记录，member 用当前时间戳加 UUID 保证唯一性，score 为当前时间String member = now + "_" + UUID.randomUUID().toString();redisTemplate.opsForZSet().add(key, member, now);// 统计当前窗口内的请求数量Long count = redisTemplate.opsForZSet().count(key, windowStart, now);// 为了避免 key 永不过期，设置一个过期时间（窗口长度）redisTemplate.expire(key, window, TimeUnit.SECONDS);if (count != null && count > rate) {return false;}return true;}// 采用轮询方式等待获取令牌public void acquireSleep() {int count = 0;while (!acquire()){ThreadUtil.sleep(1, TimeUnit.SECONDS);count++;log.info("RedisSlidingWindowRateLimiter[{}] try acquire sleep {}", key, count);}}public boolean acquireSleep(int waitSecond) {int count = 0;while (!acquire()){if (count >= waitSecond){return false;}ThreadUtil.sleep(1, TimeUnit.SECONDS);count++;log.info("RedisSlidingWindowRateLimiter[{}] try acquire sleep {}", key, count);}return true;}
}

代码说明

  • 移除过期记录：调用 removeRangeByScore 清理掉窗口外的请求数据。
  • 添加当前请求：将当前请求的时间戳与 UUID 组合后添加到 ZSet 中，score 为当前时间，确保在滑动窗口内计数。
  • 统计计数：通过 count 方法统计当前窗口内的请求数，如果超出限制则返回 false。

3. 基于 Redis 的令牌桶限流

原理：

  • 设定一个容量为 max_tokens 的令牌桶，初始装满
  • 以固定速率向桶中添加令牌（如每秒 10 个）
  • 每次请求需要消耗一个令牌，没有令牌时拒绝请求
  • 通常使用 Redis 的 Lua 脚本实现原子操作

优缺点： ✅ 更加平滑，支持突发流量
❌ 需要额外的定时任务或后台线程补充令牌

原理说明

  • 令牌桶算法中，设定一个桶最大容量 capacity，同时以一定速率 refillRate 补充令牌。
  • 每次请求需要消耗一个令牌，若当前桶内令牌不足，则拒绝请求。
  • 为保证原子性，利用 Redis 的 Lua 脚本将令牌获取和补充过程封装为原子操作。
@Slf4j
public class RedisTokenBucketRateLimiter {private final StringRedisTemplate redisTemplate;private final String key;// 桶的容量（最大令牌数）private final int capacity;// 令牌补充速率，单位：个/秒private final double refillRate;// Lua 脚本，用于原子化处理令牌桶逻辑private static final String LUA_SCRIPT = "local tokens_key = KEYS[1] .. ':tokens' \n" +"local timestamp_key = KEYS[1] .. ':ts' \n" +"local capacity = tonumber(ARGV[1]) \n" +"local refill_rate = tonumber(ARGV[2]) \n" +"local current_time = tonumber(ARGV[3]) \n" +"local requested = tonumber(ARGV[4]) \n" +"local tokens = tonumber(redis.call('get', tokens_key) or capacity) \n" +"local last_refill = tonumber(redis.call('get', timestamp_key) or current_time) \n" +"local delta = current_time - last_refill \n" +"local tokens_to_add = delta * refill_rate \n" +"tokens = math.min(capacity, tokens + tokens_to_add) \n" +"if tokens < requested then \n" +"   return 0 \n" +"else \n" +"   tokens = tokens - requested \n" +"   redis.call('set', tokens_key, tokens) \n" +"   redis.call('set', timestamp_key, current_time) \n" +"   return 1 \n" +"end";public RedisTokenBucketRateLimiter(StringRedisTemplate redisTemplate, String key, int capacity, double refillRate) {this.redisTemplate = redisTemplate;this.key = key;this.capacity = capacity;this.refillRate = refillRate;}// 检查并获取令牌public boolean acquire() {// 当前时间（单位秒）long currentTime = System.currentTimeMillis() / 1000;// 请求消耗 1 个令牌Long result = redisTemplate.execute((RedisCallback<Long>) connection -> {List<byte[]> keys = Collections.singletonList(key.getBytes());List<byte[]> args = Arrays.asList(String.valueOf(capacity).getBytes(),String.valueOf(refillRate).getBytes(),String.valueOf(currentTime).getBytes(),"1".getBytes());return connection.eval(LUA_SCRIPT.getBytes(), ReturnType.INTEGER, keys.size(), keys.toArray(new byte[0][]), args.toArray(new byte[0][]));});return result != null && result == 1;}
}

代码说明

  • Lua 脚本逻辑：
    • 获取当前桶中剩余令牌数和上次补充时间，若不存在则默认初始化为满桶状态。
    • 根据当前时间与上次更新时间的差值计算应补充的令牌数，并更新桶内令牌。
    • 判断是否有足够令牌供本次请求（默认请求 1 个令牌），若不足返回 0，否则扣减令牌并更新上次补充时间，返回 1。
  • 原子执行：通过 redisTemplate 的 eval 方法保证 Lua 脚本的原子性，避免并发问题。

4. 基于 Redis 的漏桶限流

原理：

  • 设定一个队列模拟漏桶
  • 按固定速率从队列取出请求执行
  • 请求过多时，超出队列长度的请求被丢弃

优缺点： ✅ 输出速率稳定，不受突发流量影响
❌ 可能会丢弃部分流量

原理说明

  • 漏桶算法中，将请求看作向桶中注入的“水”，桶以固定速率漏水（处理请求）。
  • 当桶中水量超过预设容量时，则拒绝新请求。
  • 同样利用 Lua 脚本保证原子操作。
@Slf4j
public class RedisLeakyBucketRateLimiter {private final StringRedisTemplate redisTemplate;private final String key;// 桶的容量（允许的最大突发请求数）private final int capacity;// 漏水速率，单位：个/秒，表示每秒可处理的请求数private final double leakRate;// Lua 脚本，用于原子化处理漏桶逻辑private static final String LUA_SCRIPT = "local level_key = KEYS[1] .. ':level' \n" +"local timestamp_key = KEYS[1] .. ':ts' \n" +"local capacity = tonumber(ARGV[1]) \n" +"local leak_rate = tonumber(ARGV[2]) \n" +"local current_time = tonumber(ARGV[3]) \n" +"local level = tonumber(redis.call('get', level_key) or '0') \n" +"local last_time = tonumber(redis.call('get', timestamp_key) or current_time) \n" +"local delta = current_time - last_time \n" +"local leaked = delta * leak_rate \n" +// 计算漏水后桶内水量，不能低于 0"level = math.max(0, level - leaked) \n" +"if level + 1 > capacity then \n" +"   return 0 \n" +"else \n" +"   level = level + 1 \n" +"   redis.call('set', level_key, level) \n" +"   redis.call('set', timestamp_key, current_time) \n" +"   return 1 \n" +"end";public RedisLeakyBucketRateLimiter(StringRedisTemplate redisTemplate, String key, int capacity, double leakRate) {this.redisTemplate = redisTemplate;this.key = key;this.capacity = capacity;this.leakRate = leakRate;}// 检查并获取请求处理资格public boolean acquire() {// 当前时间（单位秒）long currentTime = System.currentTimeMillis() / 1000;Long result = redisTemplate.execute((RedisCallback<Long>) connection -> {List<byte[]> keys = Collections.singletonList(key.getBytes());List<byte[]> args = Arrays.asList(String.valueOf(capacity).getBytes(),String.valueOf(leakRate).getBytes(),String.valueOf(currentTime).getBytes());return connection.eval(LUA_SCRIPT.getBytes(), ReturnType.INTEGER, keys.size(), keys.toArray(new byte[0][]), args.toArray(new byte[0][]));});return result != null && result == 1;}
}

代码说明

  • Lua 脚本逻辑：
    • 从 Redis 中获取当前桶内水量（即请求数量）和上次更新的时间。
    • 根据当前时间与上次更新时间的差值和设定的漏水速率计算“漏掉”的水量，并更新桶内水量（不能低于 0）。
    • 判断加入当前请求后是否超过桶的容量，超过则返回 0（拒绝），否则将水量加 1 并更新记录，返回 1 表示允许。
  • 原子执行：同样通过 eval 方法保证操作原子性，避免并发修改问题。

总结

  • 滑动窗口：使用 Redis ZSet 记录请求时间戳，动态统计窗口内请求数，平滑控制突发流量。
  • 令牌桶：通过 Lua 脚本实现令牌的自动补充和扣减，支持一定的突发请求。
  • 漏桶：用固定漏水速率保证请求以均匀的速率被处理，避免瞬间大量请求。

http://www.mrgr.cn/news/94376.html

相关文章：

  • Pytorch实现之BCGAN实现双生成器架构的人脸面部生成
  • Interview preparation.md
  • Redis 2025/3/9
  • 整合记录-持续
  • oracle11.2.0.4 RAC 保姆级静默安装（二) DB数据库软件
  • ES 使用geo point 查询离目标地址最近的数据
  • 【lf中的git实战】(我的代码合并到别人那用squash,别人合我这不用!!!)
  • python画图文字显示不全+VScode新建jupyter文件
  • Webservice如何调用
  • 【Node.js入门笔记4---fs 目录操作】
  • 微信小程序从右向左无限滚动组件封装(类似公告)
  • AI学习——深度学习核心技术深度解析
  • Linux--gdb/cgdb
  • Linux入门 全面整理终端 Bash、Vim 基础命令速记
  • 缓存和客户端数据存储体系（Ark Data Kit）--- 应用数据持久化(首选项持久化、K-V、关系型数据库）持续更新中...
  • Ubuntu20.04安装运行DynaSLAM
  • ArcGIS Pro 车牌分区数据处理与地图制作全攻略
  • 深度学习项目--基于DenseNet网络的“乳腺癌图像识别”，准确率90%+，pytorch复现
  • 考研408-数据结构完整代码 线性表的顺序存储结构 - 顺序表
  • 【Java从入门到精通】一篇文章彻底搞懂：类和对象到底是什么？