如何在Python中实现每次事件set都执行绑定函数的机制

原有代码问题说明 #

threading.Event是Python标准库提供的线程同步原语，并非为事件分发场景设计：它的wait()方法只会阻塞到信号第一次被set()，之后不会自动重置阻塞状态。你写的监听函数仅调用了一次wait()，执行完打印逻辑后函数就直接运行结束，对应子线程随之退出，后续再set()事件自然不会触发执行。

实现方案 #

方案1：对齐JS原生事件逻辑的通用实现 #

JS自定义事件的核心能力是：事件实例支持绑定/解绑监听回调，派发事件时按绑定顺序执行所有对应类型的回调。以下是纯标准库实现，接口和JS完全对齐：

from collections import defaultdict
import threading
from typing import Callable, Any

class EventTarget:
    def __init__(self):
        self._listeners: dict[str, list[Callable]] = defaultdict(list)
        self._lock = threading.Lock()

    def addEventListener(self, event_type: str, callback: Callable, useCapture: bool = False) -> None:
        """对齐JS的addEventListener接口，useCapture参数为接口兼容保留，暂不实现事件冒泡/捕获逻辑"""
        with self._lock:
            if callback not in self._listeners[event_type]:
                self._listeners[event_type].append(callback)

    def removeEventListener(self, event_type: str, callback: Callable) -> None:
        """对齐JS的removeEventListener接口"""
        with self._lock:
            if callback in self._listeners[event_type]:
                self._listeners[event_type].remove(callback)

    def dispatchEvent(self, event_type: str, *event_args: Any, **event_kwargs: Any) -> None:
        """对齐JS的dispatchEvent接口，触发事件时将参数传给所有绑定的监听回调"""
        with self._lock:
            callbacks = self._listeners[event_type].copy()
        for cb in callbacks:
            cb(*event_args, **event_kwargs)

对应你给出的JS示例的等价Python写法：

import time

# 等价JS中实现了EventTarget接口的DOM元素elem
elem = EventTarget()
x = 0

# 事件监听回调
def myevent_handler(e=None):
    global x
    print(f"event triggered, x = {x}")
    x += 1

# 绑定事件监听，等价 elem.addEventListener('myevent', function (e) { /* ... */ }, false)
elem.addEventListener('myevent', myevent_handler, False)

# 循环派发事件，等价 elem.dispatchEvent(event)
if __name__ == "__main__":
    while True:
        elem.dispatchEvent('myevent')
        time.sleep(1)

方案2：基于threading.Event的修正版本 #

如果你需要监听逻辑跑在独立子线程，不阻塞主线程，可以给监听函数加常驻循环，每次收到信号执行完逻辑后清空事件、重新进入等待状态：

import threading
import time

event = threading.Event()
x = 0

def myFunction():
    global x
    while True:
        event.wait()
        event.clear() # 收到信号立刻清空，避免重复触发
        print("event is set : " + str(x))
        x += 1
        time.sleep(0.3)

# 设为守护线程，主线程退出时子线程自动终止
th1 = threading.Thread(target=myFunction, daemon=True)
th1.start()

while True:
    event.set()
    time.sleep(1)

注意：该threading版本仅适合简单场景，存在事件丢失风险：如果子线程还在处理上一次事件、未进入wait状态时主线程调用了set()，本次触发会被直接跳过。生产环境优先使用方案1的EventTarget实现。

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