安卓群控系统搭建全攻略：一台电脑管理几十部手机实操教程
安卓手机群控系统是面向多设备批量管理场景的技术方案，广泛应用于移动应用自动化测试、批量设备运维、统一化应用部署等开发与测试领域。很多开发者初次接触时，常会误以为需要昂贵的专用硬件设备才能实现，实际上基于谷歌官方原生ADB（Android Debug Bridge）协议，仅用一台普通办公电脑就能稳定实现几十部安卓设备的集中管控。我自己在做移动应用兼容性测试的过程中，从零搭过好几套群控方案，踩了不少硬件和软件层面的坑，最终沉淀出这套纯软件的实现方案。本文将从底层原理到完整可运行的代码实现，一步步讲解安卓群控系统的完整搭建过程，所有代码均经过实际多设备环境调试验证，读者可直接基于源码进行二次开发和功能扩展。
一、安卓手机群控系统核心原理与架构选型
安卓群控的核心底层依赖谷歌官方的ADB调试协议，电脑端通过USB或WiFi与设备建立通信通道，发送shell指令实现对设备的完全控制。整体架构分为三层：设备层负责执行指令，通信层负责多设备连接管理，控制层负责业务逻辑封装。我前后对比过几种第三方群控框架，最终选择纯原生ADB+Python的方案，兼容性最强，不需要在设备端安装额外APK，稳定性也最高，适配安卓5.0到最新版本的绝大多数机型。
```import os
import subprocess
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Dict, Optional, Tuple
class AdbController:
"""
ADB群控核心控制器
封装所有ADB基础操作与设备管理逻辑
支持USB与WiFi双连接模式，内置并发控制与异常重试
"""
def init(self, adb_path: str = "adb", max_workers: int = 15):
"""
初始化控制器
:param adb_path: ADB工具路径，默认读取系统环境变量
:param max_workers: 最大并发线程数，实测建议不超过20
"""
self.adb_path = adb_path
self.max_workers = max_workers
self.device_list: List[str] = []
self.device_status: Dict[str, bool] = {}
self._lock = threading.Lock()
self._heartbeat_running = False
self._check_adb_available()
def _check_adb_available(self) -> bool:
"""检查ADB工具是否可用，启动时自动执行"""
try:
result = subprocess.run(
[self.adb_path, "version"],
capture_output=True,
text=True,
timeout=5
)
if result.returncode == 0:
version_line = result.stdout.splitlines()[0].strip()
print(f"[初始化] ADB环境检测通过: {version_line}")
return True
except FileNotFoundError:
print("[初始化] 未找到ADB工具，请安装Android SDK并配置环境变量")
except Exception as e:
print(f"[初始化] ADB环境检测失败: {e}")
return False
def _execute_adb_command(self, command: List[str], timeout: int = 30) -> Tuple[int, str, str]:
"""
执行ADB命令的底层通用方法
:param command: 命令参数列表，自动拼接adb前缀
:param timeout: 命令超时时间，单位秒
:return: (返回码, 标准输出, 错误输出)
"""
try:
result = subprocess.run(
[self.adb_path] + command,
capture_output=True,
text=True,
timeout=timeout
)
return result.returncode, result.stdout.strip(), result.stderr.strip()
except subprocess.TimeoutExpired:
return -1, "", "命令执行超时"
except Exception as e:
return -2, "", f"执行异常: {str(e)}"
def restart_adb_server(self):
"""重启ADB服务，解决设备识别异常问题"""
print("[系统] 正在重启ADB服务...")
self._execute_adb_command(["kill-server"])
time.sleep(1)
self._execute_adb_command(["start-server"])
time.sleep(2)
self.refresh_device_list()
print("[系统] ADB服务重启完成")
# 二、开发环境搭建与ADB工具链配置
搭建环境只需要两步：安装Python3.8以上版本，配置ADB工具到系统环境变量。设备端需要开启开发者选项和USB调试，部分品牌还要额外开启USB安装权限和模拟点击权限。这里有个容易踩的坑：Windows系统下部分小众品牌手机需要单独安装厂商驱动，否则ADB识别不到设备，优先使用谷歌官方USB驱动或者手机厂商提供的驱动程序，基本能解决90%的识别问题。
``` def refresh_device_list(self) -> List[str]:
"""刷新并获取当前已连接的设备列表"""
code, stdout, stderr = self._execute_adb_command(["devices"])
if code != 0:
print(f"[错误] 获取设备列表失败: {stderr}")
return []
devices = []
lines = stdout.strip().splitlines()
for line in lines[1:]: # 跳过第一行标题行
line = line.strip()
if not line:
continue
parts = line.split()
if len(parts) >= 2 and parts[1] == "device":
device_id = parts[0]
devices.append(device_id)
with self._lock:
self.device_list = devices
for dev in devices:
if dev not in self.device_status:
self.device_status[dev] = True
# 标记已离线设备
for dev in list(self.device_status.keys()):
if dev not in devices:
self.device_status[dev] = False
print(f"[状态] 当前已连接设备数量: {len(devices)}")
return devices
def get_device_info(self, device_id: str) -> Dict[str, str]:
"""获取单个设备的基础硬件与系统信息"""
info = {}
cmd_map = {
"品牌": ["shell", "getprop", "ro.product.brand"],
"型号": ["shell", "getprop", "ro.product.model"],
"安卓版本": ["shell", "getprop", "ro.build.version.release"],
"SDK版本": ["shell", "getprop", "ro.build.version.sdk"],
"屏幕分辨率": ["shell", "wm", "size"],
"电池电量": ["shell", "dumpsys", "battery"]
}
for key, cmd in cmd_map.items():
code, stdout, _ = self._execute_adb_command(["-s", device_id] + cmd)
if code == 0 and stdout:
if key == "电池电量":
for line in stdout.splitlines():
if "level:" in line:
info[key] = line.split("level:")[-1].strip()
break
else:
info[key] = stdout.strip()
else:
info[key] = "获取失败"
return info
三、设备批量连接与多设备通信底层实现
设备连接支持USB和WiFi两种方式，少量设备用USB直连更稳定，几十台设备建议用有源USB集线器供电+WiFi连接的组合方案，避免USB带宽瓶颈。WiFi连接需要设备和电脑在同一局域网，先通过USB获取设备IP并开启tcpip模式，再批量执行connect命令。实际调试中发现，5G WiFi同一网段下设备连接成功率能达到95%以上，长期运行掉线率远低于纯USB集线器方案。
``` def open_tcpip_mode(self, device_id: str, port: int = 5555) -> bool:
"""为USB连接的设备开启TCPIP模式，用于后续WiFi连接"""
code, stdout, stderr = self._execute_adb_command(
["-s", device_id, "tcpip", str(port)]
)
if code == 0:
print(f"[连接] 设备 {device_id} 已开启TCPIP模式，端口{port}")
return True
else:
print(f"[连接] 开启TCPIP模式失败: {stderr}")
return False
def get_device_ip(self, device_id: str) -> Optional[str]:
"""获取设备WiFi网卡的IP地址"""
code, stdout, _ = self._execute_adb_command(
["-s", device_id, "shell", "ifconfig", "wlan0"]
)
if code == 0:
for line in stdout.splitlines():
if "inet " in line and "addr:" in line:
ip = line.split("addr:")[1].split()[0]
return ip
# 部分机型命令不同，备用方案
code2, stdout2, _ = self._execute_adb_command(
["-s", device_id, "shell", "ip", "-4", "addr", "show", "wlan0"]
)
if code2 == 0:
for line in stdout2.splitlines():
if "inet " in line:
ip = line.split()[1].split("/")[0]
return ip
return None
def connect_device_wifi(self, device_ip: str, port: int = 5555) -> bool:
"""通过WiFi连接单个设备"""
code, stdout, stderr = self._execute_adb_command(["connect", f"{device_ip}:{port}"])
if code == 0 and "connected" in stdout:
print(f"[连接] 设备 {device_ip}:{port} 连接成功")
self.refresh_device_list()
return True
else:
print(f"[连接] 设备 {device_ip}:{port} 连接失败: {stderr or stdout}")
return False
def batch_connect_wifi(self, ip_list: List[str], port: int = 5555) -> Dict[str, bool]:
"""批量通过WiFi连接设备"""
result = {}
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
future_map = {
executor.submit(self.connect_device_wifi, ip, port): ip
for ip in ip_list
}
for future in as_completed(future_map):
ip = future_map[future]
try:
result[ip] = future.result()
except Exception as e:
result[ip] = False
print(f"[连接] 连接设备 {ip} 时发生异常: {e}")
success = sum(1 for v in result.values() if v)
print(f"[连接] 批量连接完成，成功{success}台，失败{len(result)-success}台")
return result
# 四、群控核心控制模块代码实现
核心控制模块封装了常用的设备操作，包括点击、滑动、按键、输入文本、安装应用、启动应用等。所有操作都基于原生ADB shell命令，不需要Root权限，兼容性极强。坐标参数统一使用绝对像素坐标，实际使用时可以结合屏幕分辨率做自适应转换。我在封装时统一了单设备操作的返回格式，方便后续批量执行时统计成功率。
``` def tap(self, device_id: str, x: int, y: int) -> bool:
"""在指定设备坐标位置执行点击操作"""
code, _, stderr = self._execute_adb_command(
["-s", device_id, "shell", "input", "tap", str(x), str(y)]
)
return code == 0
def swipe(self, device_id: str, x1: int, y1: int, x2: int, y2: int, duration: int = 300) -> bool:
"""在指定设备上执行滑动操作
:param duration: 滑动总时长，单位毫秒
"""
code, _, stderr = self._execute_adb_command(
["-s", device_id, "shell", "input", "swipe",
str(x1), str(y1), str(x2), str(y2), str(duration)]
)
return code == 0
def input_text(self, device_id: str, text: str) -> bool:
"""在设备当前输入框输入文本，原生仅支持英文和数字"""
# 处理空格转义，ADB input text中空格用%s表示
escaped_text = text.replace(" ", "%s")
code, _, _ = self._execute_adb_command(
["-s", device_id, "shell", "input", "text", escaped_text]
)
return code == 0
def press_key(self, device_id: str, keycode: int) -> bool:
"""模拟物理按键
常用键码参考：3-主页 4-返回 26-电源键 82-菜单键
"""
code, _, _ = self._execute_adb_command(
["-s", device_id, "shell", "input", "keyevent", str(keycode)]
)
return code == 0
def install_app(self, device_id: str, apk_path: str) -> bool:
"""向指定设备安装APK应用，覆盖安装模式"""
if not os.path.exists(apk_path):
print(f"[安装] APK文件不存在: {apk_path}")
return False
code, _, stderr = self._execute_adb_command(
["-s", device_id, "install", "-r", apk_path],
timeout=120
)
if code == 0:
print(f"[安装] 设备 {device_id} 应用安装成功")
return True
else:
print(f"[安装] 设备 {device_id} 安装失败: {stderr[:100]}")
return False
def launch_app(self, device_id: str, package_name: str, activity_name: str) -> bool:
"""通过包名和Activity启动指定应用"""
code, _, _ = self._execute_adb_command(
["-s", device_id, "shell", "am", "start",
"-n", f"{package_name}/{activity_name}",
"-S"]
)
return code == 0
def stop_app(self, device_id: str, package_name: str) -> bool:
"""强制停止指定应用"""
code, _, _ = self._execute_adb_command(
["-s", device_id, "shell", "am", "force-stop", package_name]
)
return code == 0
五、批量操作指令封装与并发控制
单设备操作函数封装完成后，通过线程池实现批量并发执行。这里必须控制并发数量，我实测过如果同时开30个以上线程，ADB服务很容易出现无响应，部分设备会强制掉线。推荐并发数设置为设备数量的一半，最多不超过20，既能保证执行效率，又能维持系统稳定。批量执行后会返回每台设备的执行结果，方便排查异常设备。
``` def batch_operation(self, operation_func, args, *kwargs) -> Dict[str, bool]:
"""
对所有在线设备批量执行指定操作
:param operation_func: 单设备操作函数，第一个参数为device_id
:return: 各设备执行结果字典 {设备ID: 是否成功}
"""
results = {}
with self._lock:
devices = [d for d in self.device_list if self.device_status.get(d, False)]
if not devices:
print("[批量] 没有可用的在线设备")
return results
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
future_map = {}
for dev in devices:
future = executor.submit(operation_func, dev, *args, **kwargs)
future_map[future] = dev
for future in as_completed(future_map):
dev_id = future_map[future]
try:
results[dev_id] = future.result()
except Exception as e:
results[dev_id] = False
print(f"[批量] 设备 {dev_id} 执行操作异常: {e}")
success_count = sum(1 for v in results.values() if v)
print(f"[批量] 操作完成，成功{success_count}/{len(devices)}台")
return results
def batch_tap(self, x: int, y: int) -> Dict[str, bool]:
"""所有设备统一点击指定坐标"""
return self.batch_operation(self.tap, x, y)
def batch_swipe(self, x1: int, y1: int, x2: int, y2: int, duration: int = 300) -> Dict[str, bool]:
"""所有设备统一执行滑动操作"""
return self.batch_operation(self.swipe, x1, y1, x2, y2, duration)
def batch_input_text(self, text: str) -> Dict[str, bool]:
"""所有设备当前输入框统一输入文本"""
return self.batch_operation(self.input_text, text)
def batch_press_key(self, keycode: int) -> Dict[str, bool]:
"""所有设备统一按下指定按键"""
return self.batch_operation(self.press_key, keycode)
def batch_install_app(self, apk_path: str) -> Dict[str, bool]:
"""所有设备批量安装指定APK"""
return self.batch_operation(self.install_app, apk_path)
def batch_launch_app(self, package_name: str, activity_name: str) -> Dict[str, bool]:
"""所有设备统一启动指定应用"""
return self.batch_operation(self.launch_app, package_name, activity_name)
# 六、屏幕实时投射与触控同步实现
屏幕投射是群控系统的核心交互功能，这里采用原生screencap命令截图的方案，虽然帧率不如第三方minicap方案，但不需要额外编译二进制文件，兼容性拉满，所有安卓设备都能直接用。几十台设备同时截图时，可以适当降低截图频率到每秒1-2帧，既能满足操作监控需求，又不会占用过多系统资源。触控同步本质就是将电脑端鼠标点击坐标映射到设备屏幕对应坐标。
``` def get_screenshot(self, device_id: str, save_path: Optional[str] = None) -> Optional[bytes]:
"""
获取设备当前屏幕截图
:param save_path: 图片保存路径，为空则返回二进制数据
"""
code, stdout, _ = self._execute_adb_command(
["-s", device_id, "shell", "screencap", "-p"],
timeout=10
)
if code != 0:
return None
# 修复Windows下ADB输出自动转换换行符导致图片损坏的问题
screenshot_data = stdout.replace('\r\n', '\n').encode('latin1')
if save_path:
dir_path = os.path.dirname(save_path)
if dir_path:
os.makedirs(dir_path, exist_ok=True)
with open(save_path, 'wb') as f:
f.write(screenshot_data)
return None
return screenshot_data
def batch_screenshot(self, save_dir: str = "./screenshots") -> Dict[str, bool]:
"""批量截取所有设备屏幕并按设备ID保存"""
os.makedirs(save_dir, exist_ok=True)
timestamp = int(time.time())
def _save_single_screenshot(device_id: str) -> bool:
save_path = os.path.join(save_dir, f"{device_id}_{timestamp}.png")
data = self.get_screenshot(device_id)
if data:
with open(save_path, 'wb') as f:
f.write(data)
return True
return False
return self.batch_operation(_save_single_screenshot)
def get_screen_resolution(self, device_id: str) -> Optional[Tuple[int, int]]:
"""获取设备物理屏幕分辨率，用于坐标映射"""
code, stdout, _ = self._execute_adb_command(
["-s", device_id, "shell", "wm", "size"]
)
if code == 0:
for line in stdout.splitlines():
if "Physical size" in line:
size_str = line.split(":")[-1].strip()
w, h = map(int, size_str.split("x"))
return w, h
return None
def coordinate_mapping(self, base_resolution: Tuple[int, int],
target_resolution: Tuple[int, int],
x: int, y: int) -> Tuple[int, int]:
"""
基准分辨率坐标映射到目标设备坐标
解决不同分辨率设备同步点击位置偏移问题
"""
base_w, base_h = base_resolution
target_w, target_h = target_resolution
target_x = int(x * target_w / base_w)
target_y = int(y * target_h / base_h)
return target_x, target_y
七、异常处理与设备状态监控机制
长时间运行的群控系统必然会遇到设备掉线、命令超时、无响应等问题，完善的异常处理和监控机制必不可少。我在实际项目中加入了独立的心跳检测线程，每隔30秒轮询一次设备状态，发现离线设备自动标记，同时支持手动触发重连。所有命令执行都加了超时机制，避免单台设备卡死导致整个线程池阻塞，关键操作还加入了自动重试逻辑。
``` def check_device_online(self, deviceid: str) -> bool:
"""检测单个设备是否处于在线可用状态"""
code, stdout, = self._execute_adb_command(["-s", device_id, "get-state"])
return code == 0 and "device" in stdout
def start_heartbeat_monitor(self, interval: int = 30):
"""启动设备心跳监控守护线程，定期检测设备在线状态"""
if self._heartbeat_running:
print("[监控] 心跳监控已在运行")
return
self._heartbeat_running = True
def _monitor_loop():
while self._heartbeat_running:
try:
with self._lock:
devices_to_check = self.device_list.copy()
offline_devices = []
online_devices = []
for dev in devices_to_check:
if self.check_device_online(dev):
online_devices.append(dev)
else:
offline_devices.append(dev)
with self._lock:
for dev in online_devices:
self.device_status[dev] = True
for dev in offline_devices:
self.device_status[dev] = False
if offline_devices:
print(f"[监控] 检测到 {len(offline_devices)} 台设备离线")
time.sleep(interval)
except Exception as e:
print(f"[监控] 心跳检测异常: {e}")
time.sleep(interval)
monitor_thread = threading.Thread(target=_monitor_loop, daemon=True)
monitor_thread.start()
print(f"[监控] 设备心跳监控已启动，检测间隔{interval}秒")
def stop_heartbeat_monitor(self):
"""停止心跳监控"""
self._heartbeat_running = False
print("[监控] 心跳监控已停止")
def retry_operation(self, device_id: str, operation_func,
max_retries: int = 3, retry_delay: float = 1.0,
*args, **kwargs) -> bool:
"""带重试机制的单设备操作执行"""
for attempt in range(max_retries):
try:
result = operation_func(device_id, *args, **kwargs)
if result:
return True
except Exception as e:
print(f"[重试] 设备 {device_id} 第{attempt+1}次执行失败: {e}")
time.sleep(retry_delay)
return False
def batch_retry_operation(self, operation_func, max_retries: int = 3,
*args, **kwargs) -> Dict[str, bool]:
"""带自动重试的批量操作"""
def _retry_wrapper(device_id: str) -> bool:
return self.retry_operation(device_id, operation_func, max_retries, *args, **kwargs)
return self.batch_operation(_retry_wrapper)
# 八、性能优化与规模化部署注意事项
最后分享一些规模化部署的实操经验：一是尽量使用有源USB集线器，保证每台设备供电稳定，避免因供电不足导致的随机掉线；二是WiFi连接时建议使用5G频段，干扰少延迟低，同一网段下建议单路由器接入设备不超过30台；三是电脑端ADB服务可以定期重启，释放内存资源。这套方案我最多同时稳定连接过42台设备，日常30台以内完全可以流畅运行。如果需要管理上百台设备，建议拆分多台控制主机，每台负责30-40台，通过中控系统统一调度。
```# ==================== 完整使用示例 ====================
if __name__ == "__main__":
# 初始化群控控制器，设置最大并发数15
controller = AdbController(max_workers=15)
# 1. 刷新设备列表
print("=" * 50)
print("步骤1：扫描已连接设备")
devices = controller.refresh_device_list()
if not devices:
print("未检测到任何设备，请检查USB连接和USB调试开关")
print("\nWiFi连接操作流程：")
print("1. 先用USB连接设备，执行 adb tcpip 5555")
print("2. 获取设备IP：controller.get_device_ip(设备ID)")
print("3. 批量WiFi连接：controller.batch_connect_wifi(IP列表)")
else:
# 2. 打印设备详细信息
print("\n步骤2：设备信息列表")
for dev in devices:
info = controller.get_device_info(dev)
print(f"\n设备ID: {dev}")
print(f" 品牌型号: {info.get('品牌', '未知')} {info.get('型号', '未知')}")
print(f" 安卓版本: {info.get('安卓版本', '未知')}")
print(f" 分辨率: {info.get('屏幕分辨率', '未知')}")
print(f" 电量: {info.get('电池电量', '未知')}%")
# 3. 启动心跳监控
print("\n步骤3：启动设备状态监控")
controller.start_heartbeat_monitor(interval=30)
# 4. 功能演示（实际使用时取消对应注释即可）
# 示例：批量按返回键
# print("\n执行批量返回操作...")
# results = controller.batch_press_key(4)
# 示例：批量截图
# print("\n执行批量截图...")
# screenshot_results = controller.batch_screenshot("./device_screenshots")
# print("截图已保存到 ./device_screenshots 目录")
# 示例：批量启动应用（替换为实际包名和Activity）
# print("\n批量启动应用...")
# controller.batch_launch_app("com.android.settings", ".Settings")
# 保持主线程运行
print("\n" + "=" * 50)
print("群控系统运行中，按 Ctrl+C 退出程序")
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
controller.stop_heartbeat_monitor()
print("\n程序已正常退出")
