Python函数作用域内实现单例？事件追踪单例重置问题及异常安全方案

你的单例实现存在关键漏洞：__new__虽然保证全局只有一个实例，但每次调用Events()时，Python都会自动调用该实例的__init__方法。而你的__init__每次执行都会重置self.events = []，所以跨模块调用register_event时，每次创建events = Events()都会触发__init__，把事件列表清空，导致计数始终从0开始。

要解决这个问题，需确保__init__只在实例第一次创建时执行，后续调用不会重置属性，以下是几种可行方案：

方案1：在__new__中初始化实例属性 #

把原本在__init__里的属性初始化逻辑移到__new__中，仅在第一次创建实例时执行：

class Events:
    def __new__(cls):
        if not hasattr(cls, "instance"):
            cls.instance = super(Events, cls).__new__(cls)
            # 仅初始化一次属性
            cls.instance.events = []
            cls.instance.max_events = 25
            cls.instance.rate_limit = 60.0
        return cls.instance

    def __call__(self, event):
        print(len(self.events))
        self.events.append(event)
        # ... 其他业务逻辑

方案2：在__init__中添加初始化判断 #

给实例加一个标记，判断是否已完成初始化，避免重复执行重置逻辑：

class Events:
    def __new__(cls):
        if not hasattr(cls, "instance"):
            cls.instance = super(Events, cls).__new__(cls)
            cls.instance._initialized = False
        return cls.instance

    def __init__(self):
        if self._initialized:
            return
        self.events = []
        self.max_events = 25
        self.rate_limit = 60.0
        self._initialized = True

    def __call__(self, event):
        print(len(self.events))
        self.events.append(event)
        # ... 其他业务逻辑

方案3：利用Python模块的天然单例特性 #

Python模块本身就是单例，你可以直接把事件追踪逻辑放在模块里，无需编写类：

# events.py
events = []
max_events = 25
rate_limit = 60.0

def register_event(event):
    print(len(events))
    events.append(event)
    # ... 其他业务逻辑

跨模块使用时直接导入：

from events import register_event
register_event("event_name")

如果需要在函数作用域内保证单例，可以通过闭包或装饰器实现：

闭包方式 #

def get_events_singleton():
    instance = None
    def inner():
        nonlocal instance
        if instance is None:
            instance = Events()  # 使用修复后的Events类
        return instance
    return inner

# 获取单例获取函数
get_events = get_events_singleton()

# 在注册函数中使用
@TryExcept(verbose=True)
def register_event(name: str, **kwargs):
    events = get_events()
    events(name, kwargs)

装饰器方式实现单例 #

def singleton(cls):
    instances = {}
    def wrapper(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]
    return wrapper

@singleton
class Events:
    def __init__(self):
        self.events = []
        self.max_events = 25
        self.rate_limit = 60.0

    def __call__(self, event):
        print(len(self.events))
        self.events.append(event)
        # ... 其他业务逻辑

这样每次调用Events()都会返回同一个实例，且__init__仅执行一次。

异常安全需要保证：注册事件过程中发生异常时，不丢失已有事件，同时能正确捕获并处理异常。

1. 完善异常装饰器 #

确保TryExcept装饰器仅捕获函数内部异常，不干扰单例状态，同时可添加异常兜底逻辑：

def TryExcept(verbose=True):
    def decorator(func):
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                if verbose:
                    print(f"注册事件失败: {str(e)}")
                # 可选：将失败事件暂存到持久化存储，避免丢失
                # save_failed_event(args, kwargs, e)
        return wrapper
    return decorator

2. 保证事件操作的原子性 #

在多模块/多线程场景下，给事件操作加锁，确保修改events列表的逻辑是原子的：

import threading

class Events:
    def __new__(cls):
        if not hasattr(cls, "instance"):
            cls.instance = super(Events, cls).__new__(cls)
            cls.instance.events = []
            cls.instance.max_events = 25
            cls.instance.rate_limit = 60.0
            cls.instance._lock = threading.Lock()
        return cls.instance

    def __call__(self, event):
        with self._lock:
            print(len(self.events))
            self.events.append(event)
            # 检查速率限制、发送事件等逻辑也放在锁内，保证原子性
            # ... 其他业务逻辑

3. 异常时的事件兜底存储 #

如果注册事件发生异常，可将事件暂存到文件或数据库，后续再重试发送：

import json
import os

FAILED_EVENTS_FILE = "failed_events.json"

@TryExcept(verbose=True)
def register_event(name: str, **kwargs):
    events = Events()
    try:
        events(name, kwargs)
    except Exception as e:
        # 保存失败事件到文件
        failed_event = {"name": name, "kwargs": kwargs, "error": str(e)}
        with open(FAILED_EVENTS_FILE, "a") as f:
            json.dump(failed_event, f)
            f.write("\n")
        # 按需决定是否重新抛出异常
        raise

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