如何实现可取消前次调用的Python函数速率限制装饰器？

要实现这种取消前次调用、重启计时、非阻塞的速率限制装饰器，核心是用后台线程或异步任务处理延迟执行，避免阻塞主程序。下面分同步和异步场景给出实现方案：

一、同步函数场景（基于threading.Timer） #

这个版本适用于普通同步函数，用后台线程处理延迟，完全不阻塞主线程：

import threading
from functools import wraps

def debounce(wait_time):
    def decorator(func):
        # 存储当前活跃的计时器
        timer = None

        @wraps(func)
        def wrapper(*args, **kwargs):
            nonlocal timer

            # 取消之前未触发的计时器
            if timer is not None:
                timer.cancel()

            # 定义延迟执行逻辑，带上原函数的参数
            def delayed_run():
                nonlocal timer
                func(*args, **kwargs)
                timer = None  # 执行完成后清空标记

            # 创建新计时器，延迟指定时间后执行
            timer = threading.Timer(wait_time, delayed_run)
            timer.start()

        return wrapper
    return decorator

使用示例： #

@debounce(2)
def print_msg(msg):
    print(f"执行函数：{msg}")

# 连续调用只会保留最后一次
print_msg("第一次调用")
print_msg("第二次调用")
print_msg("第三次调用")

# 主线程不受阻塞，可正常执行其他逻辑
print("主线程继续运行...")

运行后会先输出主线程继续运行...，2秒后输出执行函数：第三次调用，前两次调用被自动取消。

二、异步函数场景（基于asyncio） #

如果你的函数是异步定义的（async def），可以用asyncio的任务机制实现非阻塞防抖：

import asyncio
from functools import wraps

def async_debounce(wait_time):
    def decorator(func):
        # 存储当前待执行的异步任务
        pending_task = None

        @wraps(func)
        async def wrapper(*args, **kwargs):
            nonlocal pending_task

            # 取消之前未完成的异步任务
            if pending_task is not None and not pending_task.done():
                pending_task.cancel()

            # 定义异步延迟执行逻辑
            async def delayed_run():
                nonlocal pending_task
                await asyncio.sleep(wait_time)
                await func(*args, **kwargs)
                pending_task = None

            # 创建并启动新的异步任务
            pending_task = asyncio.create_task(delayed_run())

        return wrapper
    return decorator

使用示例： #

@async_debounce(2)
async def async_print(msg):
    print(f"异步执行：{msg}")

async def main():
    async_print("第一次异步调用")
    async_print("第二次异步调用")
    async_print("第三次异步调用")
    print("异步主线程继续运行...")
    await asyncio.sleep(3)  # 等待足够时间让最后一次执行完成

asyncio.run(main())

运行后同样先输出异步主线程继续运行...，2秒后输出异步执行：第三次异步调用。

关键说明 #

• 非阻塞核心：同步版本依赖后台线程执行延迟逻辑，主线程不受干扰；异步版本利用asyncio的非阻塞睡眠，事件循环可同时处理其他任务。
• 调用取消机制：每次触发函数时，都会检查并取消之前未完成的计时器/任务，确保只有最后一次调用会被执行。
• 通用性保障：通过*args和**kwargs保留原函数的参数传递，适配不同签名的函数。

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