如何在两台真实Android设备上并行执行Appium消息收发自动化测试并解决单设备执行问题

嘿，我来帮你搞定这个并行执行的问题！你现在遇到的核心问题是代码默认是单线程跑的，所以只会按顺序处理一台设备的操作，甚至可能因为驱动实例的覆盖直接跳过了另一台。要让两台真机同时跑测试，最常用的就是线程或者测试框架的并行插件，我给你一步步讲清楚怎么弄：

先做好前置准备 #

在写代码之前，先把基础配置搞定，不然容易踩坑：

• 分别启动两个独立的Appium服务，绑定不同端口和设备：
  • 给设备1启动服务：appium -p 4723 -U <设备1的UDID>
  • 给设备2启动服务：appium -p 4725 -U <设备2的UDID>
    （用adb devices命令可以查到两台设备的UDID，加-U参数能让每个Appium服务只对应一台设备，避免设备混淆）
• 确保两台设备的Desired Capabilities里，systemPort字段设置不同的值（比如8200和8201），UiAutomator2需要用这个端口和设备通信，重复的话会直接报错。

方案一：用Python Threading手动实现并行 #

这是最直接的方法，把单设备的测试逻辑封装成函数，然后用线程同时启动两台设备的测试。

完整代码示例 #

from appium import webdriver
from appium.options.android import UiAutomator2Options
from appium.webdriver.common.appiumby import AppiumBy
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import threading

# 把单设备的测试逻辑封装成独立函数
def run_device_test(device_cap, appium_port):
    # 初始化当前设备的驱动
    driver = webdriver.Remote(
        command_executor=f'http://localhost:{appium_port}/wd/hub',
        options=device_cap
    )
    
    try:
        udid = device_cap.get_capability("udid")
        if udid == "你的设备1UDID":
            # Device1的操作：打开APP、输入内容、发送消息
            print(f"[{udid}] 开始执行消息发送操作...")
            # 这里替换成你实际的元素定位和操作
            # 比如找到输入框并输入测试消息
            input_box = WebDriverWait(driver, 10).until(
                EC.presence_of_element_located((AppiumBy.ID, "com.your.app:id/input_field"))
            )
            input_box.send_keys("这是一条Appium测试消息")
            # 点击发送按钮
            send_btn = driver.find_element(AppiumBy.ID, "com.your.app:id/send_btn")
            send_btn.click()
            print(f"[{udid}] 消息发送完成")
        else:
            # Device2的操作：等待接收消息并验证
            print(f"[{udid}] 等待接收消息...")
            # 显式等待消息元素出现，验证内容
            received_msg = WebDriverWait(driver, 15).until(
                EC.presence_of_element_located((AppiumBy.ID, "com.your.app:id/latest_message"))
            )
            assert received_msg.text == "这是一条Appium测试消息", f"[{udid}] 接收到的消息内容不符合预期！"
            print(f"[{udid}] 消息接收验证成功")
    except Exception as e:
        print(f"[{device_cap.get_capability('udid')}] 测试出错：{str(e)}")
    finally:
        # 不管成功失败，都要关闭驱动
        driver.quit()

if __name__ == "__main__":
    # 配置设备1的Desired Capabilities
    device1_cap = UiAutomator2Options().load_capabilities({
        "platformName": "Android",
        "appium:udid": "你的设备1UDID",
        "appium:deviceName": "Device1",
        "appium:appPackage": "com.your.app",
        "appium:appActivity": "com.your.app.MainActivity",
        "appium:systemPort": 8200,
        "appium:automationName": "UiAutomator2",
        "appium:noReset": True  # 可选，避免每次测试重置APP状态
    })

    # 配置设备2的Desired Capabilities
    device2_cap = UiAutomator2Options().load_capabilities({
        "platformName": "Android",
        "appium:udid": "你的设备2UDID",
        "appium:deviceName": "Device2",
        "appium:appPackage": "com.your.app",
        "appium:appActivity": "com.your.app.MainActivity",
        "appium:systemPort": 8201,
        "appium:automationName": "UiAutomator2",
        "appium:noReset": True
    })

    # 创建两个线程，分别对应两台设备的测试
    thread_device1 = threading.Thread(target=run_device_test, args=(device1_cap, 4723))
    thread_device2 = threading.Thread(target=run_device_test, args=(device2_cap, 4725))

    # 启动两个线程
    thread_device1.start()
    thread_device2.start()

    # 等待两个线程都执行完成
    thread_device1.join()
    thread_device2.join()

    print("两台设备的并行测试全部执行完毕！")

这个方案的关键注意点 #

• 每个线程必须用独立的驱动实例：绝对不能在多个线程间共享同一个driver，不然会出现各种混乱的操作
• 用显式等待代替time.sleep：WebDriverWait比固定的睡眠更可靠，能根据元素的实际状态等待，避免因为设备性能差异导致的测试失败
• 线程安全问题：如果你的测试逻辑里用到了共享的全局变量，记得用threading.Lock()加锁，不过如果只是各自独立的设备操作，一般不需要

方案二：用Pytest-xdist实现并行（更适合工程化项目） #

如果你的测试是用pytest组织的，那可以直接用pytest-xdist插件，不用自己写线程逻辑，更简洁。

实现步骤 #

1. 安装插件：pip install pytest-xdist
2. 把两台设备的配置写成pytest的fixture，比如在conftest.py里：

import pytest
from appium import webdriver
from appium.options.android import UiAutomator2Options

@pytest.fixture(params=[
    {"port": 4723, "udid": "设备1UDID", "systemPort": 8200},
    {"port": 4725, "udid": "设备2UDID", "systemPort": 8201}
])
def driver(request):
    cap = UiAutomator2Options().load_capabilities({
        "platformName": "Android",
        "appium:udid": request.param["udid"],
        "appium:deviceName": f"Device_{request.param['udid'][-4:]}",
        "appium:appPackage": "com.your.app",
        "appium:appActivity": "com.your.app.MainActivity",
        "appium:systemPort": request.param["systemPort"],
        "appium:automationName": "UiAutomator2"
    })
    driver = webdriver.Remote(
        command_executor=f'http://localhost:{request.param["port"]}/wd/hub',
        options=cap
    )
    yield driver
    driver.quit()

3. 写两个测试函数，分别对应发送和接收（或者用参数化区分）：

def test_device1_send(driver):
    # Device1的发送逻辑，判断udid后执行
    if driver.capabilities["udid"] == "设备1UDID":
        # 发送消息操作
        pass

def test_device2_receive(driver):
    # Device2的接收逻辑
    if driver.capabilities["udid"] == "设备2UDID":
        # 接收验证操作
        pass

4. 用并行命令启动测试：pytest -n 2（-n 2表示用2个进程并行执行）

最后再给你提几个小建议 #

• 启动Appium服务时，可以加上--log-level info参数，方便排查连接问题
• 测试前确保两台设备的APP版本一致，避免因为UI差异导致定位失败
• 如果遇到设备连接不稳定的情况，可以试试重启adb服务：adb kill-server && adb start-server

备注：内容来源于stack exchange，提问作者sivapriya vimalkumar