Python脚本如何限流高频同步任务频次

wen python案例 2026-06-25 31

本文目录导读：

使用 time.sleep() 简单限流
使用 time.perf_counter() 精确限流
令牌桶算法（Token Bucket）
漏桶算法（Leaky Bucket）
使用第三方库 ratelimit
对于循环任务的实用方案
推荐方案

在Python中限制高频同步任务的频次，常见的方法有以下几种,适用于不同的场景：

使用 `time.sleep()` 简单限流

适合对执行间隔要求不严格的场景：

import time
def rate_limited_task(interval=1.0):  # 间隔1秒
    while True:
        # 执行你的任务
        print(f"执行任务: {time.time()}")
        time.sleep(interval)

使用 `time.perf_counter()` 精确限流

控制每秒执行次数（QPS）：

import time
def rate_limiter(target_qps=5):
    """限制每秒执行次数"""
    interval = 1.0 / target_qps
    last_time = 0
    def decorator(func):
        def wrapper(*args, **kwargs):
            nonlocal last_time
            current_time = time.perf_counter()
            elapsed = current_time - last_time
            if elapsed < interval:
                time.sleep(interval - elapsed)
            last_time = time.perf_counter()
            return func(*args, **kwargs)
        return wrapper
    return decorator
@rate_limiter(target_qps=5)
def my_task():
    print(f"执行任务: {time.time()}")
# 使用
for _ in range(10):
    my_task()

令牌桶算法（Token Bucket）

适合需要容忍突发请求的场景：

import time
import threading
class TokenBucket:
    def __init__(self, rate, capacity):
        """
        rate: 令牌生成速率（每秒）
        capacity: 桶容量（最大突发量）
        """
        self.rate = rate
        self.capacity = capacity
        self.tokens = capacity
        self.last_refill_time = time.time()
        self.lock = threading.Lock()
    def consume(self, tokens=1):
        """尝试消耗令牌，返回是否成功"""
        with self.lock:
            self._refill()
            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False
    def _refill(self):
        """补充令牌"""
        now = time.time()
        elapsed = now - self.last_refill_time
        self.tokens = min(self.capacity, 
                         self.tokens + elapsed * self.rate)
        self.last_refill_time = now
    def wait_and_consume(self, tokens=1):
        """阻塞直到获取到令牌"""
        while not self.consume(tokens):
            time.sleep(0.001)  # 短暂等待
# 使用示例
bucket = TokenBucket(rate=5, capacity=10)  # 每秒5个，最大突发10个
def rate_limited_task():
    bucket.wait_and_consume()
    print(f"执行任务: {time.time()}")
for _ in range(20):
    rate_limited_task()

漏桶算法（Leaky Bucket）

适合需要平滑输出速率的场景：

import time
from collections import deque
class LeakyBucket:
    def __init__(self, rate, capacity):
        """
        rate: 处理速率（每秒）
        capacity: 队列容量
        """
        self.rate = rate
        self.capacity = capacity
        self.queue = deque()
        self.last_process_time = time.time()
    def put(self, item):
        """尝试放入队列，返回是否成功"""
        if len(self.queue) < self.capacity:
            self.queue.append((item, time.time()))
            return True
        return False
    def get(self):
        """获取并处理任务"""
        if not self.queue:
            return None
        # 检查是否可以处理
        now = time.time()
        elapsed = now - self.last_process_time
        if elapsed >= 1.0 / self.rate:
            self.last_process_time = now
            return self.queue.popleft()[0]
        return None
# 使用示例
bucket = LeakyBucket(rate=2, capacity=5)  # 每秒处理2个，队列容量5
def process_task():
    task = bucket.get()
    if task:
        print(f"处理任务 {task} at {time.time()}")
# 模拟任务提交
for i in range(10):
    bucket.put(f"task_{i}")
# 处理任务
for _ in range(10):
    process_task()
    time.sleep(0.3)  # 模拟处理间隔

使用第三方库 `ratelimit`

需要先安装：pip install ratelimit

from ratelimit import limits, sleep_and_retry
import time
@sleep_and_retry
@limits(calls=5, period=1)  # 每秒最多5次调用
def my_task():
    print(f"执行任务: {time.time()}")
# 使用
for _ in range(15):
    my_task()

对于循环任务的实用方案

混合使用时间检查和睡眠：

import time
class SimpleRateLimiter:
    def __init__(self, max_calls, period=1.0):
        """
        max_calls: 周期内最大调用次数
        period: 时间周期（秒）
        """
        self.max_calls = max_calls
        self.period = period
        self.call_times = []
    def acquire(self):
        """阻塞直到可以执行"""
        now = time.time()
        # 移除过期的记录
        self.call_times = [t for t in self.call_times 
                          if now - t < self.period]
        if len(self.call_times) >= self.max_calls:
            # 计算需要等待的时间
            wait_time = self.call_times[0] + self.period - now
            if wait_time > 0:
                time.sleep(wait_time)
        self.call_times.append(time.time())
# 使用示例
limiter = SimpleRateLimiter(max_calls=10, period=1)
for _ in range(30):
    limiter.acquire()
    print(f"执行任务: {time.time()}")