pytest并行测试中multiprocessing锁未生效，如何解决同步问题？

每个测试worker都是独立的子进程，multiprocessing.Lock是进程内的本地对象——当每个worker启动时，都会重新加载conftest.py并创建属于自己的锁实例。这些锁完全独立，根本无法实现跨进程的同步，自然挡不住多个worker同时进入临界区。

方法1：使用文件锁（通用跨进程方案） #

文件锁基于文件系统实现，能跨多个独立进程生效，是这种场景下最直接的解决方案。可以用跨平台的第三方库filelock来简化实现：

修改conftest.py如下：

import pytest
from filelock import FileLock

@pytest.fixture
def preprocess(request):
    # 指定一个全局唯一的文件路径作为锁载体
    with FileLock("/tmp/test_preprocess.lock"):
        # 这里写你的临界区预处理逻辑
        config = request.param
        # 示例：模拟预处理操作
        print(f"正在执行[{config}]的预处理...")
        # 你的实际预处理代码
        processed_result = f"处理完成的{config}数据"
        return processed_result

先安装依赖：pip install filelock

方法2：全局一次性预处理（适合无需重复执行的场景） #

如果你的预处理逻辑全局只需要执行一次，不需要每个测试都触发，可以借助pytest-xdist的钩子机制，让主进程单独完成预处理，再把结果传递给所有worker：

修改conftest.py如下：

import pytest

# 主进程中存储预处理结果
_global_preprocessed = {}

def pytest_configure(config):
    # 只有主进程会进入这个分支（worker进程会有workerinput属性）
    if not hasattr(config, "workerinput"):
        # 在这里执行全局一次性预处理
        _global_preprocessed["config1"] = "预处理完成的config1数据"
        _global_preprocessed["config2"] = "预处理完成的config2数据"

def pytest_configure_node(node):
    # 将预处理结果传递给每个worker
    node.workerinput["preprocessed_data"] = _global_preprocessed

@pytest.fixture
def preprocess(request):
    # 直接从worker的共享数据中获取结果，无需锁
    return node.workerinput["preprocessed_data"][request.param]

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