使用Redis实现API网关或单个服务的请求限流的具体代码

import redis
import time

r = redis.Redis()

def is_rate_limited_fixed_window(user_id: str, limit: int, window: int) -> bool:
    key = f"rate_limit:{user_id}"
    current_requests = r.get(key)

    if current_requests is None:
        # 使用 pipeline 保证原子性
        pipe = r.pipeline()
        pipe.incr(key)
        pipe.expire(key, window)
        pipe.execute()
        return False

    if int(current_requests) >= limit:
        return True

    r.incr(key)
    return False

# 示例: 每位用户每60秒最多10个请求
for i in range(15):
    if is_rate_limited_fixed_window("user123", 10, 60):
        print("请求被限制")
    else:
        print("请求成功")
    time.sleep(1)

-- rate_limiter.lua
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])

local current = tonumber(redis.call("GET", key) or "0")

if current >= limit then
    return 1 -- 1 表示被限制
end

if current == 0 then
    redis.call("INCR", key)
    redis.call("EXPIRE", key, window)
else
    redis.call("INCR", key)
end

return 0 -- 0 表示未被限制

import redis.clients.jedis.Jedis;

public class RateLimiter {
    private final Jedis jedis;
    private final String script;

    public RateLimiter(Jedis jedis) {
        this.jedis = jedis;
        // 实际项目中应从文件加载
        this.script = "local key = KEYS[1]..."
    }

    public boolean isRateLimited(String key, int limit, int window) {
        Object result = jedis.eval(this.script, 1, key, String.valueOf(limit), String.valueOf(window));
        return "1".equals(result.toString());
    }
}

import redis
import time

r = redis.Redis()

def is_rate_limited_sliding_log(user_id: str, limit: int, window: int) -> bool:
    key = f"rate_limit_log:{user_id}"
    now = int(time.time() * 1000)
    window_start = now - window * 1000

    # 使用 pipeline 保证原子性
    pipe = r.pipeline()
    # 移除窗口外的数据
    pipe.zremrangebyscore(key, 0, window_start)
    # 添加当前请求
    pipe.zadd(key, {f"{now}:{int(time.time()*1000000)}": now}) # 保证 member 唯一
    # 获取窗口内的请求数
    pipe.zcard(key)
    # 设置过期时间，防止冷数据占用内存
    pipe.expire(key, window)
    results = pipe.execute()

    current_requests = results[2]
    return current_requests > limit

-- sliding_window.lua
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local now = redis.call("TIME")
local now_ms = (now[1] * 1000) + math.floor(now[2] / 1000)
local window_start = now_ms - window * 1000

-- 移除过期的请求记录
redis.call("ZREMRANGEBYSCORE", key, 0, window_start)

-- 获取当前窗口内的请求数
local count = redis.call("ZCARD", key)

if count >= limit then
    return 1 -- 被限制
end

-- 添加当前请求
redis.call("ZADD", key, now_ms, now_ms .. "-" .. count) -- member 保证唯一性
redis.call("EXPIRE", key, window)

return 0 -- 未被限制

// 依赖: bucket4j-core, jcache-api, redisson
// 配置 RedissonClient...

// 创建一个基于 Redis 的代理管理器
ProxyManager<String> proxyManager = new JCacheProxyManager<>(jcache);

// 定义令牌桶配置：每分钟补充10个令牌，桶容量为10
BucketConfiguration configuration = BucketConfiguration.builder()
        .addLimit(Bandwidth.simple(10, Duration.ofMinutes(1)))
        .build();

// 获取或创建令牌桶
Bucket bucket = proxyManager.getProxy("rate-limit-bucket:user123", configuration);

// 尝试消费一个令牌
if (bucket.tryConsume(1)) {
    // 请求成功
} else {
    // 请求被限制
}

spring:
  cloud:
    gateway:
      routes:
      - id: my_route
        uri: lb://my-service
        predicates:
        - Path=/my-api/**
        filters:
        - name: RequestRateLimiter
          args:
            redis-rate-limiter.replenishRate: 10      # 令牌桶每秒填充速率
            redis-rate-limiter.burstCapacity: 20     # 令牌桶容量
            redis-rate-limiter.requestedTokens: 1  # 每次请求消耗的令牌数
            key-resolver: "#{@ipKeyResolver}"     # 使用 IP 地址作为限流的 key

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import redis

app = FastAPI()
r = redis.Redis()

@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    # 简单示例：使用固定窗口计数器
    ip = request.client.host
    key = f"rate_limit:{ip}"
    limit = 5
    window = 60

    # 此处应使用 Lua 脚本保证原子性
    current = r.incr(key)
    if current == 1:
        r.expire(key, window)

    if current > limit:
        return JSONResponse(status_code=429, content={"message": "Too Many Requests"})

    response = await call_next(request)
    return response