Python上下文内函数调用监控的多线程问题及优化需求

原实现的核心问题在于全局共享的handlers实例——所有线程共用同一个_handlers列表，导致线程A的上下文会接收到线程B中函数执行的记录，完全破坏了线程隔离性，最终出现记录混乱的情况。

利用Python的threading.local()为每个线程维护独立的上下文处理器集合，确保每个线程的上下文只记录当前线程内的函数执行数据。

修改后的完整代码 #

import time
import threading
from dataclasses import dataclass

@dataclass
class MonitorRecord:
    function: str
    time: float

class MonitorContext:
    def __init__(self):
        self._records: list[MonitorRecord] = []

    def add_record(self, record: MonitorRecord) -> None:
        self._records.append(record)

    def __enter__(self) -> 'MonitorContext':
        # 获取当前线程的处理器集合，不存在则初始化
        thread_handlers = get_thread_handlers()
        thread_handlers.register(self)
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        thread_handlers = get_thread_handlers()
        thread_handlers.delete(self)
        return

class MonitorHandlers:
    def __init__(self):
        self._handlers: list[MonitorContext] = []

    def register(self, handler: MonitorContext) -> None:
        self._handlers.append(handler)

    def delete(self, handler: MonitorContext) -> None:
        self._handlers.remove(handler)

    def add_record(self, record: MonitorRecord) -> None:
        for h in self._handlers:
            h.add_record(record)

# 线程局部存储，每个线程拥有独立的MonitorHandlers实例
_local = threading.local()

def get_thread_handlers() -> MonitorHandlers:
    """获取当前线程的上下文处理器集合"""
    if not hasattr(_local, 'handlers'):
        _local.handlers = MonitorHandlers()
    return _local.handlers

def monitor_decorator(f):
    def _(*args, **kwargs):
        start = time.time()
        result = f(*args, **kwargs)  # 保留原函数返回值
        thread_handlers = get_thread_handlers()
        # 仅当当前线程有活跃上下文时才记录
        if thread_handlers._handlers:
            thread_handlers.add_record(
                MonitorRecord(
                    function=f.__name__,
                    time=time.time() - start,
                )
            )
        return result
    return _

关键改动说明 #

• 线程局部存储替代全局变量：用threading.local()创建_local对象，每个线程访问_local.handlers时都会得到自己的MonitorHandlers实例，彻底隔离线程间的上下文集合。
• 懒加载线程处理器：get_thread_handlers()函数确保每个线程首次需要时才初始化MonitorHandlers，避免不必要的资源占用。
• 保留原函数返回值：修复了原装饰器丢失函数返回值的问题。
• 空上下文判断：添加了if thread_handlers._handlers判断，确保只有在上下文内执行函数时才记录，符合需求中“仅在上下文内执行时记录”的要求。

多线程场景测试验证 #

运行以下测试代码：

@monitor_decorator
def run():
    time.sleep(0.1)

def nested():
    with MonitorContext() as m:
        run()
    print(len(m._records))

with MonitorContext() as m1:
    threads = [threading.Thread(target=nested) for i in range(10)]
    [t.start() for t in threads]
    [t.join() for t in threads]
print(len(m1._records))

预期输出 #

1
1
1
1
1
1
1
1
1
1
0

每个子线程的上下文只记录自己的1次run调用，主线程的m1没有记录子线程的执行，因为线程隔离。

如果需要主线程上下文记录自身线程内的函数执行，修改测试代码如下：

@monitor_decorator
def run():
    time.sleep(0.1)

def nested():
    with MonitorContext() as m:
        run()
    print(len(m._records))

with MonitorContext() as m1:
    run()  # 主线程内执行函数，会被m1记录
    threads = [threading.Thread(target=nested) for i in range(10)]
    [t.start() for t in threads]
    [t.join() for t in threads]
print(len(m1._records))

预期输出 #

1
1
1
1
1
1
1
1
1
1
1

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