如何用Python高效并行控制多台Android设备？

我正在开发一个需要用Python脚本同时控制多台Android设备的项目，测试阶段先用Android Studio模拟器，后续会接入真机。目前卡在多线程有效控制这一步，调研过Appium、Selenium Server等方案，但找不到匹配需求的清晰资料和简洁流程。

试过ADB和Appium，但每次只能控制一台设备；用Appium+Selenium Server的实验也困难重重，始终没找到明确的操作指引。现在项目进度受阻，希望能得到简化Python并行控制Android模拟器流程的建议，每台模拟器要执行脚本指定的不同流程，完全可以更换现有工作流。

以下是我尝试用ADB获取已连接设备，并为每台设备启动Appium服务器的代码：

# Function to get connected devices
def get_connected_devices():

    devices_configs = []
    try:
        output = subprocess.check_output([adb_path, 'devices']).decode('utf-8')
        devices = output.strip().split('\n')[1:]


        for device in devices:
            ud_id = device.split('\t')[0]
            os_version = (
                subprocess.check_output(
                    [
                        adb_path,
                        '-s',
                        ud_id,
                        'shell',
                        'getprop',
                        'ro.build.version.release',
                    ]
                )
                .decode('utf-8')
                .strip()
            )
            platform = "Android"
            system_port = str(8200 + len(devices_configs))
            chrome_driver_port = str(8100 + len(devices_configs))
            device_details = {
                "device": ud_id,
                "os_version": os_version,
                "ud_id": ud_id,
                "platform": platform,
                "systemPort": system_port,
            }
            devices_configs.append(device_details)
        return devices_configs
    except subprocess.CalledProcessError as e:
        print(f"Error executing ADB command: {e}")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")

# Retrieve connected devices
connected_devices = get_connected_devices()

# Start Appium servers for each device
processes = []
for i in range(len(connected_devices)):
    port = 4724 + i
    cmd = ["start", "appium", "-p", str(port)]
    process = subprocess.Popen(cmd, shell=True)
    processes.append(process)

尽管如此，我仍遇到无法强制Appium使用指定设备端口的问题，它始终为所有设备使用同一端口：

# Pair each connected device with a server
device_server_pairs = list(zip(connected_devices, servers))

# Start Appium sessions using threads
with concurrent.futures.ThreadPoolExecutor() as executor:
    executor.map(start_appium_session, device_server_pairs)

# Function to start Appium session for each device
def start_appium_session(device_config_server_tuple):
    device_config, server = device_config_server_tuple
    capabilities = {
        'platformName': device_config['platform'],
        'platformVersion': device_config['os_version'],
        'deviceName': device_config['ud_id'],
        'systemPort': device_config['systemPort'],
        'app': APP_PATH,
    }

    driver = webdriver.Remote(server, options=UiAutomator2Options().load_capabilities(capabilities))

核心问题修复与优化方案 #

1. 解决Appium端口冲突与设备绑定问题 #

你当前启动Appium服务器时只指定了主端口（-p），但未绑定设备UDID，导致服务器无法区分设备。同时UiAutomator2依赖的systemPort需确保唯一且正确传递，还要给每个Appium实例分配独立的--bootstrap-port避免引擎端口冲突。

修改启动Appium服务器的代码：

# 启动Appium服务器并绑定对应设备
processes = []
servers = []  # 存储每个服务器的访问地址
for idx, device in enumerate(connected_devices):
    port = 4724 + idx
    bootstrap_port = 4724 + 100 + idx  # 为每个设备分配独立的bootstrap端口
    udid = device['ud_id']
    # 命令添加--udid绑定设备，--bootstrap-port避免UiAutomator2冲突
    cmd = [
        "appium", 
        "-p", str(port), 
        "--udid", udid,
        "--bootstrap-port", str(bootstrap_port),
        "--session-override"  # 允许覆盖旧会话，避免残留问题
    ]
    # 不用shell=True，避免跨平台兼容问题，同时捕获输出方便调试
    process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    processes.append(process)
    servers.append(f"http://localhost:{port}/wd/hub")  # 记录服务器地址

2. 完善Capabilities配置 #

systemPort需为整数类型（你之前存的是字符串，可能导致识别失败），同时显式指定自动化引擎为UiAutomator2：

def start_appium_session(device_config_server_tuple):
    device_config, server = device_config_server_tuple
    capabilities = {
        'platformName': device_config['platform'],
        'platformVersion': device_config['os_version'],
        'deviceName': device_config['ud_id'],
        'udid': device_config['ud_id'],  # 显式指定UDID，消除歧义
        'automationName': 'UiAutomator2',  # 必须指定，否则可能默认使用旧引擎
        'systemPort': int(device_config['systemPort']),  # 转换为整数类型
        'app': APP_PATH,
        'noReset': True  # 根据需求选择，避免每次启动重装APP
    }

    options = UiAutomator2Options()
    options.load_capabilities(capabilities)
    try:
        driver = webdriver.Remote(server, options=options)
        # 这里添加对应设备的专属业务逻辑，比如不同的操作流程
        run_device_specific_flow(driver, device_config)
        driver.quit()  # 结束会话时主动退出，释放资源
    except Exception as e:
        print(f"设备 {device_config['ud_id']} 会话启动失败: {e}")

3. 多线程执行与资源管理 #

• 使用ThreadPoolExecutor是合理的，但要确保每个线程的业务逻辑隔离，比如定义run_device_specific_flow函数，根据设备标识执行不同流程。
• 添加进程清理逻辑，在脚本结束时关闭所有Appium进程，避免残留：

# 脚本结束时清理Appium进程
def cleanup_appium_processes(processes):
    for proc in processes:
        proc.terminate()
        proc.wait()

# 主流程中添加异常捕获与清理
try:
    with concurrent.futures.ThreadPoolExecutor() as executor:
        executor.map(start_appium_session, zip(connected_devices, servers))
finally:
    cleanup_appium_processes(processes)

4. 规模化场景替代方案：Appium Grid #

如果后续要接入大量真机，Appium Grid是更规范的选择：

• 启动一个Hub节点作为调度中心
• 为每个设备启动一个Node节点，绑定UDID和端口
• Python脚本通过Hub地址发送请求，Grid会自动将任务分配到对应设备

不过对于小规模模拟器，单服务器绑定设备的方案更简洁高效。

内容的提问来源于stack exchange，提问作者iamarthur