★ FEATURED ARTICLE 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几乎每一个测试生命周期节点都可以被干预。我主要用了其中几个关键 hookpytest_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 配额可以精确分配到一个 worker3.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(fcase 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 环境准备与依赖清单在开始之前先列一下我在项目里使用的核心依赖版本这些版本是经过实际验证的不是随便选的依赖库版本作用pytest7.x测试框架基座pytest-xdist3.x进程级并发分发pytest-rerunfailures11.x失败重试机制部分功能用自己的封装PyYAML6.x读取配置文件requests2.x接口测试客户端selenium4.xUI 自动化视情况可换 playwright4.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(tryfirstTrue) 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 --distloadgroup --htmlreport/html/report.html4.4 入口脚本设计一键触发所有类型工程落地时执行命令不能太复杂。我把所有调度细节封装进run.py外部只需要指定“跑什么环境、跑哪一组用例”# run.py import sys import subprocess def run(tag): cmd_map { all: pytest -c engine_config.yaml -n 4 --distloadgroup --htmlreport/html/report.html, api: pytest tests/api_tests -c engine_config.yaml -n 4 --distloadgroup --htmlreport/html/report_api.html, ui: pytest tests/ui_tests -c engine_config.yaml -n 2 --distloadgroup --htmlreport/html/report_ui.html, unit: pytest tests/unit_tests -c engine_config.yaml -n 8 --distloadgroup --htmlreport/html/report_unit.html, } subprocess.run(cmd_map[tag], shellTrue, checkTrue) 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, timeout30): result [] def target(): result.append(item.execute()) t Thread(targettarget) t.start() t.join(timeout) if t.is_alive(): raise TimeoutError(fcase {item.name} timeout) return result[0]5.4 重试后统计数据失真重试本身会污染测试统计数据。比如一条用例第一次跑失败、第二次跑通过如果统计逻辑不区分最终报告会显示“1 条用例 1 个失败 1 个通过”数据就对不上了。所以我要求重试后的报告必须精确展示“首次执行结果”和“最终执行结果”。我在reporter.py中维护了一个执行历史表每条用例的每一次执行都会追加一行记录用例名执行次数本次结果耗时test_login1失败1.2stest_login2通过0.8s最终报告只统计“最终结果”但“执行次数”列会暴露哪些用例反复抖动。直观的表达方式就让团队能主动关注那些不稳定用例而不是被重试机制蒙蔽。5.5 配置漂移问题的排查工程上线一段时间后经常出现“本地跑得好好的CI 上用例集体失败”的诡异问题。最终定位到是配置漂移本地和 CI 读取的engine_config.yaml不是同一份worker 数、重试次数等关键参数不一致。我给配置文件加了 MD5 校验插件启动时打印当前配置指纹并在报告里输出。这样每个人跑出来的报告都自带“环境配置快照”任何配置漂移一眼就能看出来。这算是小而美的工程实践。6. 按个人经验做的小结与后续扩展回到开头说的那个命题——并发执行引擎不是简单把 pytest-xdist 接进来就完事了它本质上是一个测试资源调度系统。从用例标记体系、分组策略、调度顺序、worker 配额、重试机制到报告聚合都需要有全局观的工程设计。如果你只是想临时把测试提速直接在命令行加-n 4就够了但如果要长期维护、多团队共用有一套统一的分组引擎带来的稳定性收益是肉眼可见的。我在做这个项目的过程中最深刻的体会是所谓工程落地不是把代码跑通就算结束而是要考虑扩展性的边界。比如团队后续加入 UI 框架从原来的 Selenium 切到 Playwright引擎能不能平滑支持还好因为我在设计时把“并发策略”和“具体驱动”解耦了UI 层只用 marker 标记驱动方式是独立封装所以换 Playwright 时引擎完全不用改只换驱动适配层即可。如果再往后迭代我最想加的两个能力一个是“智能超时调整”根据历史执行时长动态预测超时上限而不是拍脑袋定死另一个是“失败用例自动聚类”把同一类原因导致的失败批量归因。这些都需要更多数据样本支撑不过目前的分组执行引擎已经给后续演进打好了数据基础。最后给已经在动手做这件事的同学一个最实用的建议一定先梳理清楚你项目里用例之间的依赖关系然后再开并发。依赖关系没梳理清楚任何并发策略都会让你怀疑人生。理清之后你会发现花在分组和调度上的时间是整套工程里性价比最高的一笔投入。
阅读完成 · 觉得有帮助？
