防抖触发器核心是等待静默期后执行最后一次调用,并通过生成器协调状态实现串行化:每次新触发取消前序任务,用异步生成器配合asyncio.lock确保单次执行、无并发重叠。

生成器本身不直接提供防抖(debounce)能力,但可以配合事件循环、计时器和状态管理,构建一个序列化防抖触发器——即:对连续调用进行节流合并,只在最后一次调用后延迟执行,并确保每次执行是串行、无重叠的。
防抖触发器的核心逻辑
防抖不是“丢弃中间调用”,而是“等待静默期结束再执行最后一次”。序列化则要求:即使多次触发防抖,也必须等前一次执行完成,才允许下一次执行开始,避免并发冲突。
- 每次新触发时,取消前一个待执行任务(如 pending timer 或协程)
- 用生成器维护当前执行状态(例如 yield True 表示“已占用”,yield False 表示“空闲”)
- 实际执行逻辑封装为协程或普通函数,由生成器协调调度
用生成器管理执行状态
生成器可作为轻量级状态机,通过 yield 暂停并返回控制权,记录“是否正在运行”或“是否已排队”。例如:
def debounce_serializer():
running = False
pending = None
while True:
# 接收一个 callable 和 delay
func, delay = yield
if running:
# 取消上一次 pending(需支持 cancel,如 asyncio.Task)
if pending and not pending.done():
pending.cancel()
# 延迟启动新任务
pending = asyncio.create_task(asyncio.sleep(delay))
await pending
running = True
await func() # 执行业务逻辑
running = False
注意:这只是一个示意结构。真实场景中,生成器通常不直接 await,而是交由事件循环驱动;更推荐用 async generator + asyncio.Lock 实现串行保障。
推荐实现:async generator + asyncio.Lock
结合异步生成器与锁,能自然实现“一次只执行一个防抖任务”:
- 用
async with lock:确保同一时间仅一个任务进入执行区 - 用
asyncio.wait_for(..., timeout=delay)实现防抖等待 - 把待执行函数包装进队列,由单个消费者协程按序拉取并防抖执行
关键不是生成器本身做防抖,而是它让状态流转清晰可读——比如用 yield 'waiting'、yield 'executing' 标记阶段,便于调试和监控。
为什么不用纯生成器做完整防抖?
因为生成器无法主动暂停外部调用、也无法跨调用保存 timer 引用。真正的防抖依赖:
• 可取消的延迟机制(如 asyncio.create_task(asyncio.sleep()))
• 共享状态容器(如类属性或闭包变量)
• 显式调度逻辑(如单独的 runner 协程)
生成器更适合表达“执行流程的阶段性”,而非替代事件循环的调度职责。











