在Python并发编程的实际开发中，多个执行单元同时操作日志文件是常见需求，但默认的日志写入方式没有做并发控制，很容易出现日志内容交叉、部分日志丢失的情况。上下文管理器可以通过定义资源的获取和释放逻辑，配合锁机制实现安全的并发日志处理。
上下文管理器的基本原理
上下文管理器是Python中用于管理资源生命周期的对象，通过__enter__和__exit__两个魔术方法实现。当使用with语句调用上下文管理器时，会先执行__enter__方法获取资源，在with代码块执行完成后，无论是否出现异常，都会执行__exit__方法释放资源。
下面是一个最基础的上下文管理器实现示例：
class SimpleContextManager:
def __enter__(self):
print("进入上下文，获取资源")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
print("退出上下文，释放资源")
# 如果返回True，会忽略with代码块中的异常
return False
# 使用方式
with SimpleContextManager() as cm:
print("执行核心逻辑")
并发日志处理的常见问题
在没有并发控制的情况下，多个线程同时调用logging模块写入同一个日志文件时，会出现以下问题：
- 日志内容交叉：两条不同线程的日志内容被拼接到同一行，无法区分来源
- 日志丢失：后写入的日志覆盖先写入的内容，或者部分内容没有被刷入文件
- 文件句柄异常：多个线程同时操作文件句柄，导致文件写入报错
基于上下文管理器的并发日志处理实现
我们可以自定义一个日志上下文管理器，在__enter__阶段获取文件锁，在__exit__阶段释放锁，保证同一时间只有一个线程可以写入日志文件。同时结合logging模块实现日志的格式化和输出控制。
线程安全的日志上下文管理器实现
以下是支持多线程并发的日志上下文管理器完整代码：
import logging
import threading
from logging.handlers import RotatingFileHandler
class ConcurrentLogContext:
def __init__(self, log_file, max_bytes=1024*1024, backup_count=3):
self.log_file = log_file
self.max_bytes = max_bytes
self.backup_count = backup_count
self._lock = threading.Lock()
self._logger = None
def __enter__(self):
# 获取锁，保证同一时间只有一个线程初始化日志
self._lock.acquire()
# 初始化logger
self._logger = logging.getLogger(f"concurrent_log_{id(self)}")
self._logger.setLevel(logging.INFO)
# 避免重复添加handler
if not self._logger.handlers:
handler = RotatingFileHandler(
self.log_file,
maxBytes=self.max_bytes,
backupCount=self.backup_count,
encoding="utf-8"
)
formatter = logging.Formatter(
"%(asctime)s - %(threadName)s - %(levelname)s - %(message)s"
)
handler.setFormatter(formatter)
self._logger.addHandler(handler)
return self._logger
def __exit__(self, exc_type, exc_val, exc_tb):
# 释放锁
self._lock.release()
# 移除handler，避免重复日志
if self._logger and self._logger.handlers:
for handler in self._logger.handlers:
handler.close()
self._logger.handlers.clear()
# 不处理异常，异常向上抛出
return False
# 多线程测试代码
def write_log(thread_name, log_context):
for i in range(3):
with log_context as logger:
logger.info(f"{thread_name} 写入第 {i+1} 条日志")
if __name__ == "__main__":
log_context = ConcurrentLogContext("concurrent_test.log")
threads = []
for i in range(3):
t = threading.Thread(
target=write_log,
args=(f"线程{i+1}", log_context),
name=f"Thread-{i+1}"
)
threads.append(t)
t.start()
for t in threads:
t.join()
代码逻辑说明
- 使用
threading.Lock实现线程锁，保证同一时间只有一个线程可以获取logger实例 - 在
__enter__方法中初始化logging的RotatingFileHandler，支持日志文件自动滚动，避免单个日志文件过大 - 日志格式中添加了线程名称，方便区分不同线程的日志内容
- 在
__exit__方法中释放锁并清理handler，避免内存泄漏和重复日志问题
多进程场景下的适配
如果是多进程并发场景，线程锁无法生效，需要替换为文件锁。可以使用fcntl模块（Linux/Mac）或者msvcrt模块（Windows）实现文件级别的锁控制，上下文管理器的__enter__和__exit__方法只需要替换为对应的文件锁获取和释放逻辑即可，核心的日志初始化逻辑不需要修改。
使用注意事项
- 上下文管理器的实例可以复用，但建议每个并发场景使用独立的实例，避免锁竞争过于频繁
- 日志文件路径需要确保执行进程有读写权限，否则会抛出文件操作异常
- 如果不需要日志滚动功能，可以将RotatingFileHandler替换为普通的FileHandler
- 如果并发量非常高，可以考虑使用队列将日志先收集再统一写入，进一步降低锁竞争的开销