扩展 Python 事件机制：支持等待事件消失

扩展 Python 事件机制支持等待事件消失标签#Python #并发编程 #事件机制 #异步编程日期2026-05-19摘要Python 的threading.Event只能等待事件发生但人类的自然认知中事件应该有发生和消失两种状态。本文介绍如何扩展事件对象让它同时支持等待事件发生和等待事件消失用起来更符合直觉。前言在 Python 中threading.Event是常用的线程同步机制。但用它的时候我总感觉哪里不对劲。后来我意识到问题所在事件有发生和消失两种状态但我们只能等待发生不能等待事件消失。今天就来分享我的解决方案——扩展的事件对象。一、传统事件机制的痛点 threading.Event 的用法fromthreadingimportEvent eventEvent()# 线程A: 设置事件event.set()# 事件发生# 线程B: 等待事件event.wait()# 等待事件发生 使用中的别扭感假设一个场景等待下载完成然后下载完成后再等待下载开始。用传统的方式你会发现很难表达# 问题下载完成后我想等待下载开始这个事件再次触发# 但 set() 之后wait() 就永远返回了除非调用 clear()eventEvent()event.set()# 下载完成# 这里我们想等待下一次下载开始# 但 wait() 立即返回因为事件已经是发生状态event.wait()# ❌ 立即返回不是我想要的核心问题传统 Event 只有两种状态但现实中的事件有生命周期发生 → 持续 → 消失我们天然地认为事件有两种等待方式等它来或等它走。二、扩展事件对象的设计 设计目标支持两种等待wait_appear(): 等待事件发生wait_disappear(): 等待事件消失 实现代码importthreadingclassMyEvent:多线程事件支持等待事件发生与等待事件消失。def__init__(self):# 内部使用两个 Event避免轮询self._occurredthreading.Event()# 事件发生时为 Trueself._disappearedthreading.Event()# 事件消失时为 Trueself._disappeared.set()# 初始状态事件未发生消失状态成立self._lockthreading.Lock()deffire(self):触发事件设置事件发生。withself._lock:self._occurred.set()self._disappeared.clear()defclear(self):清除事件让事件变为未发生。withself._lock:self._occurred.clear()self._disappeared.set()defwait_occur(self,timeoutNone)-bool:等待事件发生直到被 fire() 或超时。 返回 True 表示事件发生False 表示超时。 returnself._occurred.wait(timeout)defwait_disappear(self,timeoutNone)-bool:等待事件消失直到被 clear() 或超时。 返回 True 表示事件已消失False 表示超时。 returnself._disappeared.wait(timeout)defis_set(self)-bool:查询当前事件是否处于发生状态。returnself._occurred.is_set()三、使用示例importtimeimportthreadingdefworker(evt):print(子线程等待事件发生...)ifevt.wait_occur(5):print(子线程检测到事件发生开始工作)else:print(子线程超时事件未发生)print(子线程等待事件消失...)ifevt.wait_disappear(5):print(子线程事件已消失后续处理)else:print(子线程超时事件仍未消失)evtMyEvent()tthreading.Thread(targetworker,args(evt,))t.start()time.sleep(1)evt.fire()# 触发事件time.sleep(1)evt.clear()# 清除事件t.join()四、与传统 Event 的对比功能threading.EventEvent2等待事件发生✅wait()✅wait_occur()等待事件消失❌ 不支持✅wait_disappear()链式等待❌ 不支持✅ 可以支持等待事件发生之后再消失后续还可以加入回调机制六、总结 本文要点概念说明传统 Event 的局限只能等待发生无法等待消失扩展的 Event2同时支持wait_occur()和wait_disappear()可以支持链式等待wait_appear_then_disappear()/wait_disappear_then_appear()更符合直觉事件有生命周期发生 → 持续 → 消失 参考资料Python threading.Event 官方文档AsyncIO Event 文档设计模式观察者模式本文为本人原创首发于掘金。如果你有任何问题或想法欢迎在评论区交流