1. 多设备并发跑Appium,为什么值得认真折腾
做过移动端自动化的人都有一个共同的痛:一台机器、一根USB线、一个模拟器,跑完一轮回归测试要等十几分钟甚至半小时。如果手上有三台真机、两个模拟器,还按串行的方式一台一台跑,那基本上一整个下午就耗在等测试结果上了。pytest学习(六) - 多设备并发appium+pytest多线程这个主题,解决的就是这个非常具体的效率问题——让多台设备同时跑测试用例,把原本串行几十分钟的任务压缩到几分钟内完成。
这篇文章适合两类人看。一类是已经写过一些Appium脚本、用pytest组织过测试用例,但还没尝试过多设备并发的同学;另一类是用过并发但踩了坑,比如设备抢端口、用例互相干扰、日志混在一起分不清是谁输出的,想找一套稳定可复现方案的同学。我会从整体设计思路讲起,把多线程模型、设备分配策略、端口管理、用例隔离、日志分离这些关键环节全部拆开,配上可以直接抄的代码结构和参数配置,最后再把我自己踩过的坑整理成一张速查表。
需要提前说明的是,多设备并发不是简单加个ThreadPoolExecutor就完事了。Appium本身是一个C/S架构的服务,每个设备需要独立的Appium Server实例、独立的系统端口、独立的设备UDID绑定,再加上pytest本身的用例收集机制、fixture作用域、报告生成方式,全都要重新考虑。任何一个环节没处理好,结果就是设备之间互相抢占资源,测试结果不可信,排查问题比串行还累。所以下面我会把“为什么这么设计”讲透,而不只是丢一段代码出来。
2. 整体架构设计与方案选型思路
2.1 串行 vs 多线程 vs 多进程,到底选哪个
在动手之前,先要把并发模型选清楚。Python里做并发常见的有三种:多线程、多进程、协程。放到Appium这个场景里,协程基本可以排除,因为Appium客户端库(Appium-Python-Client)底层是同步的HTTP请求,硬套asyncio只会增加复杂度。真正需要权衡的是多线程和多进程。
多进程的优势是每个进程有独立的内存空间和GIL,不会因为Python的全局解释器锁互相影响。但代价是进程间通信麻烦,设备分配、结果汇总、日志收集都要额外做IPC,而且每个进程重新导入pytest、重新初始化环境,启动开销大。多线程的优势是共享内存、启动快、代码改动小,缺点是GIL。但这里有个关键点:Appium测试的瓶颈在等待设备响应,属于IO密集型任务,不是CPU密集型。线程在等待HTTP响应时会释放GIL,所以多线程完全能跑满多台设备的并发,GIL根本不是瓶颈。
我实测下来,4台设备用多线程并发,CPU占用率不到30%,主要时间都花在WebDriverWait等待元素上。所以结论很明确:IO密集型场景,多线程是性价比最高的选择。除非你要在测试里做大量图像比对、OCR识别这类CPU密集操作,那才需要考虑多进程。
2.2 每个设备一套独立Appium Server的必要性
很多人第一反应是:能不能一个Appium Server管多台设备?答案是可以启动,但强烈不建议。Appium Server虽然支持通过--udid指定设备,但一个Server实例同时处理多个session时,端口复用、日志混杂、session管理都会变得很脆弱。一旦某个session异常退出,可能影响同Server下的其他设备。
更稳妥的做法是一台设备对应一个Appium Server实例,每个实例绑定不同的端口。比如设备A用4723,设备B用4725,设备C用4727。这样每个Server的日志独立、session独立、生命周期独立,一台设备崩了不影响其他设备。代价是要多占几个端口和一点内存,但换来的稳定性完全值得。
端口分配我一般用4723 + index * 2这个公式,留出间隔是为了避免端口冲突,也方便后续扩展。设备信息我习惯放在一个YAML或JSON配置文件里,包含udid、platformVersion、port、systemPort这几个字段。其中systemPort是Android特有的,用于UiAutomator2驱动和设备通信,多设备并发时如果不指定,默认都是8200,必然冲突。
2.3 pytest的并发插件选型:pytest-xdist还是自己写线程池
提到pytest并发,很多人会想到pytest-xdist。它确实能并行跑用例,但它的模型是“把用例分发到多个worker进程”,每个worker是独立进程,设备分配需要靠--dist loadscope之类的策略,而且worker和设备的映射关系不好精确控制。对于“每个设备跑一套完整用例”这种需求,xdist反而绕。
我的做法是自己用concurrent.futures.ThreadPoolExecutor管理设备级并发,每个线程内部用pytest的编程式调用或者直接调用测试函数。这样设备分配完全可控,一个线程绑定一台设备,跑完这台设备的所有用例再释放。如果你坚持要用xdist,也不是不行,但需要配合pytest-xdist的worker_id来做设备映射,配置起来更绕,排查问题也更麻烦。
这里有个折中方案值得提一下:用pytest的pytest_generate_tests钩子做参数化,把设备列表作为参数传进去,再配合xdist的--dist loadfile。但实测下来,设备级并发的粒度用线程池更直观,代码可读性也更好。下面我就按线程池方案展开。
3. 核心细节拆解:设备管理、端口分配与用例隔离
3.1 设备配置文件的组织方式
设备信息不要硬编码在代码里,这是铁律。我一般建一个devices.yaml,结构大概是这样:
devices: - udid: "emulator-5554" platformName: "Android" platformVersion: "13" deviceName: "Pixel_5_API_33" port: 4723 systemPort: 8200 - udid: "emulator-5556" platformName: "Android" platformVersion: "12" deviceName: "Pixel_4_API_31" port: 4725 systemPort: 8201 - udid: "real-device-001" platformName: "Android" platformVersion: "14" deviceName: "Xiaomi_14" port: 4727 systemPort: 8202读取的时候用PyYAML加载,然后按需过滤。比如只想跑前两台,就在加载后切片。这样做的好处是新增设备只改配置文件,代码一行不动。systemPort一定要手动指定且互不相同,这是Android多设备并发最容易忽略的坑,默认值冲突会导致UiAutomator2启动失败,报错信息还特别隐晦,经常是“cannot connect to system port”之类,新手很难定位。
3.2 Appium Server的启动与销毁时机
每个设备对应的Appium Server,启动和销毁的时机很关键。我的做法是在每个线程内部启动Server,跑完这台设备的所有用例后销毁。这样Server的生命周期和设备线程完全绑定,不会出现Server残留或者端口被占用的情况。
启动命令用subprocess.Popen,把stdout和stderr重定向到独立的日志文件,方便排查。命令大概长这样:
appium --port 4723 --udid emulator-5554 --log devices/emulator-5554.log --log-level info注意--log参数指定日志文件,多设备并发时每个Server的日志必须分开,否则日志混在一起根本没法看。启动后不要立刻发请求,要轮询http://127.0.0.1:4723/status直到返回ready,或者简单点用time.sleep等3到5秒。我倾向于轮询status接口,更可靠,避免设备慢的时候Server还没起来就发请求导致连接失败。
销毁的时候用process.terminate(),然后process.wait(timeout=10),如果超时再kill()。这里有个细节:Windows下terminate有时候杀不干净node进程,建议用taskkill /F /T /PID,Linux/Mac下terminate一般够用。跑完一轮测试后,最好再检查一下端口是否释放,避免下一轮启动时端口被占。
3.3 用例隔离:为什么每个设备要独立的数据和账号
多设备并发最容易翻车的地方不是技术,而是测试数据冲突。假设你的用例是“注册一个新用户”,三台设备同时跑,如果都用同一个手机号,必然有两台失败。所以并发跑之前,必须保证每台设备用的测试数据是隔离的。
我的做法是给每台设备分配一个device_index,然后测试数据里带上这个index。比如账号用test_user_{device_index}@example.com,订单号用order_{device_index}_{timestamp}。这样即使三台设备同时操作同一套后端,数据也不会撞。如果后端有唯一性约束,这个隔离是必须的。
另一个隔离点是App的状态。每台设备在跑用例前,最好用driver.reset()或者adb shell pm clear把App数据清掉,保证每台设备从干净状态开始。否则上一轮残留的登录态、缓存数据会影响结果。noReset这个capability在多设备并发时建议设为false,除非你明确知道自己在做什么。
3.4 日志与报告的分离策略
多设备并发时,日志如果不分离,排查问题就是灾难。我的做法是三层日志分离:
第一层是Appium Server日志,按设备UDID命名,存在logs/appium_{udid}.log。第二层是测试执行日志,用Python的logging模块,每个线程创建一个独立的Logger实例,handler指向logs/test_{udid}.log。第三层是pytest的报告,如果要用Allure,每个设备的报告生成到独立的allure-results/{udid}/目录,最后再合并。
这里有个技巧:logging是线程安全的,但如果你用同一个Logger实例往同一个文件写,多线程下内容会交错。所以每个线程必须用独立的Logger名字,比如logger = logging.getLogger(f"device_{udid}"),并且propagate设为False,避免日志冒泡到root logger导致重复输出。
4. 实操过程:从零搭一套多设备并发框架
4.1 环境准备与依赖安装
先把依赖装齐。核心的几个包:
pip install pytest appium-python-client PyYAML allure-pytestAppium Server本身需要Node环境,用npm install -g appium装。Android环境需要adb、ANDROID_HOME配好,adb devices能列出所有设备。这些是前置条件,不展开讲,假设你已经能单设备跑通Appium脚本。
验证环境是否OK,先手动启动一个Appium Server,用单设备跑一个最简单的脚本,确认能打开App、找到元素。这一步不能跳过,多设备并发是建立在单设备稳定的基础上的。如果单设备都不稳,并发只会让问题放大十倍。
4.2 设备管理类的实现
我写了一个DeviceManager类,负责加载配置、分配设备、启动和停止Server。核心方法有三个:load_devices()、start_appium(device)、stop_appium(device)。代码结构大概这样:
import yaml import subprocess import time import requests class DeviceManager: def __init__(self, config_path="devices.yaml"): with open(config_path, "r", encoding="utf-8") as f: self.devices = yaml.safe_load(f)["devices"] def start_appium(self, device): cmd = [ "appium", "--port", str(device["port"]), "--udid", device["udid"], "--log", f"logs/appium_{device['udid']}.log", "--log-level", "info" ] proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) self._wait_ready(device["port"]) return proc def _wait_ready(self, port, timeout=30): url = f"http://127.0.0.1:{port}/status" start = time.time() while time.time() - start < timeout: try: resp = requests.get(url, timeout=2) if resp.status_code == 200: return True except requests.RequestException: pass time.sleep(1) raise RuntimeError(f"Appium on port {port} not ready") def stop_appium(self, proc): proc.terminate() try: proc.wait(timeout=10) except subprocess.TimeoutExpired: proc.kill()_wait_ready这个轮询很关键,不要用固定sleep。我试过固定等5秒,结果在慢设备上Server还没起来,脚本就发请求了,报连接拒绝。轮询status接口虽然多几行代码,但稳定性提升明显。
4.3 线程池驱动的并发执行
主入口用ThreadPoolExecutor,每个设备一个任务:
from concurrent.futures import ThreadPoolExecutor, as_completed def run_device(device, test_cases): dm = DeviceManager() proc = dm.start_appium(device) try: driver = create_driver(device) for case in test_cases: case(driver, device) finally: driver.quit() dm.stop_appium(proc) def main(): dm = DeviceManager() devices = dm.devices with ThreadPoolExecutor(max_workers=len(devices)) as executor: futures = {executor.submit(run_device, d, TEST_CASES): d for d in devices} for future in as_completed(futures): device = futures[future] try: future.result() print(f"{device['udid']} passed") except Exception as e: print(f"{device['udid']} failed: {e}")max_workers设成设备数量,不要设太大。设大了线程会排队等设备,反而增加调度开销。as_completed用来收集结果,哪个设备先跑完先处理,不用等所有设备都结束。
create_driver里要注意capability的配置,systemPort必须从device配置里取,udid也要指定,否则Appium可能连错设备。newCommandTimeout建议设长一点,比如300秒,避免用例执行时间长导致session超时。
4.4 参数化与用例组织的配合
如果你用pytest组织用例,有两种方式接入。一种是每个设备线程内部用pytest.main()编程式调用,传不同的--alluredir。另一种是把测试逻辑写成普通函数,线程直接调用,不走pytest的收集机制。前者能复用pytest的fixture和报告,后者更轻量。
我倾向于前者,因为pytest的fixture在管理driver生命周期上很方便。具体做法是每个线程调用pytest.main(["-s", "-v", "tests/", f"--alluredir=allure-results/{udid}"]),然后通过环境变量或者pytest_generate_tests把设备信息传进去。环境变量最简单,线程启动前os.environ["DEVICE_UDID"] = udid,fixture里读这个变量创建driver。
这里有个坑:pytest.main()在同一进程里多次调用,pytest的插件状态可能残留。实测下来,只要不在同一个进程里并发调用pytest.main(),而是每个线程独立调用,一般没问题。但如果遇到奇怪的插件报错,可以考虑用subprocess起独立进程跑pytest,代价是启动慢一点。
5. 常见问题与排查技巧实录
5.1 设备抢端口、抢systemPort的典型表现
最常见的报错是UiAutomator2 failed to start或者cannot bind to system port 8200。原因就是多台设备的systemPort没区分。解决办法前面说了,配置文件里手动指定,每台设备一个。另外port也要区分,Appium Server的端口冲突会直接导致启动失败,报EADDRINUSE。
还有一个隐蔽的坑:adb的端口转发。Appium启动UiAutomator2时会用adb forward做端口转发,如果多台设备同时操作,偶尔会出现转发规则冲突。这种情况重启adb server(adb kill-server && adb start-server)通常能解决,但根治办法还是保证systemPort唯一。
5.2 用例互相干扰的排查思路
如果发现某台设备的用例结果不稳定,时好时坏,优先怀疑数据冲突。排查方法是把并发改成串行,如果串行稳定、并发不稳定,基本就是数据隔离没做好。检查点包括:登录账号是否唯一、订单号是否带设备标识、后端是否有全局锁、App本地存储是否被其他设备影响(这个一般不会,但如果用了共享的测试账号,服务端session可能互相踢)。
另一个干扰源是时间。如果用例依赖“当前时间”,多台设备同时跑,时间戳可能相同,导致数据撞。解决办法是在时间戳后面加设备index或者随机数。
5.3 日志混杂、报告覆盖的处理
日志混杂的根源是多个线程写同一个文件。解决办法是每个线程独立Logger、独立文件。Allure报告覆盖的根源是多个线程往同一个allure-results目录写,文件名冲突。解决办法是每个设备一个子目录,最后用allure generate合并。
如果不想用Allure,pytest自带的--junitxml也可以,每个设备生成一个xml,最后用工具合并。我一般用Allure,因为报告更直观,失败用例的截图和日志都能附上。
5.4 常见问题速查表
| 问题现象 | 可能原因 | 解决办法 |
|---|---|---|
| Appium启动报EADDRINUSE | 端口被占用 | 检查port配置,确保唯一;杀残留node进程 |
| UiAutomator2启动失败 | systemPort冲突 | 配置文件中每台设备指定不同systemPort |
| 用例结果不稳定 | 测试数据冲突 | 数据带设备index,账号/订单号隔离 |
| 日志内容交错 | 多线程写同一文件 | 每线程独立Logger和文件 |
| Allure报告被覆盖 | 多线程写同一目录 | 每设备独立alluredir,最后合并 |
| 设备连错 | udid未指定或错误 | capability中明确指定udid |
| session超时 | newCommandTimeout太短 | 设为300秒或更长 |
| Server未就绪就发请求 | 固定sleep不够 | 轮询/status接口直到ready |
5.5 几个我踩过的坑
第一个坑是noReset。一开始为了加快速度设了noReset=true,结果多设备并发时,App的登录态互相影响,因为测试账号是同一个。后来改成noReset=false,每台设备跑前清数据,问题解决。代价是每轮多花十几秒清数据,但结果可靠多了。
第二个坑是Appium Server的日志级别。默认info级别日志量很大,多设备并发时磁盘IO压力不小。后来改成warn,只在出问题时才调回info。日志文件也要定期清理,不然跑几天磁盘就满了。
第三个坑是线程池的异常处理。一开始没在run_device里加try/finally,结果某个设备用例失败抛异常,driver没quit,Appium Server没停,端口一直占着,下一轮直接启动失败。加上finally后,无论用例成功失败,资源都能释放。
第四个坑是adb的并发。多台设备同时执行adb命令时,偶尔会卡住。后来发现是adb server的单线程模型导致的,解决办法是尽量少用adb命令,能用Appium API就用API。如果必须用,加超时和重试。
6. 性能调优与扩展思路
6.1 并发数量的合理上限
不是设备越多越好。我实测下来,一台普通开发机(8核16G)跑4到6台设备比较稳,再多就会出现明显的资源竞争,CPU和内存都吃紧,反而拖慢整体速度。上限主要受限于内存,每个Appium Server大概占200到300MB,每个模拟器占1到2G,真机占用少一些。所以并发数量要根据机器配置来定,不要盲目堆设备。
如果确实需要跑更多设备,可以考虑分布式,把设备分散到多台机器上,每台机器跑一部分,最后汇总结果。这就涉及到更复杂的调度,一般团队用不上,除非是大型回归测试。
6.2 用例粒度的优化
多设备并发时,用例粒度太细会导致频繁的driver创建和销毁,开销大。粒度太粗又会导致单台设备跑太久,并发优势发挥不出来。我的经验是每个设备的用例集控制在5到15分钟能跑完,太短了启动开销占比高,太长了整体等待时间长。
另外,把不依赖设备的用例(比如纯接口测试、数据准备)抽出来,不要放在设备线程里跑。设备线程只跑必须真机/模拟器执行的UI用例,这样能最大化并发效率。
6.3 后续可扩展的方向
这套框架搭好后,可以往几个方向扩展。一是接入CI,每次提交代码自动触发多设备回归,结果推送到群里。二是加失败重试,某台设备某个用例失败后自动重跑一次,减少偶发失败带来的误报。三是加设备健康检查,跑之前先确认设备在线、Appium能启动,避免跑到一半才发现设备掉线。
还有一个方向是动态设备分配。现在是配置文件写死设备,如果设备池是动态的(比如云真机平台),可以改成从接口拉设备列表,动态分配端口和systemPort。这个改动不大,主要是把配置来源从文件换成接口。
我个人在实际操作中的体会是,多设备并发最难的从来不是写并发代码,而是把设备隔离、数据隔离、日志隔离这三件事做扎实。代码本身几十行就能跑起来,但要让结果稳定可信,需要在细节上反复打磨。我建议第一次搭的时候先用两台设备,把流程跑通、把坑踩完,再逐步加设备。一上来就上六台,出了问题根本不知道是哪台的锅。