并发代码测试：多线程/多进程/异步测试

Python并发编程专题 · 保证并发程序正确性的质量保障体系

专题：Python并发编程系统学习

关键词：Python, 并发编程, 并发测试, pytest-asyncio, 单元测试, 死锁检测, 时序测试

一、并发测试的挑战

并发程序的测试远比串行程序复杂，其根本原因在于执行的非确定性（Non-determinism）。同一段并发代码在多次运行中可能表现出完全不同的行为，这使得"测试通过一次"远远不够。

常见的挑战包括：线程/进程调度顺序不可控，导致时序依赖（Timing-dependent）的bug难以稳定复现；竞争条件（Race Condition）仅在特定交错执行路径下才触发，常规单次测试几乎无法覆盖；测试覆盖不完整，即使大量运行也未必能暴露所有并发路径；此外，死锁、活锁、资源泄漏等问题在常规测试中极易被忽略。

核心难点：并发bug是"概率性"的。一次测试通过不代表代码正确，需要专门设计确定性测试策略来提高bug暴露概率。

二、多线程代码的测试

测试多线程代码的核心策略是确定性（Deterministic）测试：通过显式同步原语（如threading.Event）控制线程间的执行顺序，强制代码走特定的交错路径。

2.1 基础线程安全测试

以经典的线程安全累加器为例，我们需要验证加锁机制是否真正生效：

import threading

class Counter:
    _lock = threading.Lock()
    _value = 0
    def increment(self):
        with self._lock:
            self._value += 1

测试时需要启动多个线程并发调用 increment，然后断言最终值等于调用次数：

import pytest
from concurrent.futures import ThreadPoolExecutor

def test_counter_thread_safety():
    counter = Counter()
    n = 1000
    with ThreadPoolExecutor(max_workers=10) as exe:
        for _ in range(n):
            exe.submit(counter.increment)
    assert counter._value == n

2.2 确定性时序控制（基于Event）

对于复杂的线程交互场景，使用 threading.Event 可以精确控制线程执行的"栅栏点"：

import threading

def test_deterministic_thread_order():
    step = threading.Event()
    results = []

    def worker_a():
        results.append("A_start")
        step.set()                    # 通知主线程
        step.wait()                     # 等待主线程放行
        results.append("A_after_wait")

    t = threading.Thread(target=worker_a)
    t.start()
    step.wait()                           # 等worker_a到达栅栏
    results.append("main_after_a_start")
    step.clear()
    step.set()                           # 放行worker_a
    t.join()
    assert results == ["A_start", "main_after_a_start", "A_after_wait"]

通过这种方式，测试可以复现特定的执行顺序，确保某些先决条件在子线程执行前就绪。

2.3 线程池与并发执行器测试

实际项目中推荐使用 concurrent.futures.ThreadPoolExecutor 简化线程管理，配合 pytest 进行常规测试：

def test_concurrent_file_writes():
    written = []
    lock = threading.Lock()

    def safe_write(msg: str):
        with lock:
            written.append(msg)

    with ThreadPoolExecutor(max_workers=4) as pool:
        for i in range(100):
            pool.submit(safe_write, f"msg-{i}")

    assert len(written) == 100

三、多进程代码的测试

多进程测试面临进程隔离带来的特殊挑战：每个进程有独立的内存空间，无法直接共享Python对象。测试时需要关注进程间通信（IPC）的正确性和共享状态的最终一致性。

3.1 使用multiprocessing.Manager模拟共享状态

import multiprocessing

def test_shared_counter_via_manager():
    manager = multiprocessing.Manager()
    counter = manager.Value('i', 0)
    lock = manager.Lock()

    def worker(c, lk):
        for _ in range(500):
            with lk:
                c.value += 1

    procs = [multiprocessing.Process(target=worker, args=(counter, lock))
             for _ in range(4)]
    for p in procs: p.start()
    for p in procs: p.join()

    assert counter.value == 2000

3.2 进程池测试

from multiprocessing.pool import Pool

def test_process_pool_map():
    def double(x):
        return x * 2

    with Pool(processes=4) as pool:
        result = pool.map(double, [1, 2, 3, 4, 5])
    assert result == [2, 4, 6, 8, 10]

3.3 超时测试（防止死进程）

多进程测试中，如果子进程死锁，测试会无限挂起。必须设置超时保护：

import pytest

@pytest.mark.timeout(5)  # 5秒超时
def test_process_with_timeout():
    def bad_worker():
        while True: pass   # 死循环模拟死锁

    p = multiprocessing.Process(target=bad_worker)
    p.start()
    p.join(timeout=3)
    if p.is_alive():
        p.terminate()
        pytest.fail("进程超时未返回")

四、asyncio代码测试

异步代码的测试在Python生态中已相当成熟。pytest-asyncio 是最主流的方案，它允许在pytest中直接编写和运行协程测试。

4.1 安装与基本用法

# 安装：pip install pytest-asyncio

import pytest

@pytest.mark.asyncio
async def test_async_fetch():
    result = await fetch_data()
    assert result is not None

4.2 事件循环管理

pytest-asyncio 会自动为每个测试函数创建独立的事件循环，但也支持自定义事件循环策略：

@pytest.fixture
def event_loop():
    loop = asyncio.new_event_loop()
    yield loop
    loop.close()

@pytest.mark.asyncio
async def test_custom_event_loop():
    assert asyncio.get_running_loop() is not None

4.3 异步超时测试

import asyncio

@pytest.mark.asyncio
async def test_async_timeout():
    with pytest.raises(asyncio.TimeoutError):
        await asyncio.wait_for(
            asyncio.sleep(10),
            timeout=0.1
        )

4.4 异步上下文管理器与资源清理测试

@pytest.mark.asyncio
async def test_async_resource_cleanup():
    class AsyncResource:
        def __init__(self):
            self.closed = False
        async def close(self):
            self.closed = True

    async def use_resource():
        res = AsyncResource()
        await res.close()
        return res.closed

    result = await use_resource()
    assert result is True

五、超时测试

超时测试是并发测试中最重要的安全网之一。无论是线程死锁、进程挂起还是协程阻塞，没有超时保护的测试可能无限期挂起，拖垮整个CI流水线。

5.1 pytest-timeout 插件

# 安装：pip install pytest-timeout
# 全局设置：pytest --timeout=30
# 或者在 pytest.ini 中设置：
# [pytest]
# timeout = 30

# 单函数超时（使用装饰器）
@pytest.mark.timeout(5)
def test_maybe_deadlock():
    # 如果死锁，5秒后自动失败
    run_concurrent_code()

5.2 死锁检测的定时中断

更精细的死锁检测可以结合信号机制（仅限Unix）或通过看门狗线程实现：

import signal

class TimeoutException(Exception):
    pass

def handler(signum, frame):
    raise TimeoutException("操作超时，疑似死锁")

def test_with_alarm():
    signal.signal(signal.SIGALRM, handler)
    signal.alarm(3)  # 3秒后触发中断
    try:
        run_concurrent_code()
    except TimeoutException:
        pytest.fail("死锁检测：测试超时")
    finally:
        signal.alarm(0)  # 取消闹钟

最佳实践：始终为并发测试设置全局超时（pytest-timeout），同时为个别"高风险"测试设置更短的超时时间。超时时间应足够长以通过正常测试，又足够短以避免CI阻塞。

六、模拟（Mock）外部依赖

并发测试中，外部依赖（网络请求、数据库、文件系统）会引入额外的不确定性。unittest.mock 可以有效隔离这些依赖，使测试更加确定和快速。

6.1 模拟网络调用

from unittest.mock import AsyncMock, patch

@pytest.mark.asyncio
async def test_fetch_with_mock():
    mock_response = {"status": "ok", "data": [1, 2, 3]}

    with patch("my_module.fetch_data", AsyncMock(return_value=mock_response)):
        from my_module import processor
        result = await processor.process()
        assert result["status"] == "ok"

6.2 模拟并发场景中的共享资源

from unittest.mock import PropertyMock

def test_concurrent_mock_shared_state():
    with patch("my_module.Config.TIMEOUT",
               PropertyMock(return_value=0.01)):
        # 在极短超时下测试并发行为
        results = run_concurrent_tasks()
        assert len(results) == expected_count

七、压力测试与稳定性验证

单元测试和集成测试之外，压力测试（Stress Testing）是发现并发bug的关键手段。通过在高压条件下反复执行目标代码，增加线程交错排列的覆盖范围。

7.1 重复执行暴露竞争条件

def test_race_condition_stress():
    for _ in range(100):  # 重复100次提高竞争条件暴露概率
        counter = Counter()
        with ThreadPoolExecutor(max_workers=8) as exe:
            futures = [exe.submit(counter.increment) for _ in range(500)]
        for f in futures:
            f.result()
        assert counter._value == 500

7.2 使用stress工具框架

可以使用 pytest-repeat 插件或自定义的stress runner进行更高强度的压力测试：

# 安装：pip install pytest-repeat
# 运行：pytest --count=200 -x test_concurrent.py
# --count=200 表示每个测试执行200次
# -x 表示首次失败即停止

# 也可结合多进程并行加速：
# pip install pytest-xdist
# pytest --count=200 -n auto test_concurrent.py

7.3 压力测试结果分析

测试结果应至少包含以下指标：

成功率：N次运行中通过的比例
最小/最大/平均执行时间：检测是否存在偶发性性能退化
资源泄漏：线程数、文件描述符、内存使用是否持续增长
失败模式归类：同一assertion失败的比例越高，越容易修复

八、测试最佳实践总结

核心原则：确定性优先，超时兜底，压力验证。

确定性优先：尽可能使用 Event、Condition 等同步原语控制执行顺序，编写可预测的测试用例。
隔离外部依赖：使用 Mock/AsyncMock 隔离网络、数据库等不确定性来源，使测试聚焦于并发逻辑本身。
超时保护：所有并发测试必须设置超时（pytest-timeout全局超时 + 特定测试的细粒度超时），防止死锁导致CI卡死。
充分的并发压力：利用 pytest-repeat 多次执行同一测试，增加并发交错路径的覆盖率；结合 pytest-xdist 多进程加速。
资源泄漏检测：测试结束后验证线程/进程是否全部回收，文件描述符是否关闭。
日志辅助调试：在关键同步点加入结构化日志，测试失败时能重建执行时序。
分层测试：从单元测试（mock外部依赖）到集成测试（真实IPC）再到压力测试（高并发重复执行），逐层验证。

"并发程序的正确性不是靠运气，而是靠系统化的测试策略。每一个Event、每一个超时、每一个重复执行，都在缩小非确定性的不确定性空间。"