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

苏承栈

2026-10-18 • 👁 0人浏览过

redis 限流通过控制单位时间请求量保护系统，常见算法有计数器、令牌桶和漏桶。1.计数器算法简单但存在临界问题，适合对精度要求不高的场景；2.令牌桶允许突发流量，通过恒定填充令牌实现更平滑的限流；3.漏桶以固定速率处理请求，严格限制流量但无法应对突发流量。选择时需根据业务需求权衡实现复杂度与限流效果，并结合redis集群或分布式锁解决性能与一致性问题，从而构建稳定可靠的系统。 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. 令牌桶算法：更平滑的限流令牌桶算法以恒定的速率向桶中放入令牌。每个请求到来时，需要从桶中获取一个令牌，如果获取不到，就拒绝请求。 Redis 8.2.3 Redis 8.2.3 是一款安全优先的高性能键值存储系统。该版本紧急修复了可能引发远程代码执行（RCE）的高危漏洞（CVE-2025-62507），并解决了 HyperLogLog 及 Cuckoo Filter 等数据结构在特定场景下的崩溃问题。建议所有用户立即升级，以保障生产环境的系统稳定与数据安全。下载

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)

优点：可以平滑流量，保证请求以恒定的速率被处理。缺点：无法处理突发流量，因为桶的容量是有限的。如何选择合适的Redis限流算法？选择哪种限流算法，需要根据具体的业务场景来决定。如果需要简单快速的实现，且对流量平滑性要求不高，可以选择计数器算法。如果需要允许一定程度的突发流量，可以选择令牌桶算法。如果需要严格控制请求的速率，保证请求以恒定的速率被处理，可以选择漏桶算法。 Redis限流的常见问题与解决方案在使用Redis限流的过程中，可能会遇到一些问题，比如： Redis性能瓶颈：如果限流的请求量非常大，可能会导致Redis的性能瓶颈。可以考虑使用Redis集群，或者使用更高效的限流算法。数据一致性问题：在分布式环境下，需要保证限流数据的一致性。可以使用Redis的分布式锁，或者使用CAP理论中的CP系统。误判问题：由于网络延迟等原因，可能会导致限流算法误判。可以适当调整限流的阈值，或者使用更复杂的限流算法。总的来说，Redis限流是一种非常有效的保护系统的方法。选择合适的限流算法，并解决可能遇到的问题，可以帮助你构建更稳定、更可靠的系统。

相关文章