redis如何限流 redis限流算法的3种实现对比-Redis-PHP中文网

redis如何限流 redis限流算法的3种实现对比

尼克

发布： 2025-08-18 08:33:01

原创

729人浏览过

redis限流通过控制单位时间请求量保护系统，常见算法有计数器、令牌桶和漏桶。1.计数器算法简单但存在临界问题，适合对精度要求不高的场景；2.令牌桶允许突发流量，通过恒定填充令牌实现更平滑的限流；3.漏桶以固定速率处理请求，严格限制流量但无法应对突发流量。选择时需根据业务需求权衡实现复杂度与限流效果，并结合redis集群或分布式锁解决性能与一致性问题，从而构建稳定可靠的系统。

redis如何限流 redis限流算法的3种实现对比

Redis限流，本质上就是控制请求访问资源的速度，防止系统被突发流量冲垮。它通过限制单位时间内请求的数量来实现。

要理解Redis限流，得先明白它背后的原理。简单来说，就是利用Redis的原子性操作，比如

INCR

登录后复制

，来记录单位时间内的请求次数。当请求次数超过预设的阈值时，就拒绝后续的请求。

Redis限流：令牌桶、漏桶、计数器，哪种更适合你？

在Redis中实现限流，常见的算法有令牌桶、漏桶和计数器。选择哪种算法，取决于你的具体需求和场景。

1. 计数器算法：简单粗暴，但有缺陷

计数器算法是最简单的限流实现方式。它通过Redis的

INCR

登录后复制

命令，对某个键（比如用户ID）进行计数，并设置一个过期时间。每次请求到来时，计数加1，如果超过预设的阈值，就拒绝请求。

import redis
import time

redis_client = redis.Redis(host='localhost', port=6379, db=0)

def is_allowed(user_id, limit, period):
  """
  使用计数器算法进行限流

  Args:
    user_id: 用户ID
    limit: 单位时间内允许的请求数量
    period: 时间窗口，单位秒

  Returns:
    True: 允许请求
    False: 拒绝请求
  """
  key = f"limit:{user_id}"
  current_count = redis_client.incr(key)
  if current_count == 1:
    redis_client.expire(key, period)
  if current_count > limit:
    return False
  return True

# 示例
user_id = "user123"
limit = 5
period = 60  # 60秒内最多5次请求

for i in range(10):
  if is_allowed(user_id, limit, period):
    print(f"请求 {i+1} 允许")
  else:
    print(f"请求 {i+1} 拒绝")
  time.sleep(5)

登录后复制

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

缺点： 存在临界问题。如果在时间窗口的末尾和下一个时间窗口的开始，都发送了接近阈值的请求，那么实际的请求数量可能会超过阈值。例如，在第59秒发送了5个请求，在第61秒又发送了5个请求，虽然每个时间窗口内都没超过5个请求，但在2秒内却发送了10个请求。

2. 令牌桶算法：更平滑的限流

令牌桶算法以恒定的速率向桶中放入令牌。每个请求到来时，需要从桶中获取一个令牌，如果获取不到，就拒绝请求。

标书对比王

标书对比王是一款标书查重工具，支持多份投标文件两两相互比对，重复内容高亮标记，可快速定位重复内容原文所在位置，并可导出比对报告。

查看详情

import redis
import time

redis_client = redis.Redis(host='localhost', port=6379, db=0)

def is_allowed_token_bucket(user_id, limit, period, fill_rate):
  """
  使用令牌桶算法进行限流

  Args:
    user_id: 用户ID
    limit: 桶的容量
    period: 填充令牌的时间间隔，单位秒
    fill_rate: 令牌填充速率，单位 个/秒

  Returns:
    True: 允许请求
    False: 拒绝请求
  """
  key = f"token:{user_id}"
  now = time.time()
  last_refill_time = redis_client.get(f"last_refill:{user_id}")
  if last_refill_time is None:
    last_refill_time = now
  else:
    last_refill_time = float(last_refill_time)

  # 计算应该填充的令牌数量
  refill_tokens = (now - last_refill_time) * fill_rate
  if refill_tokens > 0:
    # 更新桶中的令牌数量
    current_tokens = redis_client.get(key)
    if current_tokens is None:
      current_tokens = 0
    else:
      current_tokens = int(current_tokens)

    current_tokens = min(limit, current_tokens + refill_tokens)
    redis_client.set(key, current_tokens)
    redis_client.set(f"last_refill:{user_id}", now)

  # 尝试获取令牌
  current_tokens = redis_client.get(key)
  if current_tokens is None or int(current_tokens) <= 0:
    return False
  else:
    redis_client.decr(key)
    return True

# 示例
user_id = "user456"
limit = 10
period = 1
fill_rate = 2  # 每秒填充2个令牌

for i in range(20):
  if is_allowed_token_bucket(user_id, limit, period, fill_rate):
    print(f"请求 {i+1} 允许")
  else:
    print(f"请求 {i+1} 拒绝")
  time.sleep(0.2)

登录后复制

优点： 允许一定程度的突发流量，因为桶中可以存储一定数量的令牌。

缺点： 实现相对复杂，需要维护令牌桶的状态。

3. 漏桶算法：更严格的限流

漏桶算法以恒定的速率从桶中漏出请求。请求先进入桶中，如果桶满了，就拒绝请求。

import redis
import time

redis_client = redis.Redis(host='localhost', port=6379, db=0)

def is_allowed_leaky_bucket(user_id, limit, rate):
  """
  使用漏桶算法进行限流

  Args:
    user_id: 用户ID
    limit: 桶的容量
    rate: 漏水速率，单位 个/秒

  Returns:
    True: 允许请求
    False: 拒绝请求
  """
  key = f"bucket:{user_id}"
  bucket_size = redis_client.llen(key)

  if bucket_size < limit:
    redis_client.lpush(key, time.time()) # 将当前时间戳放入桶中
    redis_client.expire(key, limit / rate + 1) # 设置过期时间，防止桶无限增长

    # 移除过期的请求
    while True:
      oldest_request_time = redis_client.rpop(key)
      if oldest_request_time is None:
        break
      if time.time() - float(oldest_request_time) < limit / rate:
        redis_client.rpush(key, oldest_request_time) # 重新放回桶中
        break

    return True
  else:
    return False


# 示例
user_id = "user789"
limit = 5 # 桶的容量
rate = 1 # 每秒漏出1个请求

for i in range(10):
  if is_allowed_leaky_bucket(user_id, limit, rate):
    print(f"请求 {i+1} 允许")
  else:
    print(f"请求 {i+1} 拒绝")
  time.sleep(0.5)

登录后复制

优点： 可以平滑流量，保证请求以恒定的速率被处理。

缺点： 无法处理突发流量，因为桶的容量是有限的。