滑动窗口限流案例

wen java案例 2026-08-04 158

本文目录导读：

滑动窗口限流案例

基础版：固定窗口计数器（演示问题）
滑动窗口日志版（精确但耗内存）
滑动窗口计数器版（高效实用）
Redis实现（分布式场景）
完整实战案例：API网关限流
高级优化版本：自适应限流

我来给你几个不同场景的滑动窗口限流案例,从简单到复杂。

基础版：固定窗口计数器（演示问题）

import time
import threading
class FixedWindowRateLimiter:
    """固定窗口限流器 - 演示边界问题"""
    def __init__(self, max_requests, window_seconds):
        self.max_requests = max_requests  # 窗口内最大请求数
        self.window_seconds = window_seconds  # 窗口大小
        self.current_window_start = time.time()
        self.request_count = 0
        self.lock = threading.Lock()
    def allow_request(self):
        with self.lock:
            now = time.time()
            # 如果当前时间超出窗口，重置计数
            if now - self.current_window_start >= self.window_seconds:
                self.current_window_start = now
                self.request_count = 0
            # 检查是否超限
            if self.request_count < self.max_requests:
                self.request_count += 1
                return True
            return False
# 演示边界问题
limiter = FixedWindowRateLimiter(max_requests=5, window_seconds=60)
# 在窗口边界处可能绕过限流

滑动窗口日志版（精确但耗内存）

from collections import deque
import time
import threading
class SlidingWindowLogRateLimiter:
    """滑动窗口日志限流器 - 使用时间戳记录"""
    def __init__(self, max_requests, window_seconds):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.requests = deque()  # 存储请求时间戳
        self.lock = threading.Lock()
    def allow_request(self):
        with self.lock:
            now = time.time()
            # 移除窗口外的旧请求
            while self.requests and now - self.requests[0] >= self.window_seconds:
                self.requests.popleft()
            # 检查当前窗口内请求数
            if len(self.requests) < self.max_requests:
                self.requests.append(now)
                return True
            return False
    def get_current_count(self):
        """获取当前窗口内请求数"""
        with self.lock:
            now = time.time()
            while self.requests and now - self.requests[0] >= self.window_seconds:
                self.requests.popleft()
            return len(self.requests)
# 使用示例
limiter = SlidingWindowLogRateLimiter(max_requests=10, window_seconds=60)
for i in range(15):
    if limiter.allow_request():
        print(f"请求 {i+1}: 允许")
    else:
        print(f"请求 {i+1}: 拒绝")

滑动窗口计数器版（高效实用）

import time
import threading
from collections import deque
class SlidingWindowCounterRateLimiter:
    """滑动窗口计数器限流器 - 使用子窗口聚合"""
    def __init__(self, max_requests, window_seconds, sub_window_count=10):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.sub_window_size = window_seconds / sub_window_count  # 每个子窗口大小
        self.windows = deque()  # (window_start_time, count) 对
        self.total_count = 0
        self.lock = threading.Lock()
    def allow_request(self):
        with self.lock:
            now = time.time()
            self._cleanup(now)
            if self.total_count < self.max_requests:
                self._add_request(now)
                return True
            return False
    def _cleanup(self, now):
        """清理过期的子窗口"""
        while self.windows and now - self.windows[0][0] >= self.window_seconds:
            _, count = self.windows.popleft()
            self.total_count -= count
    def _add_request(self, now):
        """添加请求到当前子窗口"""
        # 找到当前子窗口的起始时间
        current_window_start = now - (now % self.sub_window_size)
        # 如果当前子窗口不存在或已过期，创建新的
        if not self.windows or self.windows[-1][0] < current_window_start:
            self.windows.append((current_window_start, 1))
        else:
            # 更新当前子窗口的计数
            window_start, count = self.windows[-1]
            self.windows[-1] = (window_start, count + 1)
        self.total_count += 1
    def get_stats(self):
        """获取统计信息"""
        with self.lock:
            now = time.time()
            self._cleanup(now)
            return {
                'total_requests': self.total_count,
                'window_size': self.window_seconds,
                'sub_windows': [(time.strftime('%H:%M:%S', time.localtime(ws)), c) 
                               for ws, c in self.windows]
            }
# 使用示例
limiter = SlidingWindowCounterRateLimiter(max_requests=100, window_seconds=60, sub_window_count=6)

Redis实现（分布式场景）

import redis
import time
import json
class RedisSlidingWindowRateLimiter:
    """使用Redis实现分布式滑动窗口限流"""
    def __init__(self, redis_client, key_prefix='ratelimit', max_requests=100, window_seconds=60):
        self.redis = redis_client
        self.key_prefix = key_prefix
        self.max_requests = max_requests
        self.window_seconds = window_seconds
    def allow_request(self, user_id):
        """
        使用Redis ZSET实现滑动窗口
        每个元素为请求时间戳，score为时间戳
        """
        key = f"{self.key_prefix}:{user_id}"
        now = time.time()
        # 使用pipeline保证原子性
        pipe = self.redis.pipeline()
        # 添加当前请求
        pipe.zadd(key, {str(now): now})
        # 移除窗口外的旧请求
        pipe.zremrangebyscore(key, 0, now - self.window_seconds)
        # 获取当前窗口内请求数
        pipe.zcard(key)
        # 设置过期时间
        pipe.expire(key, self.window_seconds)
        _, _, request_count, _ = pipe.execute()
        if request_count <= self.max_requests:
            return True
        return False
    def get_remaining(self, user_id):
        """获取剩余可用请求数"""
        key = f"{self.key_prefix}:{user_id}"
        now = time.time()
        # 清理过期数据
        self.redis.zremrangebyscore(key, 0, now - self.window_seconds)
        current_count = self.redis.zcard(key)
        return max(0, self.max_requests - current_count)
# 分布式使用示例
redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)
limiter = RedisSlidingWindowRateLimiter(
    redis_client, 
    max_requests=100, 
    window_seconds=60
)

完整实战案例：API网关限流

from fastapi import FastAPI, HTTPException, Request
from typing import Optional
import time
import threading
from collections import defaultdict, deque
from datetime import datetime
import asyncio
class APIRateLimiter:
    """完整API限流器，支持多用户、多接口"""
    def __init__(self):
        self.limits = {
            'default': {'max_requests': 100, 'window': 60},
            'user': {'max_requests': 10, 'window': 60},
            'path': {'max_requests': 30, 'window': 60},
            'admin': {'max_requests': 1000, 'window': 60},
        }
        self.user_records = defaultdict(deque)
        self.path_records = defaultdict(deque)
        self.ip_records = defaultdict(deque)
        self.lock = threading.Lock()
        self.blocked_ips = {}
    def allow_request(self, user_id: Optional[str] = None, 
                     path: str = None, ip: str = None) -> dict:
        """
        检查请求是否允许
        返回：{'allowed': bool, 'retry_after': seconds, 'remaining': int}
        """
        now = time.time()
        with self.lock:
            # 检查IP是否被临时封禁
            if ip in self.blocked_ips:
                block_until = self.blocked_ips[ip]
                if now < block_until:
                    return {
                        'allowed': False,
                        'retry_after': int(block_until - now),
                        'remaining': 0,
                        'reason': 'IP暂时被封禁'
                    }
                else:
                    del self.blocked_ips[ip]
            # 检查各类限制
            checks = []
            # 用户限制
            if user_id:
                check = self._check_limit(
                    self.user_records[user_id],
                    self.limits['user']['max_requests'],
                    self.limits['user']['window'],
                    now
                )
                checks.append(check)
            # 路径限制
            if path:
                check = self._check_limit(
                    self.path_records[path],
                    self.limits['path']['max_requests'],
                    self.limits['path']['window'],
                    now
                )
                checks.append(check)
            # IP限制
            if ip:
                check = self._check_limit(
                    self.ip_records[ip],
                    self.limits['default']['max_requests'],
                    self.limits['default']['window'],
                    now
                )
                checks.append(check)
            # 汇总结果
            if checks and all(c['allowed'] for c in checks):
                return {
                    'allowed': True,
                    'remaining': min(c['remaining'] for c in checks),
                    'retry_after': 0
                }
            else:
                # 找到最长的等待时间
                retry_after = max(c['retry_after'] for c in checks if c['retry_after'] > 0)
                # 连续失败可能触发临时封禁
                if checks and any(c['remaining'] == 0 for c in checks) and ip:
                    self._block_ip(ip, now)
                return {
                    'allowed': False,
                    'retry_after': retry_after,
                    'remaining': 0,
                    'reason': '达到限流阈值'
                }
    def _check_limit(self, records: deque, max_requests: int, 
                     window: int, now: float) -> dict:
        """检查单个限流规则"""
        # 清理过期记录
        while records and now - records[0] >= window:
            records.popleft()
        remaining = max_requests - len(records)
        if remaining > 0:
            records.append(now)
            return {'allowed': True, 'remaining': remaining, 'retry_after': 0}
        else:
            oldest = records[0] if records else now
            retry_after = max(0, window - (now - oldest))
            return {'allowed': False, 'remaining': 0, 'retry_after': retry_after}
    def _block_ip(self, ip: str, now: float, duration: int = 300):
        """临时封禁IP"""
        self.blocked_ips[ip] = now + duration
# FastAPI集成示例
app = FastAPI()
limiter = APIRateLimiter()
@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    """全局限流中间件"""
    user_id = request.headers.get('X-User-ID')
    ip = request.client.host
    path = request.url.path
    result = limiter.allow_request(
        user_id=user_id,
        path=path,
        ip=ip
    )
    if not result['allowed']:
        raise HTTPException(
            status_code=429,
            detail=f"请求过于频繁，请在 {result['retry_after']} 秒后重试",
            headers={
                'Retry-After': str(result['retry_after']),
                'X-RateLimit-Remaining': str(result['remaining'])
            }
        )
    response = await call_next(request)
    # 添加限流响应头
    response.headers['X-RateLimit-Remaining'] = str(result['remaining'])
    return response
# 示例接口
@app.get("/api/data")
async def get_data(request: Request):
    return {"message": "成功获取数据"}
# 测试代码
def test_rate_limiter():
    """测试限流器"""
    limiter = APIRateLimiter()
    # 测试用户限流
    print("=== 用户限流测试 ===")
    for i in range(15):
        result = limiter.allow_request(user_id="user1", path="/api/data", ip="192.168.1.1")
        if result['allowed']:
            print(f"请求 {i+1}: ✓ 允许 (剩余: {result['remaining']})")
        else:
            print(f"请求 {i+1}: ✗ 拒绝 (等待: {result['retry_after']}秒, {result.get('reason', '')})")
    # 模拟等待
    print("\n=== 等待5秒后 ===")
    time.sleep(5)
    # 测试管理员
    print("\n=== 管理员限流测试 ===")
    for i in range(5):
        result = limiter.allow_request(user_id="admin", path="/api/data", ip="192.168.1.2")
        print(f"请求 {i+1}: {'✓ 允许' if result['allowed'] else '✗ 拒绝'}")
if __name__ == "__main__":
    test_rate_limiter()

高级优化版本：自适应限流

import threading
import time
from collections import deque
from typing import Dict, Tuple
class AdaptiveSlidingWindowLimiter:
    """自适应滑动窗口限流器"""
    def __init__(self, max_requests, window_seconds, 
                 threshold_high=0.8, threshold_low=0.3):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.threshold_high = threshold_high  # 触发扩缩容的高阈值
        self.threshold_low = threshold_low    # 触发扩容的低阈值
        self.windows: Dict[int, int] = {}  # {子窗口起始时间: 请求数}
        self.total_count = 0
        self.lock = threading.Lock()
        # 动态调整参数
        self.current_max = max_requests
        self.adjustment_counter = 0
    def allow_request(self) -> bool:
        with self.lock:
            now = time.time()
            self._cleanup(now)
            if self.total_count < self.current_max:
                self._add_request(now)
                self._adjust_capacity(now)
                return True
            return False
    def _cleanup(self, now: float):
        """清理过期的子窗口"""
        expire_threshold = now - self.window_seconds
        for start_time in list(self.windows.keys()):
            if start_time < expire_threshold:
                self.total_count -= self.windows.pop(start_time)
    def _add_request(self, now: float):
        """添加请求"""
        window_start = int(now // self.window_seconds)
        self.windows[window_start] = self.windows.get(window_start, 0) + 1
        self.total_count += 1
    def _adjust_capacity(self, now: float):
        """根据负载动态调整容量"""
        self.adjustment_counter += 1
        if self.adjustment_counter % 10 != 0:  # 每10个请求评估一次
            return
        usage_ratio = self.total_count / self.current_max
        if usage_ratio > self.threshold_high:
            # 高负载，降低限制
            self.current_max = max(1, int(self.current_max * 0.8))
        elif usage_ratio < self.threshold_low:
            # 低负载，提高限制
            self.current_max = min(self.max_requests * 2, 
                                  int(self.current_max * 1.2))
    def get_stats(self) -> Dict:
        """获取统计信息"""
        with self.lock:
            return {
                'current_max': self.current_max,
                'total_count': self.total_count,
                'usage_percentage': (self.total_count / self.current_max * 100) 
                                    if self.current_max else 0,
                'windows': len(self.windows)
            }
# 测试自适应限流
def test_adaptive_limiter():
    limiter = AdaptiveSlidingWindowLimiter(
        max_requests=100,
        window_seconds=60,
        threshold_high=0.8,
        threshold_low=0.3
    )
    print("=== 自适应限流测试 ===")
    for i in range(200):
        allowed = limiter.allow_request()
        if i % 20 == 0:
            stats = limiter.get_stats()
            print(f"请求 {i:3d}: {'✓' if allowed else '✗'} | 当前容量: {stats['current_max']:4d} | "
                  f"已使用: {stats['total_count']:3d} | 使用率: {stats['usage_percentage']:.1f}%")
if __name__ == "__main__":
    test_adaptive_limiter()

固定窗口问题：存在边界突发，可以通过滑动窗口解决
日志方式精确：记录每个请求的时间戳，但内存消耗大
计数器方式高效：通过子窗口聚合，权衡内存和精度
Redis方案适合分布式：保证一致性，支持多实例部署
实际应用建议：
- 使用子窗口（如10个小窗口）平衡精度和性能
- 设置合理的缓存过期时间
- 支持动态调整参数
- 记录限流日志用于分析

这些案例覆盖了从简单到复杂的场景,你可以根据实际需求选择适合的实现方式。

本文地址： https://www.bmjup.com/post/38471.html