如何为Python类实现按需触发、分发后执行的事件机制？

先明确核心逻辑：discord.py的@client.event本质是订阅-发布模式——先把处理函数“订阅”到对应事件上，当事件触发时，框架自动调用所有订阅该事件的函数并传递数据。

下面是实现自己的事件系统的步骤和可运行代码：

核心实现思路 #

1. 维护一个事件注册表：用字典存储事件名称和对应的处理函数列表
2. 用装饰器简化事件注册：让用户能像discord.py那样用@bot.event()标记处理函数
3. 实现事件触发方法：当事件发生时，遍历对应注册表的函数并调用，同时传递数据

可运行的示例代码 #

import time
import threading

class Bot:
    def __init__(self):
        # 事件注册表：键为事件名称，值为该事件的处理函数列表
        self._event_handlers = {}
        # 启动事件循环（用线程避免阻塞主线程）
        self._start_event_loop()

    def event(self, event_name=None):
        """事件注册装饰器"""
        def decorator(func):
            # 若未指定事件名，默认使用函数名作为事件标识
            target_event = event_name or func.__name__
            # 初始化该事件的处理函数列表（如果不存在）
            if target_event not in self._event_handlers:
                self._event_handlers[target_event] = []
            # 将当前函数添加到对应事件的处理列表中
            self._event_handlers[target_event].append(func)
            return func
        return decorator

    def dispatch(self, event_name, *args, **kwargs):
        """触发指定事件，调用所有订阅该事件的处理函数"""
        # 检查该事件是否有注册的处理函数
        if event_name in self._event_handlers:
            # 遍历并调用所有处理函数，传递参数
            for handler in self._event_handlers[event_name]:
                handler(*args, **kwargs)

    def _start_event_loop(self):
        """模拟事件循环：每秒触发一次on_time_change事件"""
        def loop():
            while True:
                current_timestamp = time.time()
                # 触发on_time_change事件，传递当前时间戳
                self.dispatch("on_time_change", current_timestamp)
                time.sleep(1)
        
        # 以守护线程启动循环，避免程序无法正常退出
        threading.Thread(target=loop, daemon=True).start()

# 创建Bot实例
bot = Bot()

# 注册on_time_change事件的处理函数
@bot.event()
def on_time_change(ctx):
    print(f"收到时间更新：{ctx}")

代码说明 #

• self._event_handlers：核心存储结构，负责记录所有事件和对应的处理函数
• event装饰器：让用户可以通过简洁的语法注册事件，无需手动维护函数列表
• dispatch方法：事件触发的入口，负责把数据分发给所有订阅了该事件的处理函数
• 事件循环用线程启动：避免阻塞主线程的事件注册操作，保证程序能正常运行

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