最近在技术社区看到不少关于“云手机”的讨论很多开发者、测试同学甚至普通用户都对如何低成本、自动化地使用云手机资源感兴趣。无论是用于自动化测试、多开应用、还是作为临时的远程开发环境手动管理云手机实例既繁琐又昂贵。本文将从一个技术实现的角度深入探讨如何利用开源工具和脚本构建一套高度自动化的云手机资源管理与调度系统。我们将从核心概念讲起逐步拆解环境搭建、核心脚本编写、自动化流程设计并最终给出一个可运行的原型示例。无论你是想学习自动化运维还是为团队搭建测试云手机池这篇文章都能提供清晰的路径和可复用的代码。1. 云手机自动化管理的核心概念与价值在深入技术细节之前我们首先要明确几个关键概念和为什么要做自动化管理。云手机Cloud Phone并非指某个具体品牌而是一种服务模式。它本质上是在云端服务器上运行的虚拟安卓或其它移动操作系统实例用户可以通过网络远程访问其图形界面并进行操作就像在使用一台真实的手机。常见的应用场景包括移动应用自动化测试在多种机型、系统版本上并行执行测试用例。应用多开与挂机运行需要长期在线的应用如游戏、社交软件。安全沙箱在隔离环境中运行可能存在风险的应用。远程演示与开发作为临时的移动端开发或演示环境。“自动化”与“资源管理”是本文的核心。手动在云服务商控制台创建、开机、连接、销毁云手机效率极低且容易出错。自动化的目标是通过程序脚本来完成这些生命周期操作并实现诸如“按需创建、定时释放”、“异常自动重启”、“批量初始化配置”等功能。这对于控制成本避免云手机24小时运行产生高额费用和提升效率至关重要。技术栈选择我们将主要使用Python作为自动化脚本语言因为它拥有丰富的库来操作云服务商API如阿里云、腾讯云、华为云等、处理HTTP请求以及执行系统命令。同时我们会用到ADBAndroid Debug Bridge这个安卓调试神器来与云手机实例进行交互安装APK、执行命令、截图等。整个系统的灵魂是一套调度策略它决定了何时创建、何时销毁手机。2. 环境准备与工具安装在开始编写自动化脚本前你需要准备好本地或跳板机环境。2.1 基础环境操作系统推荐 Linux (如 Ubuntu 20.04) 或 macOSWindows也可但部分命令可能略有不同。Python版本 3.8 及以上。确保已安装pip。版本控制Git可选但推荐用于管理脚本。2.2 必需工具安装安装Python依赖库我们将使用requests调用云API使用schedule进行定时任务管理或使用系统cron。pip install requests schedule如果生产环境使用建议将依赖写入requirements.txt文件。安装并配置ADBLinux/macOS: 通常可通过包管理器安装如sudo apt-get install android-tools-adb。Windows: 从 Android开发者官网 下载Platform-Tools解压并将目录添加到系统PATH环境变量。验证安装打开终端输入adb version应能显示版本信息。准备云服务商账号与凭证本文以阿里云的“弹性云手机”CPH服务为例进行演示其他云厂商腾讯云、华为云原理类似。你需要在阿里云控制台开通CPH服务。获取关键的API访问凭证AccessKey ID和AccessKey Secret。请妥善保管切勿泄露。了解你所在的地域RegionId例如cn-hangzhou。在CPH控制台创建一个服务器镜像包含你需要的预装App和配置并记录其ImageId。2.3 项目结构规划创建一个清晰的项目目录便于管理cloud_phone_manager/ ├── config/ # 配置文件目录 │ └── config.yaml # 存放AK/SK、地域等敏感和通用配置 ├── src/ # 源代码目录 │ ├── cloud_api.py # 封装云服务商API调用 │ ├── adb_operator.py # 封装ADB操作 │ ├── scheduler.py # 核心调度逻辑 │ └── main.py # 程序主入口 ├── logs/ # 日志目录 ├── requirements.txt # Python依赖列表 └── README.md # 项目说明3. 核心模块拆解与代码实现接下来我们分模块实现自动化系统的核心功能。3.1 配置文件管理 (config/config.yaml)使用YAML文件管理配置避免将敏感信息硬编码在脚本中。# config/config.yaml aliyun: access_key_id: 你的AccessKeyId # 重要此处仅为示例实际应从环境变量或密钥管理服务读取 access_key_secret: 你的AccessKeySecret region_id: cn-hangzhou endpoint: cph.cn-hangzhou.aliyuncs.com cloud_phone: instance_type: cloud_phone.general.large # 实例规格根据云厂商定义 image_id: android_9_0_xxxxxx # 你准备好的镜像ID vpc_id: vpc-xxxxx # 云手机需要加入的VPC vswitch_id: vsw-xxxxx # 交换机ID security_group_id: sg-xxxxx # 安全组ID scheduler: check_interval_minutes: 5 # 调度器检查间隔分钟 max_idle_time_minutes: 30 # 云手机最大空闲时间分钟超过则销毁 max_running_instances: 5 # 最大同时运行的云手机实例数3.2 云服务商API封装 (src/cloud_api.py)这个模块负责与阿里云CPH API交互。我们实现几个关键功能创建实例、查询实例列表、启动/停止实例、释放实例。# src/cloud_api.py import json import time import hmac import hashlib import base64 import urllib.parse from typing import Dict, List, Optional import requests import yaml class AliCloudPhoneClient: 阿里云云手机API客户端封装 def __init__(self, config_pathconfig/config.yaml): with open(config_path, r) as f: config yaml.safe_load(f) aliyun_cfg config[aliyun] self.access_key_id aliyun_cfg[access_key_id] self.access_key_secret aliyun_cfg[access_key_secret] self.region_id aliyun_cfg[region_id] self.endpoint aliyun_cfg.get(endpoint, fcph.{self.region_id}.aliyuncs.com) self.cloud_phone_cfg config[cloud_phone] def _sign_request(self, params: Dict) - Dict: 生成阿里云API签名简化版生产环境建议使用阿里云SDK params.update({ Format: JSON, Version: 2022-08-30, # CPH API版本请以官方文档为准 AccessKeyId: self.access_key_id, SignatureMethod: HMAC-SHA1, Timestamp: time.strftime(%Y-%m-%dT%H:%M:%SZ, time.gmtime()), SignatureVersion: 1.0, SignatureNonce: str(int(time.time() * 1000)), }) # 对参数排序并编码 sorted_params sorted(params.items()) canonicalized_query_string urllib.parse.urlencode(sorted_params) # 构造签名字符串 string_to_sign GET%2F urllib.parse.quote(canonicalized_query_string, safe) # 计算签名 key self.access_key_secret signature base64.b64encode(hmac.new(key.encode(utf-8), string_to_sign.encode(utf-8), hashlib.sha1).digest()) params[Signature] signature.decode(utf-8) return params def create_instance(self, instance_name: str None) - Optional[str]: 创建一台云手机实例返回实例ID action CreateCloudPhone params { Action: action, RegionId: self.region_id, InstanceType: self.cloud_phone_cfg[instance_type], ImageId: self.cloud_phone_cfg[image_id], VpcId: self.cloud_phone_cfg[vpc_id], VSwitchId: self.cloud_phone_cfg[vswitch_id], SecurityGroupId: self.cloud_phone_cfg[security_group_id], Amount: 1, } if instance_name: params[InstanceName] instance_name signed_params self._sign_request(params) url fhttps://{self.endpoint}/? try: # 注意此处为示例实际阿里云CPH API的调用方式、参数和返回结构请严格参考官方最新文档 # 生产环境强烈建议使用官方SDKaliyun-python-sdk-cph response requests.get(url, paramssigned_params, timeout30) result response.json() if InstanceIds in result and result[InstanceIds]: print(f[INFO] 实例创建成功: {result[InstanceIds]}) return result[InstanceIds][0] # 返回第一个实例ID else: print(f[ERROR] 创建实例失败: {result}) return None except Exception as e: print(f[ERROR] 调用创建实例API异常: {e}) return None def list_instances(self, status: str None) - List[Dict]: 查询云手机实例列表可过滤状态 action ListCloudPhoneInstances params { Action: action, RegionId: self.region_id, } signed_params self._sign_request(params) url fhttps://{self.endpoint}/? try: response requests.get(url, paramssigned_params, timeout30) result response.json() instances result.get(Instances, {}).get(Instance, []) if status: instances [inst for inst in instances if inst.get(Status) status] return instances except Exception as e: print(f[ERROR] 查询实例列表异常: {e}) return [] def stop_instance(self, instance_id: str) - bool: 停止一台云手机实例 # 类似 create_instance调用 StopCloudPhoneInstance API # 此处省略具体实现请参考阿里云文档 print(f[INFO] 停止实例: {instance_id}) return True # 模拟成功 def release_instance(self, instance_id: str) - bool: 释放删除一台云手机实例 # 类似 create_instance调用 ReleaseCloudPhoneInstance API # 此处省略具体实现请参考阿里云文档 print(f[INFO] 释放实例: {instance_id}) return True # 模拟成功 def get_instance_connection_info(self, instance_id: str) - Optional[Dict]: 获取实例的连接信息如ADB连接地址 # 调用 DescribeCloudPhoneInstance API 或相关接口 # 返回格式如: {adb_host: xxx.xxx.xxx.xxx, adb_port: 5555} # 此处返回模拟数据 return {adb_host: 127.0.0.1, adb_port: 7555} # 示例端口重要说明以上API调用代码为原理演示。阿里云等厂商的API签名机制复杂且可能更新生产环境务必使用官方提供的SDK如pip install aliyun-python-sdk-cph上面的_sign_request函数仅为说明流程可能不适用于最新API。3.3 ADB操作封装 (src/adb_operator.py)这个模块负责通过ADB与云手机交互。# src/adb_operator.py import subprocess import time from typing import Optional, List class ADBOperator: ADB命令操作封装 staticmethod def connect(host: str, port: int) - bool: 连接到指定的ADB服务 cmd [adb, connect, f{host}:{port}] try: result subprocess.run(cmd, capture_outputTrue, textTrue, timeout10) if connected in result.stdout or already connected in result.stdout: print(f[INFO] ADB连接成功: {host}:{port}) return True else: print(f[ERROR] ADB连接失败: {result.stderr}) return False except subprocess.TimeoutExpired: print(f[ERROR] 连接ADB超时: {host}:{port}) return False staticmethod def disconnect(host: str, port: int): 断开ADB连接 cmd [adb, disconnect, f{host}:{port}] subprocess.run(cmd, capture_outputTrue) staticmethod def is_device_online(host: str, port: int) - bool: 检查设备是否在线 cmd [adb, devices] result subprocess.run(cmd, capture_outputTrue, textTrue) target_device f{host}:{port} return target_device in result.stdout and device in result.stdout staticmethod def execute_shell(host: str, port: int, shell_cmd: str) - Optional[str]: 在设备上执行shell命令并返回输出 if not ADBOperator.is_device_online(host, port): if not ADBOperator.connect(host, port): return None full_cmd [adb, -s, f{host}:{port}, shell, shell_cmd] try: result subprocess.run(full_cmd, capture_outputTrue, textTrue, timeout30) return result.stdout.strip() except subprocess.TimeoutExpired: print(f[ERROR] 执行命令超时: {shell_cmd}) return None staticmethod def install_apk(host: str, port: int, apk_path: str) - bool: 安装APK到设备 if not ADBOperator.is_device_online(host, port): ADBOperator.connect(host, port) cmd [adb, -s, f{host}:{port}, install, -r, apk_path] # -r 覆盖安装 result subprocess.run(cmd, capture_outputTrue, textTrue, timeout60) if Success in result.stdout: print(f[INFO] APK安装成功: {apk_path}) return True else: print(f[ERROR] APK安装失败: {result.stderr}) return False staticmethod def get_top_activity(host: str, port: int) - Optional[str]: 获取设备当前最顶层的Activity包名/类名 # 方法1通过dumpsys activity top cmd dumpsys activity top | grep ACTIVITY output ADBOperator.execute_shell(host, port, cmd) if output: # 简化解析实际需要根据输出格式调整 for line in output.split(\n): if ACTIVITY in line: parts line.split() if len(parts) 1: return parts[1] # 返回类似 com.android.settings/.Settings return None staticmethod def simulate_activity(host: str, port: int) - bool: 模拟一次用户操作例如点击、滑动防止设备因无操作进入深度休眠。 这是一个示例在屏幕上执行一次无效的滑动。 # 在屏幕坐标(100,200)到(100,300)之间滑动这是一个很小的垂直滑动通常不会触发实际UI变化 cmd input swipe 100 200 100 300 100 output ADBOperator.execute_shell(host, port, cmd) return output is not None3.4 核心调度逻辑 (src/scheduler.py)调度器是大脑它定期检查云手机状态并根据策略决定创建或销毁实例。# src/scheduler.py import time import threading from datetime import datetime, timedelta from typing import Dict from .cloud_api import AliCloudPhoneClient from .adb_operator import ADBOperator import yaml class CloudPhoneScheduler: 云手机调度器 def __init__(self, config_pathconfig/config.yaml): with open(config_path, r) as f: self.config yaml.safe_load(f) self.scheduler_cfg self.config[scheduler] self.cloud_client AliCloudPhoneClient(config_path) self.adb_operator ADBOperator() # 用于记录实例的最后活动时间 {instance_id: last_active_time} self.instance_activity: Dict[str, datetime] {} self.running False def _update_activity(self, instance_id: str): 更新实例的活动时间戳 self.instance_activity[instance_id] datetime.now() def _is_instance_idle(self, instance_id: str) - bool: 判断实例是否空闲超时 if instance_id not in self.instance_activity: return True # 没有记录视为需要检查或可能已空闲 last_active self.instance_activity[instance_id] idle_threshold timedelta(minutesself.scheduler_cfg[max_idle_time_minutes]) return datetime.now() - last_active idle_threshold def check_and_manage_instances(self): 核心调度循环检查状态管理实例生命周期 print(f[SCHEDULER] 开始调度检查 {datetime.now()}) # 1. 获取所有运行中的云手机实例 running_instances self.cloud_client.list_instances(statusRunning) # 假设状态为Running print(f[INFO] 当前运行中实例数: {len(running_instances)}) # 2. 检查每个实例的活动状态 for inst in running_instances: instance_id inst[InstanceId] conn_info self.cloud_client.get_instance_connection_info(instance_id) if not conn_info: continue # 尝试通过ADB连接并检查 if self.adb_operator.connect(conn_info[adb_host], conn_info[adb_port]): # 检查当前顶层Activity或执行一个简单命令来判断是否“活跃” top_activity self.adb_operator.get_top_activity(conn_info[adb_host], conn_info[adb_port]) # 这里定义“活跃”例如顶层Activity不是锁屏或待机界面 # 简化逻辑如果能成功执行命令就认为有活动或者根据业务定义 if top_activity and com.android.systemui not in top_activity: # 示例非系统UI self._update_activity(instance_id) print(f[INFO] 实例 {instance_id} 处于活跃状态.) else: # 设备可能空闲模拟一次操作保活并检查是否真的需要销毁 self.adb_operator.simulate_activity(conn_info[adb_host], conn_info[adb_port]) if self._is_instance_idle(instance_id): print(f[WARN] 实例 {instance_id} 空闲超时准备释放...) # 先尝试正常停止 if self.cloud_client.stop_instance(instance_id): time.sleep(30) # 等待停止完成 # 释放实例 if self.cloud_client.release_instance(instance_id): print(f[INFO] 实例 {instance_id} 已释放.) if instance_id in self.instance_activity: del self.instance_activity[instance_id] else: print(f[WARN] 实例 {instance_id} ADB连接失败可能实例异常。) # 可以考虑强制释放或重启 # 3. 检查是否需要创建新实例例如当前数量小于最小保有量或有待处理任务 current_count len(running_instances) max_allowed self.scheduler_cfg[max_running_instances] # 假设我们有一个任务队列这里用简单逻辑如果数量少于最大限制的一半就创建一个 if current_count max_allowed // 2: print(f[INFO] 当前实例数({current_count})较少尝试创建新实例...) new_instance_id self.cloud_client.create_instance() if new_instance_id: print(f[INFO] 新实例创建成功: {new_instance_id}) # 创建后可以在这里执行初始化操作如安装特定APK # 需要等待实例完全启动并获取连接信息 time.sleep(60) # 等待实例启动 conn_info self.cloud_client.get_instance_connection_info(new_instance_id) if conn_info: self.adb_operator.connect(conn_info[adb_host], conn_info[adb_port]) # 示例安装一个测试APK # self.adb_operator.install_apk(conn_info[adb_host], conn_info[adb_port], test.apk) self._update_activity(new_instance_id) print(f[SCHEDULER] 调度检查结束 {datetime.now()}) def start(self, interval_minutes: int None): 启动调度器定时执行检查 if interval_minutes is None: interval_minutes self.scheduler_cfg[check_interval_minutes] self.running True def run_scheduler(): while self.running: try: self.check_and_manage_instances() except Exception as e: print(f[ERROR] 调度循环发生异常: {e}) time.sleep(interval_minutes * 60) scheduler_thread threading.Thread(targetrun_scheduler, daemonTrue) scheduler_thread.start() print(f[INFO] 云手机调度器已启动检查间隔 {interval_minutes} 分钟.) def stop(self): 停止调度器 self.running False print([INFO] 调度器停止.)3.5 主程序入口 (src/main.py)将各部分组合起来提供一个简单的启动方式。# src/main.py import signal import sys import time from scheduler import CloudPhoneScheduler def signal_handler(sig, frame): print(\n接收到中断信号正在停止调度器...) scheduler.stop() sys.exit(0) if __name__ __main__: # 注册信号处理方便CtrlC退出 signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) print( 云手机自动化管理调度系统启动 ) # 初始化调度器 scheduler CloudPhoneScheduler(config_path../config/config.yaml) # 注意路径 # 启动调度循环 scheduler.start() # 主线程保持运行 try: while True: time.sleep(1) except KeyboardInterrupt: scheduler.stop()4. 运行与验证4.1 配置与启动根据你的阿里云账号信息修改config/config.yaml文件。在项目根目录安装依赖pip install -r requirements.txt。确保你的本地ADB可以正常工作 (adb devices)。在cloud_phone_manager/src/目录下运行主程序cd cloud_phone_manager/src python main.py程序启动后会在控制台打印日志并开始周期性地检查和管理云手机实例。4.2 观察与验证查看日志观察调度器是否按预期运行创建、检查、释放实例。登录阿里云控制台在弹性云手机CPH控制台查看实例列表的变化是否与脚本逻辑一致。手动ADB连接当脚本创建新实例后你可以尝试用脚本中打印的连接信息手动连接验证ADB通道是否畅通。adb connect adb_host:adb_port adb -s adb_host:adb_port shell pm list packages # 列出安装的包5. 常见问题与排查思路在实现和运行上述系统时你可能会遇到以下问题问题现象可能原因排查思路与解决方案API调用失败返回 Signature 错误1. AccessKey ID/Secret 错误或失效。2. API签名算法实现有误。3. 请求参数格式或编码错误。1. 检查AK/SK是否正确是否有访问CPH的权限。2.强烈建议改用阿里云官方Python SDK避免自己实现签名。3. 使用网络抓包工具对比官方SDK的请求与自己构造的请求差异。adb connect成功但adb shell无响应1. 云手机实例尚未完全启动或系统未就绪。2. 云手机镜像未开启ADB调试或端口未开放。3. 网络策略安全组未放行ADB端口通常为5555或7xxx。1. 创建实例后等待足够长时间如2-3分钟。2. 确保使用的云手机镜像已预先开启“开发者选项”中的“USB调试”。3. 检查云手机实例所属安全组的入方向规则是否允许本地IP访问ADB端口。调度器无法正确判断实例“空闲”1. “活跃”判断逻辑过于简单或不符合业务场景。2. 实例活动时间戳更新策略有误。1. 根据实际业务定义“空闲”。例如可以监控特定App的前后台状态、网络流量、CPU使用率等需在镜像内安装监控Agent。2. 优化_update_activity和_is_instance_idle的逻辑考虑更细粒度的活动信号。达到最大实例数后新任务无法执行调度策略仅基于实例数量未与任务队列关联。引入一个任务队列如Redis、数据库。调度器检查队列中是否有待处理任务再决定是否创建新实例。实例完成任务后可标记为空闲或直接销毁。脚本长时间运行后内存泄漏或卡死1. 网络请求或子进程未妥善处理异常和资源释放。2. 线程管理不当。1. 为所有网络请求和subprocess调用添加超时和异常捕获。2. 考虑将调度器改为使用APScheduler等成熟的调度库而非简单time.sleep循环。3. 定期重启调度器进程或使用进程守护工具如systemd。6. 最佳实践与工程化建议将脚本升级为可维护、可靠的生产系统需要考虑以下几点配置与密钥管理绝对不要将AK/SK等敏感信息硬编码或提交到代码仓库。使用环境变量、密钥管理服务如阿里云KMS或专门的配置中心来注入配置。配置文件与代码分离。日志记录将print语句替换为标准的logging模块并配置不同级别INFO, WARNING, ERROR的输出。日志应输出到文件并设置日志轮转便于后期排查问题。import logging logging.basicConfig(levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[logging.FileHandler(scheduler.log), logging.StreamHandler()]) logger logging.getLogger(__name__)错误处理与重试云API调用和网络操作必须添加重试机制如使用tenacity库。区分可重试错误如网络超时和不可重试错误如权限不足。状态持久化当前示例将实例活动时间保存在内存字典中进程重启会丢失。生产环境应将状态如实例ID、创建时间、最后活动时间、任务绑定关系持久化到数据库如SQLite、MySQL或Redis中。监控与告警监控调度器进程本身的存活状态可用systemd或supervisor。监控云手机实例的资源使用率CPU、内存、费用消耗。设置告警当实例创建失败、释放异常或费用超预算时及时通知负责人。代码优化与扩展将CloudPhoneScheduler抽象为基类便于支持不同云厂商阿里云、腾讯云。使用异步编程asyncio来提高I/O密集型操作如批量ADB命令的并发效率。考虑容器化部署将整个调度系统打包成Docker镜像便于环境一致性和水平扩展。通过以上步骤你不仅实现了一个基础的云手机自动化管理系统更掌握了一套构建资源自动化运维框架的方法论。这套系统的核心思想——状态监控、策略决策、API驱动、异常处理——可以迁移到管理其他类型的云资源如ECS服务器、容器实例、数据库等。