如何在纯Python中实现基于指定键的事件限流？

当然可以用纯Python实现这种基于message和host字段的事件限流！我给你分享两个实用方案，分别适配单进程和分布式/多进程场景，你可以按需选择：

如果你的应用是单个Python进程在跑，完全不需要外部数据库或Redis，用内存里的字典就能搞定。核心思路是：用一个缓存记录每个(message, host)组合最近一次通过限流的时间戳，每次新事件过来时，对比当前时间和缓存里的时间差，超过你设定的间隔就允许通过，否则就拦截。

给你写个可直接用的示例类：

import time
from typing import Dict, Tuple

class LocalEventRateLimiter:
    def __init__(self, interval: float):
        # interval是限流时间间隔，单位秒
        self.interval = interval
        # 缓存键是(message, host)元组，值是上次允许通过的时间戳
        self.last_allowed: Dict[Tuple[str, str], float] = {}

    def is_allowed(self, event: dict) -> bool:
        # 提取事件的关键标识
        key = (event["message"], event["host"])
        now = time.time()
        
        # 检查缓存中是否有记录，且未超过间隔
        if key in self.last_allowed:
            time_since_last = now - self.last_allowed[key]
            if time_since_last < self.interval:
                return False
        
        # 更新缓存时间，允许通过
        self.last_allowed[key] = now
        return True

# 用法示例
limiter = LocalEventRateLimiter(interval=10)  # 10秒内同一事件只允许一次

event1 = {"id":111,"message":"hello","host":"example"}
print(limiter.is_allowed(event1))  # 输出True，第一次通过

event2 = {"id":112,"message":"hello","host":"example"}
print(limiter.is_allowed(event2))  # 输出False，10秒内重复

# 10秒后再试
time.sleep(10)
print(limiter.is_allowed(event2))  # 输出True

如果你的应用是多进程部署，或者分布式系统，内存缓存就没法共享状态了，这时候Redis是最优选择——它的原子操作能完美避免竞态条件，而且性能极高。

核心思路是：把每个(message, host)组合作为Redis的键，设置过期时间等于你的限流间隔。当新事件过来时，尝试设置这个键（只有键不存在时才成功），设置成功就允许通过，失败就拦截。这样Redis会自动帮我们管理过期的键，不用手动清理。

先确保你安装了redis库：pip install redis，然后看示例代码：

import redis
from typing import Optional

class RedisEventRateLimiter:
    def __init__(self, redis_host: str = "localhost", redis_port: int = 6379, db: int = 0, interval: float = 10):
        self.redis_client = redis.Redis(host=redis_host, port=redis_port, db=db)
        self.interval = interval
        # 键的前缀，避免和其他Redis键冲突
        self.key_prefix = "event_limiter:"

    def _get_key(self, event: dict) -> str:
        # 生成唯一键：前缀 + message + 分隔符 + host
        return f"{self.key_prefix}{event['message']}:{event['host']}"

    def is_allowed(self, event: dict) -> bool:
        key = self._get_key(event)
        # 使用Redis的SET命令，NX表示只有键不存在时才设置，EX是过期时间（秒）
        result = self.redis_client.set(key, "1", nx=True, ex=int(self.interval))
        # 设置成功返回True，说明是该间隔内第一次出现；失败返回None，说明重复
        return result is not None

# 用法示例
limiter = RedisEventRateLimiter(interval=10)

event1 = {"id":111,"message":"hello","host":"example"}
print(limiter.is_allowed(event1))  # True

event2 = {"id":112,"message":"hello","host":"example"}
print(limiter.is_allowed(event2))  # False

# 10秒后键过期，再试就会返回True

• 单进程方案的缺点是：如果进程重启，缓存会清空，适合对限流精度要求不高的场景；
• Redis方案要确保所有节点的时间同步，否则过期时间判断会有偏差；
• 可以根据需求调整限流间隔，比如对不同的host设置不同的间隔，只需要修改键的生成逻辑或者缓存结构就行；
• 高并发场景下，Redis方案完全能扛住，因为SET命令是原子操作，不会出现多个进程同时通过的情况。

内容的提问来源于stack exchange，提问作者pm1391