如何对同一份Python脚本的多次独立执行进行速率限制？

Hey，我刚好处理过类似的需求——要让同一份Python脚本的多次独立调用遵守速率限制，核心难点在于这些脚本实例是完全独立的进程，没法共享内存里的状态，所以得找一个所有实例都能访问的共享存储来跟踪执行记录。下面给你一步步拆解解决方案：

需求回顾 #

你要实现的是：

• 测试阶段限制为每分钟3次执行，最终目标是每秒20次
• 超过限制时，后续执行要排队等待，遵循FIFO规则
• 同一份脚本被多次独立调用时生效

你给出的示例代码（我修正了语法问题）：

import time
rate = 60
limit = 3
ts = time.time()

def single_task():
    print(ts)

rate_limit(single_task())

预期的执行效果：

session_1$ ./script.py
1527483557.76
session_2$ ./script.py
1527483558.26
session_3$ ./script.py
1527483559.03
session_4$ ./script.py
hang on a minute...
1527483560.57
session_5$ ./script.py
hang on a minute...
1527483561.92

解决方案：用SQLite做共享状态存储 #

我推荐用SQLite来实现，它不需要额外安装服务，轻量且支持多进程安全访问，刚好适合单机器上的跨实例限流需求。

完整实现代码 #

下面是可以直接运行的脚本，包含了所有限流逻辑：

import time
import sqlite3
from contextlib import contextmanager

# 限流参数：测试用每分钟3次，改成每秒20次只需把RATE_WINDOW设为1，MAX_REQUESTS设为20
RATE_WINDOW = 60  # 时间窗口（秒）
MAX_REQUESTS = 3   # 窗口内允许的最大执行次数

@contextmanager
def db_connection():
    # 连接共享的SQLite数据库，所有脚本实例都读写这个文件
    conn = sqlite3.connect('rate_limit.db')
    cursor = conn.cursor()
    # 初始化请求记录表（第一次运行时自动创建）
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS requests (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            timestamp REAL NOT NULL
        )
    ''')
    conn.commit()
    try:
        yield cursor, conn
    finally:
        conn.close()

def rate_limit():
    current_time = time.time()
    with db_connection() as (cursor, conn):
        # 先清理掉时间窗口外的旧执行记录，保证统计的是当前窗口内的次数
        cursor.execute('DELETE FROM requests WHERE timestamp < ?', (current_time - RATE_WINDOW,))
        # 查询当前窗口内已经执行的次数
        cursor.execute('SELECT COUNT(*) FROM requests')
        current_count = cursor.fetchone()[0]
        
        if current_count >= MAX_REQUESTS:
            # 计算需要等待的时间：找到最早的执行记录，等到它超出窗口为止
            cursor.execute('SELECT MIN(timestamp) FROM requests')
            earliest_exec_time = cursor.fetchone()[0]
            wait_duration = (earliest_exec_time + RATE_WINDOW) - current_time
            if wait_duration > 0:
                print("hang on a minute...")
                time.sleep(wait_duration)
            # 等待后再次清理旧记录，避免重复计数
            cursor.execute('DELETE FROM requests WHERE timestamp < ?', (time.time() - RATE_WINDOW,))
        
        # 记录当前的执行时间
        cursor.execute('INSERT INTO requests (timestamp) VALUES (?)', (time.time(),))
        conn.commit()

def single_task():
    # 这里是你实际要执行的任务，示例中打印当前时间戳
    print(time.time())

if __name__ == "__main__":
    # 先执行限流检查，通过后再执行任务
    rate_limit()
    single_task()

代码逻辑说明 #

1. 共享状态存储：所有脚本实例都会读写rate_limit.db这个SQLite文件，用来记录每一次执行的时间戳。
2. 窗口清理：每次执行前先删除超出时间窗口的旧记录，确保我们统计的是最近窗口内的执行次数。
3. 等待逻辑：如果当前窗口内执行次数已经达到上限，就计算需要等待的时间，等最早的执行记录过期后再继续。
4. 多进程安全：SQLite本身会自动处理多进程访问的锁机制，不会出现多个实例同时写入导致的计数错误。

测试效果 #

当你多次独立运行这个脚本时，前3次会立即输出时间戳，第4次开始会打印hang on a minute...并等待，直到时间窗口内的旧执行记录过期，完全符合你预期的效果。

切换到每秒20次限制 #

只需要修改两个参数：

RATE_WINDOW = 1  # 1秒的时间窗口
MAX_REQUESTS = 20  # 每秒最多允许20次执行

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