什么是限流？限流算法有哪些？怎么实现的？
原创
什么是限流？限流算法有哪些？怎么实现的？
原创
代码小李
发布于 2025-01-27 19:37:14
发布于 2025-01-27 19:37:14
限流是指在系统处理请求时，通过限制单位时间内可以处理的请求数量来保护系统不被过多的流量压垮。这种技术广泛应用于高并发场景中，比如电商平台的秒杀活动、API接口调用等，以防止系统过载或崩溃。
限流算法
计数器算法（Counter）- 原理：在一定时间窗口内统计请求次数，如果超过设定的阈值，则拒绝后续请求。
- 实现：
import time
class CounterLimiter:
def __init__(self, max_requests, window_size):
self.max_requests = max_requests
self.window_size = window_size
self.requests = 0
self.start_time = time.time()
def allow_request(self):
current_time = time.time()
if current_time - self.start_time > self.window_size:
self.requests = 0
self.start_time = current_time
if self.requests < self.max_requests:
self.requests += 1
return True
else:
return False漏桶算法（Leaky Bucket）
- 原理：请求进入一个固定容量的桶中，然后以恒定的速度从桶中流出。如果桶满了，新的请求会被丢弃。
- 实现
import time
from collections import deque
class LeakyBucketLimiter:
def __init__(self, capacity, out_rate):
self.capacity = capacity
self.out_rate = out_rate
self.bucket = deque(maxlen=capacity)
self.last_time = time.time()
def allow_request(self):
current_time = time.time()
elapsed_time = current_time - self.last_time
self.last_time = current_time
# 模拟水滴流出
while self.bucket and self.bucket[0] <= current_time - (1 / self.out_rate):
self.bucket.popleft()
if len(self.bucket) < self.capacity:
self.bucket.append(current_time)
return True
else:
return False令牌桶算法（Token Bucket）
- 原理：系统以恒定的速度生成令牌，请求需要消耗一个令牌才能通过。如果令牌用完，新的请求会被拒绝。
- 实现：
import time
class TokenBucketLimiter:
def __init__(self, capacity, refill_rate):
self.capacity = capacity
self.refill_rate = refill_rate
self.tokens = capacity
self.last_refill_time = time.time()
def allow_request(self):
current_time = time.time()
elapsed_time = current_time - self.last_refill_time
self.last_refill_time = current_time
# 生成新的令牌
new_tokens = elapsed_time * self.refill_rate
self.tokens = min(self.capacity, self.tokens + new_tokens)
if self.tokens >= 1:
self.tokens -= 1
return True
else:
return False滑动窗口算法（Sliding Window）
- 原理：将时间窗口划分为多个小的时间段，每个时间段记录请求次数，通过滑动窗口的方式计算当前时间窗口内的请求总数。
- 实现：
import time
class SlidingWindowLimiter:
def __init__(self, max_requests, window_size, interval):
self.max_requests = max_requests
self.window_size = window_size
self.interval = interval
self.requests = [0] * window_size
self.current_index = 0
def allow_request(self):
current_time = time.time()
elapsed_time = int(current_time // self.interval)
# 更新请求计数
if elapsed_time != self.current_index:
self.requests[self.current_index % self.window_size] = 0
self.current_index = elapsed_time
if sum(self.requests) < self.max_requests:
self.requests[self.current_index % self.window_size] += 1
return True
else:
return False原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
目录
腾讯云开发者
Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有
深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 
粤公网安备44030502008569号
腾讯云计算（北京）有限责任公司 京ICP证150476号 | 京ICP备11018762号