限流是指在系统处理请求时，通过限制单位时间内可以处理的请求数量来保护系统不被过多的流量压垮。这种技术广泛应用于高并发场景中，比如电商平台的秒杀活动、API接口调用等，以防止系统过载或崩溃。
限流算法计数器算法（Counter）原理：在一定时间窗口内统计请求次数，如果超过设定的阈值，则拒绝后续请求。实现：代码语言：python代码运行次数：0运行复制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）
原理：请求进入一个固定容量的桶中，然后以恒定的速度从桶中流出。如果桶满了，新的请求会被丢弃。实现代码语言：python代码运行次数：0运行复制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）
原理：系统以恒定的速度生成令牌，请求需要消耗一个令牌才能通过。如果令牌用完，新的请求会被拒绝。实现：代码语言：python代码运行次数：0运行复制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）
原理：将时间窗口划分为多个小的时间段，每个时间段记录请求次数，通过滑动窗口的方式计算当前时间窗口内的请求总数。实现：代码语言：python代码运行次数：0运行复制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