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

引言

在微服务架构中，对 api 网关或单个服务的请求进行速率限制至关重要，以防止恶意攻击、资源滥用并确保系统的稳定性和可用性。 redis 凭借其高性能、原子操作和丰富的数据结构，成为实现请求限流的理想选择。

本文将详细探讨如何利用 redis 实现 api 网关或单个服务的请求限流，深入分析各种主流算法，并提供在 python 和 java 环境下的具体代码示例，包括使用 lua 脚本确保操作的原子性。

为什么选择 redis 进行限流？

redis 之所以成为实现限流的热门选择，主要得益于其以下几个关键特性：

卓越的性能： redis 是一款内存数据库，能够提供极快的读写速度，这对于需要实时处理大量请求的限流场景至关重要。
原子操作： redis 的 incr、expire 等命令是原子性的，可以避免在并发请求下出现竞态条件，确保计数和过期的准确性。
丰富的数据结构： redis 提供了字符串、哈希、列表、有序集合等多种数据结构，可以灵活地实现各种复杂的限流算法。
可扩展性： redis 支持集群和横向扩展，能够应对高并发的请求环境。
自带过期机制： 通过 expire 命令可以为键设置生存时间（ttl），轻松实现时间窗口的自动重置。

核心概念：识别请求来源

在实施限流之前，首先需要确定如何识别和区分不同的请求来源。常见的识别方式包括：

ip 地址： 基于客户端的 ip 地址进行限制，是简单有效的常用方法。
用户 id 或 api 密钥： 针对已认证的用户或第三方应用进行精细化限流。
设备 id： 对移动端等特定设备进行限制。

主流限流算法及其 redis 实现

以下是几种主流的限流算法及其使用 redis 的实现方式：

1. 固定窗口计数器 (fixed window counter)

这是最简单的限流算法。它在固定的时间窗口内（例如，每分钟）统计请求次数，如果超过预设的阈值，则拒绝后续的请求，直到下一个时间窗口开始。

优点： 实现简单，容易理解。

缺点： 在时间窗口的边界处可能会出现“突刺”流量问题。例如，在窗口结束前的瞬间和新窗口开始的瞬间，可能会有两倍于限制的请求通过。

redis 实现:

主要利用 incr 和 expire 命令。

python 示例:

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)

使用 lua 脚本保证原子性:

为了避免 get 和 incr 之间的竞态条件，强烈建议使用 lua 脚本将多个命令作为一个原子操作执行。

-- 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 表示未被限制

java (jedis) 调用 lua 脚本示例:

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());
    }
}

2. 滑动窗口日志 (sliding window log)

该算法记录每个请求的时间戳。当一个新请求到达时，会移除时间窗口之外的旧时间戳，然后统计窗口内的时间戳数量。如果数量超过限制，则拒绝请求。

优点： 限流精度高，有效解决了固定窗口的边界问题。

缺点： 需要存储所有请求的时间戳，当请求量很大时会占用较多内存。

redis 实现:

使用有序集合 (sorted set)，将成员 (member) 设置为唯一值（如请求 id 或时间戳），将分数 (score) 设置为请求的时间戳。

python 示例:

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

使用 lua 脚本实现滑动窗口:

-- 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 -- 未被限制

3. 令牌桶算法 (token bucket)

该算法的核心是一个固定容量的“令牌桶”，系统会以恒定的速率向桶里放入令牌。每个请求需要从桶里获取一个令牌才能被处理。如果桶里没有令牌，请求将被拒绝或排队等待。

优点： 能够应对突发流量，只要桶内有足够的令牌，就可以一次性处理多个请求。

缺点： 实现相对复杂，需要一个独立的进程或定时任务来持续生成令牌。

redis 实现:

可以使用 redis 的列表 (list) 作为令牌桶。一个后台任务（或在每次请求时计算）负责向列表中添加令牌。

java (spring boot with bucket4j) 示例:

bucket4j 是一个流行的 java 限流库，可以与 redis 结合使用实现分布式令牌桶。

// 依赖: 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 也内置了基于 redis 的令牌桶算法限流器。

4. 漏桶算法 (leaky bucket)

漏桶算法将请求看作是流入“漏桶”的水，而漏桶以固定的速率漏出水（处理请求）。如果流入的速率过快，导致桶溢出，则多余的请求将被丢弃。

优点： 能够平滑请求流量，保证服务以恒定的速率处理请求。

缺点： 无法有效利用系统空闲资源来处理突发流量。

redis 实现:

redis-cell 模块提供了基于 gcra (generic cell rate algorithm) 的漏桶算法实现。

api 网关限流 vs. 单个服务限流

限流逻辑可以部署在不同的层面，主要分为 api 网关层和单个微服务层。

api 网关限流

在 api 网关层面进行统一限流是一种常见的做法。

优点：

集中管理： 可以在一个地方统一配置和管理所有服务的限流规则。
保护后端服务： 将恶意或超额流量挡在微服务集群之外，保护整个系统的稳定性。

实现方式：

利用网关自带功能： 像 spring cloud gateway、kong、krakend 等 api 网关都提供了基于 redis 的限流插件或模块。
自定义中间件： 在网关中编写自定义的中间件或过滤器，嵌入上述的 redis 限流逻辑。

spring cloud gateway 示例 (application.yml):

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

单个服务限流

将限流逻辑直接实现在单个微服务内部。

优点：

灵活性： 每个服务可以根据自身的负载能力和业务需求，定制更精细化的限流策略。
独立性： 不依赖于特定的 api 网关。

实现方式：

通过 aop (面向切面编程) 创建注解，对需要限流的 controller 方法进行拦截。
在服务的入口处，如拦截器或过滤器中，调用 redis 进行限流判断。

python (fastapi 中间件) 示例:

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