★ FEATURED ARTICLE 1. 多设备并发跑 Appium 这件事到底难在哪做过移动端自动化的朋友大概率都经历过这个阶段单台设备跑 pytest Appium 脚本已经很顺了用例能过、报告能出但只要项目稍微上点规模测试机从一台变成三台五台问题就全冒出来了。最典型的就是——脚本串行跑一台跑完再跑下一台本来十分钟能跑完的回归用例硬生生拖到四十分钟CI 流水线排队排到天亮。这个项目的核心目标很明确用 pytest 驱动 Appium在多台设备上并发执行测试用例。说白了就是让三台、五台甚至更多手机同时跑脚本把原本串行的时间压缩成并行的时间。听起来简单但真正落地的时候坑比想象中多得多。适合谁来参考这篇内容如果你已经写过 Appium 的基础脚本知道webdriver.Remote怎么连设备pytest 的 fixture 和参数化也用过但一到多设备就不知道怎么组织代码结构那这篇就是写给你的。如果你还没接触过 Appium建议先把单设备跑通再来看并发不然会晕。我先说结论多设备并发的核心难点不在“并发”本身而在于设备与用例的分配、Appium 端口的隔离、driver 实例的线程安全管理以及 pytest 收集机制和多线程的配合方式。这四个问题任何一个没处理好脚本要么报错要么结果错乱要么干脆卡死不动。下面我会把这几个点逐个拆开结合我自己踩过的坑给出可以直接抄的方案。2. 整体方案设计与技术选型思路2.1 为什么选 pytest 而不是 unittest很多人一开始用的是 unittest因为 Python 标准库自带不用额外装。但一旦涉及多设备并发unittest 的短板就暴露了。pytest 的优势在于它的 fixture 机制天然适合做“每个线程独立初始化 driver”这件事scope可以精确控制到 function 级别每个测试函数拿到自己的 driver 实例互不干扰。另外 pytest 的参数化pytest.mark.parametrize配合设备列表可以很优雅地实现“同一组用例在不同设备上跑”。还有一点很实际pytest 的插件生态。pytest-xdist可以做进程级并发pytest-html出报告allure-pytest出更漂亮的报告这些都是现成的不用自己造轮子。unittest 要做同样的事得写一堆自定义的 TestLoader 和 TestRunner维护成本高。2.2 多线程 vs 多进程为什么最终选了多线程这是我在方案设计阶段纠结最久的一个问题。先说结论Appium 多设备并发用多线程threading就够了不需要上多进程multiprocessing。原因在于 Appium 测试的本质是 IO 密集型不是 CPU 密集型。脚本大部分时间花在等待元素出现、等待页面加载、等待接口返回上CPU 基本是闲着的。Python 的 GIL全局解释器锁确实限制了多线程的 CPU 并行能力但对 IO 密集型任务影响很小因为线程在等待 IO 的时候会释放 GIL。所以多线程完全能撑住五台甚至十台设备的并发。多进程的问题在于进程间通信麻烦driver 实例没法共享日志收集也乱而且每个进程都要重新初始化一套环境内存开销大。除非你的用例里有大量图像对比、OCR 识别这类 CPU 密集操作否则没必要上多进程。2.3 设备分配策略静态分配还是动态分配设备分配有两种思路。一种是静态分配启动时就确定好“线程 A 跑设备 1线程 B 跑设备 2”每个线程从头到尾绑定一台设备。另一种是动态分配用一个队列存设备线程跑完一个用例后从队列里取下一台空闲设备。我实测下来静态分配更适合大多数场景。原因是 Appium session 的创建和销毁是有成本的每次新建 session 大概要 3 到 8 秒如果每个用例都重新分配设备、重建 session光初始化就浪费大量时间。静态分配让每个线程维护一个长连接的 driver用例之间复用效率高很多。动态分配适合那种用例执行时间差异极大的场景比如有的用例跑 10 秒有的跑 5 分钟静态分配会导致某些线程早早跑完闲着而某些线程还在苦等。但这种场景比较少见大部分回归用例的执行时间是比较均匀的。2.4 端口规划别让 Appium 端口打架每台设备需要独立的 Appium server 实例每个实例监听不同的端口。这是最容易被忽略但最容易出问题的地方。如果你只启动一个 Appium server多个线程同时连上去session 会互相覆盖表现为“脚本跑到一半突然操作了另一台设备”。我的做法是设备列表和端口列表一一对应比如设备 A 用 4723设备 B 用 4725设备 C 用 4727以此类推。端口之间留出间隔避免冲突。启动 Appium server 的时候用-p参数指定端口-U参数指定设备 UDID。3. 核心细节拆解与实操要点3.1 环境准备清单在动手写代码之前先把环境理清楚。以下是我实际使用的版本组合实测稳定组件版本说明Python3.93.7 也能跑但 3.9 对多线程支持更好pytest7.x6.x 也可以7.x 的 fixture 更灵活Appium-Python-Client2.x注意 2.x 和 1.x 的 API 有差异Appium Server1.22建议用 Appium 1.22 或 2.0threading标准库不需要额外安装pytest-html最新出 HTML 报告用安装命令很简单pip install pytest Appium-Python-Client pytest-htmlAppium server 的安装这里不展开网上教程很多。重点提醒一句Appium server 的版本要和 Appium-Python-Client 的版本匹配2.x 的 client 连 1.x 的 server 可能会有兼容问题。3.2 设备信息配置的两种方式设备信息我建议单独抽成一个配置文件不要硬编码在脚本里。最直接的方式是用一个 Python 列表DEVICES [ { udid: emulator-5554, port: 4723, platform_version: 11.0, device_name: Pixel_4 }, { udid: emulator-5556, port: 4725, platform_version: 12.0, device_name: Pixel_5 }, ]如果设备多、经常变可以用 YAML 或 JSON 文件管理脚本启动时读取。我个人的习惯是用 YAML可读性好改起来方便。但要注意 YAML 的缩进很容易出错改完最好用yaml.safe_load验证一下。3.3 线程安全的 driver 管理这是整个方案里最关键的一环。绝对不能让多个线程共享同一个 driver 实例。我见过有人图省事全局定义一个 driver然后多个线程同时操作结果就是各种StaleElementReferenceException和NoSuchSessionException。正确的做法是每个线程独立创建 driver。用threading.local()可以做到线程级别的变量隔离import threading _thread_local threading.local() def get_driver(): if not hasattr(_thread_local, driver): _thread_local.driver create_driver() return _thread_local.driver这样每个线程调用get_driver()拿到的都是自己的 driver互不干扰。threading.local()的原理是每个线程有独立的命名空间同一个变量名在不同线程里指向不同的对象。3.4 Appium session 的创建参数创建 driver 的时候有几个参数必须注意from appium import webdriver from appium.options.android import UiAutomator2Options def create_driver(device): options UiAutomator2Options() options.platform_name Android options.platform_version device[platform_version] options.device_name device[device_name] options.udid device[udid] options.automation_name UiAutomator2 options.no_reset True # 关键不重置应用状态加快启动 options.new_command_timeout 300 # 命令超时设长一点避免并发时超时 driver webdriver.Remote( command_executorfhttp://127.0.0.1:{device[port]}/wd/hub, optionsoptions ) return driverno_resetTrue这个参数在多设备场景下特别重要。默认情况下每次新建 session 都会重置应用数据耗时很长。设成 True 之后应用保持上次的状态启动快很多。但要注意如果你的用例依赖干净的初始状态这个参数就不能随便开。new_command_timeout也要调大。并发的时候 Appium server 处理命令的速度会变慢默认 60 秒有时候不够设成 300 秒比较稳妥。4. 完整实操流程与核心代码实现4.1 项目目录结构先看整体结构这样你心里有个谱project/ ├── conftest.py # pytest 全局配置和 fixture ├── devices.yaml # 设备配置 ├── test_cases/ │ ├── test_login.py │ └── test_search.py ├── utils/ │ ├── driver_factory.py # driver 创建工厂 │ └── device_manager.py # 设备管理 ├── run_parallel.py # 并发执行入口 └── reports/ # 报告输出目录这个结构的好处是职责清晰设备配置、driver 创建、用例、执行入口各管各的改一处不影响其他地方。4.2 设备管理模块实现device_manager.py负责读取设备配置、分配设备给线程import yaml import threading class DeviceManager: def __init__(self, config_pathdevices.yaml): with open(config_path, r, encodingutf-8) as f: self.devices yaml.safe_load(f)[devices] self._lock threading.Lock() self._available list(self.devices) def acquire(self): with self._lock: if not self._available: return None return self._available.pop(0) def release(self, device): with self._lock: self._available.append(device)这里用threading.Lock()保证设备分配的原子性。如果不加锁两个线程可能同时判断_available非空然后同时 pop导致一个设备被分配给两个线程。这种 bug 很难复现但一旦出现就是灾难性的。4.3 并发执行入口run_parallel.py是核心负责启动多个线程每个线程跑一组用例import threading import pytest import sys from utils.device_manager import DeviceManager def run_tests_on_device(device, test_path): 在指定设备上运行测试 args [ test_path, f--device-udid{device[udid]}, f--device-port{device[port]}, -v, f--htmlreports/report_{device[udid]}.html, --self-contained-html ] pytest.main(args) def main(): manager DeviceManager() threads [] test_path sys.argv[1] if len(sys.argv) 1 else test_cases/ for _ in range(len(manager.devices)): device manager.acquire() if device is None: break t threading.Thread( targetrun_tests_on_device, args(device, test_path) ) threads.append(t) t.start() for t in threads: t.join() print(所有设备执行完毕) if __name__ __main__: main()这段代码的逻辑是有几台设备就起几个线程每个线程独立调用pytest.main()跑用例。注意pytest.main()是在当前进程内运行的多个线程同时调用它pytest 内部的状态可能会冲突。这是这个方案的一个隐患后面在常见问题里会详细说。4.4 conftest.py 中的 fixture 设计conftest.py是 pytest 的魔法文件fixture 都写在这里import pytest from utils.driver_factory import create_driver def pytest_addoption(parser): parser.addoption(--device-udid, actionstore, defaultNone) parser.addoption(--device-port, actionstore, defaultNone) pytest.fixture(scopesession) def device_config(request): udid request.config.getoption(--device-udid) port request.config.getoption(--device-port) return {udid: udid, port: int(port)} pytest.fixture(scopesession) def driver(device_config): d create_driver(device_config) yield d d.quit()这里 fixture 的 scope 设成session意味着同一个线程内的所有用例共享一个 driver。这样避免了每个用例都重建 session 的开销。yield之后的d.quit()保证测试结束后 driver 被正确关闭。4.5 用例编写示例用例本身不需要关心并发的事正常写就行import pytest from appium.webdriver.common.appiumby import AppiumBy class TestLogin: def test_login_success(self, driver): driver.find_element(AppiumBy.ID, com.example:id/username).send_keys(testuser) driver.find_element(AppiumBy.ID, com.example:id/password).send_keys(password123) driver.find_element(AppiumBy.ID, com.example:id/login_btn).click() assert driver.find_element(AppiumBy.ID, com.example:id/welcome).is_displayed() def test_login_fail(self, driver): driver.find_element(AppiumBy.ID, com.example:id/username).send_keys(wrong) driver.find_element(AppiumBy.ID, com.example:id/password).send_keys(wrong) driver.find_element(AppiumBy.ID, com.example:id/login_btn).click() assert driver.find_element(AppiumBy.ID, com.example:id/error_msg).is_displayed()用例里通过driverfixture 拿到当前线程的 driver完全感知不到并发的存在。这就是 fixture 设计的价值——把复杂性藏在框架层用例层保持干净。4.6 启动 Appium server 的脚本每台设备对应的 Appium server 要提前启动。可以写个 shell 脚本批量启动#!/bin/bash # start_appium.sh declare -A devices devices[emulator-5554]4723 devices[emulator-5556]4725 devices[emulator-5558]4727 for udid in ${!devices[]}; do port${devices[$udid]} appium -p $port -U $udid --log-level error echo Appium started on port $port for device $udid done wait--log-level error是为了减少日志输出并发的时候日志太多会拖慢速度。让每个 Appium server 在后台运行wait让脚本保持运行直到所有 server 退出。5. 常见问题与排查技巧实录5.1 pytest.main() 在多线程下的坑前面提到pytest.main()在多线程里调用有隐患。具体表现是有时候报告生成不完整有时候某个线程的用例结果跑到另一个线程的报告里去了。原因是 pytest 内部有一些全局状态多线程同时访问会冲突。解决方案有两个。第一个是用 subprocess 代替 threading每个设备起一个独立的 Python 进程跑 pytestimport subprocess def run_tests_on_device(device, test_path): cmd [ python, -m, pytest, test_path, f--device-udid{device[udid]}, f--device-port{device[port]}, f--htmlreports/report_{device[udid]}.html, --self-contained-html ] subprocess.run(cmd)这样每个 pytest 运行在独立进程里状态完全隔离报告也不会串。代价是内存开销大一点但稳定性提升明显。我现在生产环境用的就是这个方案。第二个方案是用pytest-xdist它本身就是为并行设计的-n auto自动根据 CPU 核数分配进程。但 xdist 的设备分配需要额外处理不如自己写 subprocess 灵活。5.2 设备掉线导致线程卡死多设备跑的时候最怕的就是某台设备突然掉线USB 松动、模拟器崩溃、手机锁屏。这时候对应的线程会卡在某个find_element上一直等到超时。如果超时设得很长整个并发任务就被这一个线程拖住了。我的处理方式是给每个线程加一个总超时。用threading.Thread的join(timeout)控制for t in threads: t.join(timeout1800) # 每个线程最多等 30 分钟 if t.is_alive(): print(f线程 {t.name} 超时强制结束)但join超时只是让主线程不再等子线程还在跑。要真正杀掉子线程得用 subprocess 方案然后process.kill()。这也是我推荐 subprocess 的另一个原因。5.3 端口被占用Appium server 启动失败最常见的原因就是端口被占用。可能是上次的 Appium 没退干净也可能是别的程序占了这个端口。排查方法# Linux/Mac lsof -i :4723 # Windows netstat -ano | findstr 4723找到占用进程后 kill 掉。预防措施是在启动脚本里加个检查端口被占用就跳过或换端口。5.4 常见问题速查表问题现象可能原因解决方法session 互相覆盖多线程共用一个 Appium server每台设备独立端口StaleElementReferenceExceptiondriver 被多线程共享用 threading.local 隔离报告内容错乱pytest.main 多线程冲突改用 subprocess线程卡死不退出设备掉线等待超时加 join timeout killAppium 启动失败端口被占用检查端口kill 占用进程用例执行越来越慢session 未释放内存泄漏确保 fixture 正确 quit部分设备无报告报告路径冲突报告文件名带设备 UDID5.5 几个我踩过的坑第一个坑driver.quit() 没放在 finally 里。有一次某个用例抛异常fixture 的 teardown 没执行driver 没关闭Appium server 上堆积了几十个僵尸 session后面新建 session 全部失败。后来我把 quit 放在 try/finally 里确保无论如何都执行。第二个坑设备名重复。两台设备如果device_name一样Appium 可能连错设备。一定要用udid来区分udid是设备的唯一标识不会重复。第三个坑并发数超过设备数。有次我起了 5 个线程但只有 3 台设备结果两个线程抢不到设备driver 创建失败。后来在代码里加了判断线程数严格等于设备数。第四个坑日志文件冲突。多个线程同时写同一个日志文件内容会交错。解决办法是每个线程写自己的日志文件文件名带设备 UDID。6. 性能优化与扩展思路6.1 并发数怎么定并发数不是越多越好。理论上并发数等于设备数但实际还要考虑机器的性能。每台 Appium server 大概占 200-500MB 内存每个模拟器占 1-2GB如果机器只有 8GB 内存跑 3 台模拟器就很吃力了。我的经验值是真机并发8GB 内存的机器最多跑 4 台模拟器并发16GB 内存最多跑 3 台。超过这个数机器开始频繁 swap反而更慢。可以用free -h或任务管理器观察内存使用留出 20% 的余量。6.2 用例分组策略如果用例很多可以考虑按模块分组不同设备跑不同模块。比如设备 A 跑登录相关用例设备 B 跑搜索相关用例设备 C 跑下单相关用例。这样每个设备的用例集更聚焦也方便定位问题。实现方式是在run_parallel.py里给每个设备指定不同的test_pathdevice_test_mapping { emulator-5554: test_cases/test_login.py, emulator-5556: test_cases/test_search.py, emulator-5558: test_cases/test_order.py, }这种方式的缺点是如果某个模块用例特别多那台设备就会成为瓶颈。所以分组的时候要尽量让各组的用例执行时间均衡。6.3 报告整合每个设备生成独立的 HTML 报告后可以再写个脚本把所有报告合并成一个总报告。简单的做法是用 pytest-html 的--report-log参数生成 JSON 日志然后用脚本读取 JSON 生成汇总报告。或者直接用 allureallure 天然支持多份结果合并# 每个设备生成 allure 结果到不同目录 pytest --alluredirallure-results/device1 pytest --alluredirallure-results/device2 # 合并生成报告 allure generate allure-results -o allure-reportallure 的报告比 pytest-html 好看很多而且支持历史趋势、失败重试等高级功能推荐用这个。6.4 后续可以扩展的方向这套框架跑通之后可以往几个方向扩展。一是接入 CI/CD在 Jenkins 或 GitLab CI 里配置多设备并发任务每次提交代码自动触发。二是加失败重试用pytest-rerunfailures插件用例失败自动重跑减少误报。三是加截图和录屏用例失败时自动截图方便排查问题。四是设备健康检查跑之前先检查设备是否在线、Appium 是否正常避免跑到一半才发现设备有问题。我个人在实际操作中的体会是多设备并发这件事框架搭好之后维护成本其实很低真正花时间的是前期的调试和踩坑。把设备分配、driver 隔离、端口规划这三件事做扎实后面基本就是一劳永逸。另外强烈建议用 subprocess 方案而不是 threading 方案虽然多占点内存但稳定性完全不是一个级别省下来的排查时间远比那点内存值钱。
阅读完成 · 觉得有帮助？
