Python 单例模式深度解析：从任务队列的多实例问题谈起

1. 问题的起源

在开发 AssociationRuleTaskQueue（关联规则任务队列）时，我们遇到了一个关于“实例唯一性”的疑问：

“我在运行 from core.task_queue import rule_task_queue 的时候，其实就会运行一次 rule_task_queue = AssociationRuleTaskQueue() 对吗？所以每一次导入都会实例化一次对吗？”

这个疑问触及了 Python 两个核心机制：模块导入机制 和 类实例化机制。本文将以此为切入点，深入解析 Python 的 __new__ 方法与单例模式（Singleton Pattern）。

2. Python 模块导入机制：天然的“单例”

首先回答你的疑问：每一次导入并不会都实例化一次。

Python 的模块（Module）对象在整个解释器进程中是全局唯一的。当你执行 import core.task_queue 或 from core.task_queue import rule_task_queue 时，Python 内部遵循以下流程：

检查缓存：查看 sys.modules 字典中是否已经存在名为 core.task_queue 的模块。
首次导入：如果不存在，Python 会加载该文件，执行文件中的所有代码（包括 rule_task_queue = AssociationRuleTaskQueue() 这行），并将生成的模块对象存入 sys.modules。
后续导入：如果已存在，直接从 sys.modules 返回之前创建的模块对象，不再执行文件代码。

结论：在模块层面上，rule_task_queue 这个变量只会被初始化一次。

那为什么还需要单例模式？

虽然模块导入保证了 rule_task_queue 这个全局变量的唯一性，但它无法阻止开发者在其他地方手动创建新实例：

# 开发者 A 在 main.py 中
from core.task_queue import rule_task_queue
# 此时使用的是全局唯一的队列

# 开发者 B 在 service.py 中
from core.task_queue import AssociationRuleTaskQueue
my_queue = AssociationRuleTaskQueue()  # <--- 危险！这里创建了一个全新的队列实例

如果没有单例模式保护，my_queue 和 rule_task_queue 是两个完全独立的内存对象。向 my_queue 添加的任务，永远不会被 rule_task_queue 的 Worker 消费。这就是多实例陷阱。

3. 深入理解 `new` 与 `init`

要解决这个问题，我们需要从类的实例化过程入手。Python 中创建一个对象实际上分两步：

__new__(cls, ...)：构造方法。这是一个静态方法，负责创建并返回实例对象（分配内存）。
__init__(self, ...)：初始化方法。这是一个实例方法，负责对已经创建好的实例进行属性赋值。

形象的比喻：

__new__ 是工厂：它负责制造一个空的“瓶子”。
__init__ 是灌装线：它负责往瓶子里装水、贴标签。

标准实例化流程

obj = Class()
# 等价于：
# 1. obj = Class.__new__(Class)
# 2. if isinstance(obj, Class):
#        Class.__init__(obj)

4. 使用 `new` 实现单例模式

单例模式的核心在于：拦截 __new__，不再每次都制造新瓶子，而是永远返回同一个旧瓶子。

我们的代码实现 (`core/task_queue.py`)

class AssociationRuleTaskQueue:
    _instance = None  # 类变量，用于存储唯一的那个实例

    def __new__(cls):
        # 1. 检查是否已经存在实例
        if cls._instance is None:
            # 2. 如果不存在，调用父类(object)的 __new__ 创建一个真·新实例
            cls._instance = super(AssociationRuleTaskQueue, cls).__new__(cls)
            
            # 3. 【关键】初始化逻辑也放在这里
            cls._instance.queue = asyncio.Queue()
            cls._instance.is_running = False
            cls._instance.agent = RuleMiningAgent()
            
        # 4. 返回这个唯一的实例
        return cls._instance

为什么把初始化逻辑放在 `new` 里？

这是一个常见的高级技巧。如果我们使用常规的 __init__：

class BadSingleton:
    _instance = None
    def __new__(cls):
        if not cls._instance:
            cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self):
        self.queue = asyncio.Queue()  # <--- 陷阱！