pytest并发分组调度引擎设计与实现：从用例标记到稳定执行

1. 需求拆解：为什么需要“并发 + 分组”这套组合拳

先聊一下背景。我维护的测试体系里同时存在三种类型的自动化用例：单元测试、接口测试、UI 测试。一开始它们互不干扰，各跑各的，但随着业务迭代，用例规模从几百条涨到几千条，执行时间从几分钟拉长到一小时以上，问题就来了——每次发版前，测试环节变成了最堵的瓶颈。

很多团队的第一反应是“上并发”，把 pytest-xdist 拉进来，-n auto一开，所有用例散到多个 worker 上跑。这个方案对纯单元测试和接口测试确实有效，但对 UI 测试来说，问题非常大：浏览器进程资源占用高，多个 worker 同时起浏览器，机器直接卡死，再加上 UI 用例之间往往存在强耦合（比如依赖登录态、依赖前置数据），盲目并发的结果就是一堆随机失败。

所以我在做这个引擎时，核心思路不是“无脑并发”，而是“先分组，再调度”。把用例按类型天然分组，再为不同组定制不同的并发策略和资源配额。听起来简单，真正落到工程上要考虑的事情不少。

这篇博文的受众，我默认你是有过 pytest 使用经验、正在搭建或改造自动化测试框架的测试开发同学。如果你想解决以下三类问题，这篇文章会有直接参考价值：一是测试执行时间太长，想并发提速但怕不稳定；二是 UI / 接口 / 单元用例混在一个项目里，执行策略难以差异化定制；三是并发之后用例执行结果分散，失败重试、日志聚合、报告输出都变得混乱，需要一个统一收口方案。

2. 整体架构设计：从用例到调度器，一层层拆开看

2.1 先定模型：并发不能解决所有问题，分组才是基础

我在设计这个引擎之前，把用例执行拆成了几个相互独立的维度。第一个维度是“执行方式”，也就是串行还是并发；第二个维度是“资源类型”，比如浏览器、数据库连接、Redis、临时文件；第三个维度是“稳定性要求”，比如哪些用例是高频稳定用例，哪些是容易受环境影响的用例。

把这三个维度叠加在一起，就得出了分组的基本逻辑。UI 用例必须考虑浏览器资源上限，接口用例相对轻量，但不适合跟 UI 用例混在同一进程里跑，因为 UI 用例的进程往往需要更大的内存分配。单元测试基本无外设依赖，容错率最高，可以开足马力并发。

分组不是简单按用例类型切一刀，还涉及业务模块之间的依赖关系。比如用户模块的接口用例要注册账号，而商品模块的用例可能依赖用户模块的账号数据，这种跨模块的隐式依赖在并发时最容易出问题。所以我在项目里引入了一个“确认依赖链”的步骤，先标记出有共享状态的用例组，强制串行或延迟启动。

2.2 用例分层与标记体系：让调度器认识每一类用例

实现分组的第一步，是建立一套能够被机器识别的标记体系。我在 pytest 里通过自定义 marker 来实现，而不是靠目录名或者文件名硬编码判断，因为同一个测试文件里可能会混合不同类型用例。

# conftest.py 中注册自定义 marker def pytest_configure(config): config.addinivalue_line("markers", "ui: UI 自动化用例") config.addinivalue_line("markers", "api: 接口自动化用例") config.addinivalue_line("markers", "unit: 单元测试用例")

用例写法上只需要乖乖打上标记即可：

@pytest.mark.ui def test_login_page(): ...

当然，光有 marker 还不够。调度器需要更丰富的元数据来控制并发和顺序，比如用例的“资源占用等级”、“是否允许并发执行”、“失败后是否允许重试”。这些我用 fixture 标签属性来维护，避免 marker 满天飞。

2.3 调度顺序：先跑谁、再跑谁，这是一门艺术

分完组之后，下一个问题是执行顺序。表面上所有用例最终都会跑完，但顺序其实决定了整体效率和稳定性。

我的做法是“单元优先、接口次之、UI 最后”。单元测试失败时，能最快反馈核心逻辑问题；接口依赖单元测试的模型层稳定性；而 UI 用例则是“最终验收性质”的，应该在前面两类都通过后再执行。如果 UI 用例一上来就红成一片，大概率不是 UI 自身问题，而是底层接口或数据状态不对。

除了类型顺序，同组内部还会区分“冒烟用例”和“全量用例”。冒烟用例是核心链路，每次提交代码都跑，全量用例则是在合并前或发版前跑。分组引擎要把这两类用例分开调度，避免冒烟用例被长耗时用例拖在后面。

2.4 依赖管理：并发场景下“共享状态”是最大的坑

并发测试最大的坑，就是用例之间共享了可变状态。单元测试还好，接口测试因为涉及数据库数据，并发时如果两个用例同时插入同一条记录，大概率其中一个会失败。

我在分组引擎里设计了一个依赖声明机制，核心非常简单：每个用例可以声明它依赖哪些“资源锁”，引擎在分配 worker 时会保证持有同一把锁的用例不会同时执行。

# 用例依赖声明示例 @pytest.mark.resource_lock("user_register") def test_register_duplicate_username(): ...

锁的粒度是逻辑级别的，不直接操作数据库锁或文件锁，这样对业务代码零侵入。实际效果是：允许并发，但不允许冲突。

3. 核心实现细节：并发调度引擎的代码长什么样

3.1 从 pytest 插件机制谈起

整套引擎的本质，是一个自定义的 pytest 插件。pytest 自身提供了非常丰富的 hook，几乎每一个测试生命周期节点都可以被干预。我主要用了其中几个关键 hook：

• pytest_collection_modifyitems：用例收集完成后的处理节点，在这里做分组、排序、打标。
• pytest_configure：注册插件配置和自定义 marker。
• pytest_runtest_protocol：控制单个用例的执行流程，重试逻辑和资源锁逻辑就挂在这里。
• pytest_terminal_summary：汇总执行数据，输出分组统计。

关键在哪？在pytest_collection_modifyitems这个 hook 里。因为这一步发生在用例分发到 worker 之前，可以对用例列表进行“任意处置”——排序、分组、过滤、打标。对于分组引擎来说，这是调度策略的入口。

3.2 分组与并发策略的动态注入

我需要让“策略”可以被外部配置动态改变，而不是写死在代码里。所以引擎读取一个 yaml 配置文件，定义每个分组的并发策略。

# engine_config.yaml groups: unit: pattern: "unit" parallel_mode: "process" workers: 8 retry: 1 api: pattern: "api" parallel_mode: "process" workers: 4 retry: 2 ui: pattern: "ui" parallel_mode: "thread" workers: 2 retry: 1 max_workers_per_host: 2

这里有个需要解释的细节：为什么单元和接口用process模式，而 UI 用thread模式。pytest-xdist 的默认并发单位是进程，进程隔离性强，适合 CPU 密集或数据冲突多的场景。但是 UI 用例有个特殊性——浏览器实例本身是多线程的，且同一台机器上浏览器窗口过多会导致资源竞争，所以我对 UI 组启用线程级并发，且限制同一时间最多启动两个 worker，减少浏览器风暴。

你可能要问，pytest-xdist 怎么支持线程模式呢？它不是只支持进程模式吗？答案是可以结合pytest-parallel或pytest-xdist的--dist模式配合自定义分发方式。实际上，我在工程里直接用pytest-xdist的loadgroup分发方式，配合自定义分组标记，让每个 worker 负责一整个 group，而不是单独一条用例。

3.3 worker 分配与用例分发

“按组分配”和“按条分配”是两种完全不同的并发逻辑。按条分配是 xdist 默认行为，所有 worker 对用例池做动态取用，优点是负载自动均衡，缺点是无法控制跨 worker 的用例顺序。

分组引擎采用的是“按组分配”：引擎把所有用例先分成 N 个组，每个组交给一个 worker 独立执行，组与组之间互不干扰。这样带来的直接好处是：

• 组内可以继续串行执行有依赖关系的用例
• 某个组挂了不会影响其他组
• 内存和 CPU 配额可以精确分配到一个 worker

3.4 稳定性保障：失败重试和用例隔离

并发引擎如果不做失败重试，稳定性就是一句空话。UI 用例尤其如此，偶尔网络波动、等待超时都会导致用例失败，但用例本身逻辑没变。

我在插件里封装了一个重试装饰器，挂在用例执行协议上。与网上常见的pytest-rerunfailures插件不同，我的重试逻辑做了一点增强：支持“按错误类型重试”，比如网络超时错误会重试，断言失败则不重试。

# 简化版重试逻辑 class RetryOnFailure: def __init__(self, max_retries, retry_for_exceptions): self.max_retries = max_retries self.retry_for_exceptions = retry_for_exceptions def run(self, test_item): for attempt in range(self.max_retries + 1): try: test_item.execute() return except self.retry_for_exceptions as exc: if attempt < self.max_retries: log_warning(f"case failed, retrying: {exc}") continue raise

不过这里要特别提醒：重试不是万能药。如果用例本身因为数据污染而失败，重试只会让问题更隐蔽。我的建议是给重试单独打一条日志标记，报告中明确展示哪些用例是重试后通过的。业务方看到“失败 1 次后重试通过”和“一次通过”，对稳定性的感知是完全不同的。

3.5 日志隔离与聚合

并发执行时，最头疼的问题之一就是日志相互交织。多个 worker 同时向 stdout 打印，最终日志文件根本无法定位问题。

我在插件里给每个 worker 分配了一个日志文件，所有 worker 日志按test_xxx_worker_{index}.log格式命名，在用例结束后统一聚合到report/logs/目录。最关键的一步是日志文件里要打上 worker ID 和用例 ID，这样哪怕只给一条错误用例追日志，也能快速定位到具体是哪个进程、哪一步操作。

如果上述步骤不加以控制，多进程下日志是乱成一锅粥的。控制住了日志的写入路径，排查问题的时间可以缩短一半以上。

4. 实操过程：搭建一个可运行的分组并发执行引擎

4.1 环境准备与依赖清单

在开始之前，先列一下我在项目里使用的核心依赖版本，这些版本是经过实际验证的，不是随便选的：

4.2 项目结构设计

项目目录结构决定了工程的边界感，我把调度引擎、用例、报告分离为三个独立层次：

concurrent_engine/ ├── engine/ │ ├── __init__.py │ ├── plugin.py # pytest 插件主逻辑 │ ├── config.py # 配置读取与校验 │ ├── scheduler.py # 分组策略与调度顺序 │ ├── retry.py # 重试封装 │ └── reporter.py # 报告输出与聚合 ├── tests/ │ ├── unit_tests/ │ ├── api_tests/ │ └── ui_tests/ ├── report/ │ ├── logs/ │ └── html/ ├── engine_config.yaml └── run.py # 入口脚本

engine 目录承载引擎所有核心逻辑，tests 目录严格按照用例类型分层，report 目录是输出目录。这样设计的好处是“引擎”和“用例”完全解耦，以后哪怕用例规模扩大 10 倍，引擎部分也不会有大改动。

4.3 核心代码：调度器与插件

调度器的核心责任是读取配置，对收集到的用例做分组排序。代码不复杂，但逻辑要严谨：

# engine/scheduler.py class Scheduler: def __init__(self, config: dict): self.groups = config["groups"] def organize(self, items): grouped_items = {group_name: [] for group_name in self.groups} for item in items: for group_name, group_conf in self.groups.items(): if self._match_group(item, group_conf["pattern"]): grouped_items[group_name].append(item) break return grouped_items def _match_group(self, item, pattern): # 支持按 marker 或关键字匹配 if pattern in item.keywords: return True return False

插件主逻辑挂接 hook，把调度结果应用到执行阶段：

# engine/plugin.py import pytest @pytest.hookimpl(tryfirst=True) def pytest_collection_modifyitems(session, config, items): scheduler = Scheduler(load_config()) grouped = scheduler.organize(items) # 按照：unit -> api -> ui 的顺序重新排列用例 order = ["unit", "api", "ui"] final_items = [] for g in order: if g in grouped: final_items.extend(grouped[g]) items[:] = final_items

在命令行执行时，利用 xdist 的-n参数指定 worker 数，配合分组后的用例列表，就能实现“分组并发”：

pytest -c engine_config.yaml -n 4 --dist=loadgroup --html=report/html/report.html

4.4 入口脚本设计：一键触发所有类型

工程落地时，执行命令不能太复杂。我把所有调度细节封装进run.py，外部只需要指定“跑什么环境、跑哪一组用例”：

# run.py import sys import subprocess def run(tag): cmd_map = { "all": "pytest -c engine_config.yaml -n 4 --dist=loadgroup --html=report/html/report.html", "api": "pytest tests/api_tests -c engine_config.yaml -n 4 --dist=loadgroup --html=report/html/report_api.html", "ui": "pytest tests/ui_tests -c engine_config.yaml -n 2 --dist=loadgroup --html=report/html/report_ui.html", "unit": "pytest tests/unit_tests -c engine_config.yaml -n 8 --dist=loadgroup --html=report/html/report_unit.html", } subprocess.run(cmd_map[tag], shell=True, check=True) if __name__ == "__main__": run(sys.argv[1])

不要小看这个入口脚本，它在 CI 里很重要，团队其他人并不需要理解 pytest 参数细节，只需知道“跑全部”或“跑接口”，这大大降低了使用门槛。

4.5 与 CI 流水线的集成

在 CI 里使用时，关键是阶段划分。我建议按“三级流水线”接入：代码提交后先跑单元测试（快速反馈），构建成功后再跑接口测试（模块验证），部署到测试环境后最后跑 UI 测试（端到端验收）。

对应到流水线里就是三个 stage：

# 伪代码示例 stages: - unit_test: script: python run.py unit - api_test: script: python run.py api - ui_test: script: python run.py ui

这个顺序还有个好处：每一阶段都会产出独立的报告文件和日志文件，一旦某一阶段挂掉，团队可以快速定位到是代码问题、接口问题还是 UI 问题，而不是三个层级混在一起瞎猜。

5. 踩坑实录：并发分组落地时我遇到的常见问题

5.1 并发导致数据库中的脏数据

这是接口测试并发最容易碰到的问题。两个用例同时往同一张表插入相同唯一键的数据，后一个用例必然失败。我的解决方式不是彻底避免冲突，而是“可接受部分冲突”，把冲突用例单独放到一个不可并发的组里。操作上就是给这些用例打上@pytest.mark.serial标记，调度器遇到这个标记后强制放入serial_group，该组内 worker 数为 1。

# 调度器对 serial_group 的处理 if item.get_closest_marker("serial"): grouped_items.setdefault("serial_group", []).append(item)

类似的思路可以用在 UI 登录态共享场景，比如多个用例都需要同一个登录令牌，那就把它们放进 serial_group，避免令牌过期或被刷新。

5.2 UI 用例并发时的浏览器会话隔离

UI 用例并发最大的坑是浏览器上下文串了。同一个 WebDriver 实例如果驱动两个测试会话，页面元素查询经常拿到另一条用例的残留元素。这个问题我在多个浏览器驱动下都踩过。

最终方案：每个 UI worker 分配独立的 WebDriver 实例，并且在启动前清理所有残留的 profile 目录。两个 UI 用例尽量用不同的测试账号登录，避免 cookie 相互覆盖。账号池设计是：一组用例对应一个专属账号组，资源释放时账号退回到池中。

5.3 pytest-xdist 插件之间的兼容性问题

重试插件跟 xdist 之间有时会冲突，比如重试报错显示 “Worker node went missing” 之类。排查下来发现是重试过程中用例已经被 xdist 判定为超时，导致 worker 被回收。解决办法是给重试逻辑设置“单条用例最长执行时间”，超时后不再重试，直接失败，然后由外界重跑整个 group。

我封装了一个简易的超时控制：

# 伪代码，示意超时控制思想 def run_with_timeout(item, timeout=30): result = [] def target(): result.append(item.execute()) t = Thread(target=target) t.start() t.join(timeout) if t.is_alive(): raise TimeoutError(f"case {item.name} timeout") return result[0]

5.4 重试后统计数据失真

重试本身会污染测试统计数据。比如一条用例第一次跑失败、第二次跑通过，如果统计逻辑不区分，最终报告会显示“1 条用例 1 个失败 1 个通过”，数据就对不上了。

所以我要求重试后的报告必须精确展示“首次执行结果”和“最终执行结果”。我在reporter.py中维护了一个执行历史表，每条用例的每一次执行都会追加一行记录：

最终报告只统计“最终结果”，但“执行次数”列会暴露哪些用例反复抖动。直观的表达方式就让团队能主动关注那些不稳定用例，而不是被重试机制蒙蔽。

5.5 配置漂移问题的排查

工程上线一段时间后，经常出现“本地跑得好好的，CI 上用例集体失败”的诡异问题。最终定位到是配置漂移：本地和 CI 读取的engine_config.yaml不是同一份，worker 数、重试次数等关键参数不一致。

我给配置文件加了 MD5 校验，插件启动时打印当前配置指纹，并在报告里输出。这样每个人跑出来的报告都自带“环境配置快照”，任何配置漂移一眼就能看出来。这算是小而美的工程实践。

6. 按个人经验做的小结与后续扩展

回到开头说的那个命题——并发执行引擎不是简单把 pytest-xdist 接进来就完事了，它本质上是一个测试资源调度系统。从用例标记体系、分组策略、调度顺序、worker 配额、重试机制到报告聚合，都需要有全局观的工程设计。如果你只是想临时把测试提速，直接在命令行加-n 4就够了，但如果要长期维护、多团队共用，有一套统一的分组引擎带来的稳定性收益是肉眼可见的。

我在做这个项目的过程中，最深刻的体会是：所谓工程落地，不是把代码跑通就算结束，而是要考虑扩展性的边界。比如团队后续加入 UI 框架，从原来的 Selenium 切到 Playwright，引擎能不能平滑支持？还好因为我在设计时把“并发策略”和“具体驱动”解耦了，UI 层只用 marker 标记，驱动方式是独立封装，所以换 Playwright 时引擎完全不用改，只换驱动适配层即可。

如果再往后迭代，我最想加的两个能力：一个是“智能超时调整”，根据历史执行时长动态预测超时上限，而不是拍脑袋定死；另一个是“失败用例自动聚类”，把同一类原因导致的失败批量归因。这些都需要更多数据样本支撑，不过目前的分组执行引擎已经给后续演进打好了数据基础。

最后给已经在动手做这件事的同学一个最实用的建议：一定先梳理清楚你项目里用例之间的依赖关系，然后再开并发。依赖关系没梳理清楚，任何并发策略都会让你怀疑人生。理清之后你会发现，花在分组和调度上的时间，是整套工程里性价比最高的一笔投入。