Python 单例日志记录器 ：线程安全、竞争条件与锁优化

[QPY课程团队]

我们成功创建了一个单例类 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

日志来自第一个用户  
日志来自第二个用户  
所有线程已完成  

这与我们之前得到的完全相同

[作者]

帖子