我们成功创建了一个单例类 Logger，并为所有与该类交互的用户提供了一个静态方法 getLogger。
我们的 Logger 类包含以下关键特性：
🔹 不允许直接实例化 Logger 类。
🔹 实现了自定义错误处理，以防止直接实例化。
🔹 所有类的用户使用一个公共的静态 getLogger 方法来获取实例。
然而，目前的实现并不适合生产使用，因为它没有考虑多线程的场景——即，它不是线程安全的。这正是我们在本文中要解决的问题。
前提条件
threading 与 multiprocessing 在 Python 中的比较
🧵 Python 中的线程
这是什么？
线程允许你在一个进程内运行多个线程（小型工作者）🧠。它们共享相同的内存，就像室友共享一所房子🏠。
💡 工作原理：🔹 所有线程共享相同的内存空间 🧠
🔹 适用于 I/O 密集型任务 📡（例如，网络请求、文件读写）。
🔹 由 threading 模块控制 🧵
🧠 关键概念：
🔹 线程 = 轻量级 🚴♂️
🔹 内存在线程之间共享 🏠
🔹 但是……有一个讨厌的东西叫做 GIL（全局解释器锁）⛔，这意味着同一时间只能有一个线程运行 Python 代码，因此没有真正的并行性 😢（对于 CPU 密集型任务）。
✅ 优点：
🔹 启动速度超级快 ⚡
🔹 内存使用低 🪶
🔹 线程可以轻松共享数据 🧠
❌ 缺点：
🔹 由于 GIL 的原因，无法利用多个 CPU 核心 🚫🧠
🔹 存在竞争条件和死锁等错误的风险 🕳️
🔥 Python 中的多进程
什么是多进程？
多进程创建独立的进程，每个进程都有自己的内存，它们真正并行运行 🚀——就像不同的计算机一起工作 🤝
💡 工作原理：
🔹 使用 multiprocessing 模块 🛠️。
🔹 每个进程都有自己的内存空间 📦。
🔹 这里没有 GIL——每个进程都获得一个核心！ 🧠🧠🧠
🧠 最适合：
🔹 CPU 密集型任务 🧮 – 数值计算、数据处理、机器学习等。
🔹 当你需要真正的并行性 ⚙️⚙️
✅ 优点
🔹 真正的并行性 💥
🔹 非常适合 CPU 密集型任务 🧠💪
🔹 没有 GIL = 没问题 🚫
❌ 缺点
🔹 更高的内存使用 🧠💸
🔹 启动速度较慢 ⚙️
🔹 进程间共享数据更复杂 🧵➡️📦
我们会使用哪一个？
考虑到我们在 Python 中可用的不同选项，当然我们会选择 threading 模块，因为这是一个 I/O 用例，并且我们在这里并没有进行任何重计算。
为什么它不是线程安全的？
参照上面的图示。想象一个场景，其中有两个线程 Thread 1 和 Thread 2 都试图在应用程序运行的初始阶段访问 getLogger，即 __loggerInstance 为 None
🔹 线程 1 检查 cls.__loggerInstance 是否为 None — 是的，所以它继续执行。
🔹 线程 2 同时运行，检查 cls.__loggerInstance — 仍然是 None，因此它也继续执行。
两个线程都调用 __new__() 和 __init__() → 🧨 砰！创建了两个实例！
这被称为 竞争条件，在多线程环境中确实可能发生。
🧪 如何重现该问题
您可以人为地减慢实例化过程以引发该问题：为了重现该问题，我们将对现有代码库进行以下更改
📄 Logger.py
import time
class Logger:
# 私有静态变量，用于跟踪创建的实例数量
__numInstances = 0
# 私有静态变量，用于标识实例是否已创建
__loggerInstance = None
def __new__(cls):
# 私有构造函数，使用 __new__ 防止直接实例化
raise Exception("请使用 getLogger() 创建实例。")
def __init__(self):
Logger.__numInstances = Logger.__numInstances + 1
print("日志记录器实例化，总实例数量 - ", Logger.__numInstances)
def log(self, message: str):
print(message)
@classmethod
def getLogger(cls):
# 返回单例实例，如果不存在则创建一个
if cls.__loggerInstance is None:
time.sleep(0.1) # 模拟延迟
# 绕过 __new__ 直接实例化类
cls.__loggerInstance = super(Logger, cls).__new__(cls)
# 在首次创建时手动触发 __init__
cls.__loggerInstance.__init__()
return cls.__loggerInstance
📄 main.py
from user1 import doProcessingUser1
from user2 import doProcessingUser2
import threading
import multiprocessing
if __name__ == "__main__":
t1 = threading.Thread(target=doProcessingUser1)
t2 = threading.Thread(target=doProcessingUser2)
t1.start()
t2.start()
t1.join()
t2.join()
print("所有线程已完成")
添加了 time.sleep(0.1) 来模拟执行 getLogger 函数时的延迟。
输出（我们遇到了竞争条件）
日志记录器已实例化，总实例数 - 1
来自第二个用户的日志
日志记录器已实例化，总实例数 - 2
来自第一个用户的日志
所有线程已完成
✅ 如何使其线程安全
使用 threading.Lock 来同步对单例逻辑的访问。让我们进行这些更改，然后讨论到目前为止我们所编辑的内容。
📄 logger.py
import time
import threading
class Logger:
# 私有静态变量，用于跟踪创建的实例数量
__numInstances = 0
# 私有静态变量，用于指示实例是否已创建
__loggerInstance = None
__mutexLock = threading.Lock() # 用于线程安全的单例创建（私有静态）
def __new__(cls):
# 私有构造函数，使用 __new__ 防止直接实例化
raise Exception("请使用 getLogger() 创建实例。")
def __init__(self):
Logger.__numInstances = Logger.__numInstances + 1
print("Logger 实例化，总实例数量 - ", Logger.__numInstances)
def log(self, message: str):
print(message)
@classmethod
def getLogger(cls):
# 返回单例实例，如果不存在则创建一个
with cls.__mutexLock:
if cls.__loggerInstance is None:
time.sleep(0.1) # 模拟延迟
# 跳过 __new__ 并直接实例化类
cls.__loggerInstance = super(Logger, cls).__new__(cls)
# 在首次创建时手动触发 __init__
cls.__loggerInstance.__init__()
return cls.__loggerInstanc
在这里，我们添加了以下内容
🔹 私有静态变量 __mutexLock，用于让只有一个线程，即 Thread 1 或 Thread 2 访问 getLogger 函数。
🔹 with cls.__mutexLock:
👉 这有什么作用：
with 语句在块开始时自动获取锁 ✅
然后，一旦退出块，它会自动释放锁（即使发生异常也会如此！） 🔓
因此，从技术上讲，锁在 with 下的缩进块完成后立即被释放。
✅ 所以不需要手动调用 .acquire() 或 .release() — with 块会干净且安全地处理这些 🙌
以下是一个图示，展示了这如何帮助我们避免之前遇到的原始 race condition 问题。
🧵 线程 1 和 线程 2 执行流程
🟩 线程 1（图的左侧）：
调用 getLogger()
→ 线程 1 想要获取日志实例。
获取锁 🔐
→ 由于这是第一个进入的线程，它获取了 __mutexLock 的锁。
执行 getLogger()
→ 发现 __loggerInstance 为 None，创建单例日志实例。
释放锁 🔓
→ 完成临界区，允许其他线程进入。
🟦线程 2（图的右侧）：
也在（大致）同一时间调用 getLogger()
等待锁 ⏳
→ 它遇到了锁，但线程 1 已经持有锁。
仍在等待… 😬
→ 线程 1 正在执行它的操作。线程 2 在这里静静等待。
获取锁（在线程 1 释放后） ✅
→ 现在线程 2 进入了临界区。
执行 getLogger()
→ 但这次，它发现 __loggerInstance 不是 None，所以它直接返回现有的日志记录器！
✅ 这如何防止竞争条件：
没有锁：
两个线程可能同时检查 __loggerInstance 是否为 None，并且两个线程可能会同时尝试创建它 → ❌ 多个实例（竞争条件！）。
有锁：
每次只有一个线程进入临界区，确保只创建一个日志记录器实例。
最终输出
Logger Instantiated, Total number of instances - 1
Log from the first user
Log from the second user
All threads finished
现在，正如预期的那样，我们只看到一个类的实例被实例化，即使在多线程环境中也是如此。
最后一个优化
这个优化基于锁的使用在进程中是昂贵的这一事实。因此，我们将聪明地使用它。
请注意我们当前的 getLogger 函数的一个小细节
def getLogger(cls):
# 返回单例实例，如果不存在则创建一个
with cls.__mutexLock:
if cls.__loggerInstance is None:
time.sleep(0.1) # 模拟延迟
# 绕过 __new__ 并直接实例化类
cls.__loggerInstance = super(Logger, cls).__new__(cls)
# 在首次创建时手动触发 __init__
cls.__loggerInstance.__init__()
return cls.__loggerInstance
在当前的实现中，我们在每次调用 getLogger 函数时都获取一个锁，这样效率很低。
然而，锁只在程序开始时是必要的。一旦 __loggerInstance 被设置，任何后续访问 getLogger 函数的线程都不会创建新的实例——它们只会接收到现有的实例。
为了优化这一点，我们将在获取锁之前添加一个小检查，以确保它仅在初始实例化期间使用。
def getLogger(cls):
# 返回单例实例，如果不存在则创建一个
if cls.__loggerInstance is None: # 🚀 快速无锁检查
with cls.__mutexLock:
if cls.__loggerInstance is None:
time.sleep(0.1) # 模拟延迟
# 跳过 __new__ 方法，直接实例化类
cls.__loggerInstance = super(Logger, cls).__new__(cls)
# 在首次创建时手动触发 __init__
cls.__loggerInstance.__init__()
return cls.__loggerInstance
我们添加了上述条件 if cls.__loggerInstance is None: 来检查我们是否处于执行的初始阶段，然后才获取锁以实例化该类。
这种类型的锁定被称为 双重检查锁定模式
🪄 为什么这样更好：
大多数情况下（在记录器创建之后），该方法将完全跳过锁定 🙌
锁定仅在第一次初始化时发生一次
仍然是100%线程安全的 ✅
如以下来自 main.py 的响应所测试的
Logger Instantiated, Total number of instances - 1
日志来自第一个用户
日志来自第二个用户
所有线程已完成
这与我们之前得到的完全相同
