影刀RPA 手机自动化进阶多机协同与云控平台实战作者林焱写在前面单台手机自动化能做的事有限很多场景需要多台手机同时工作——批量养号、多账号操作、大规模数据采集。本文讲多机协同的核心架构以及如何用影刀 ADB搭建一个简单的云控平台。一、多机架构设计拼多多店群自动化报活动上架基本架构控制机电脑运行影刀 ├── USB Hub │ ├── 手机1ADB连接 │ ├── 手机2ADB连接 │ ├── 手机3ADB连接 │ └── 手机N... └── 无线网络ADB WiFi连接不占USB口 ├── 手机N1 └── 手机NM设备管理importsubprocessimportthreadingimporttimeimportjsonfromdatetimeimportdatetimeclassDeviceManager:多设备管理器def__init__(self):self.devices{}# {device_id: device_info}self.refresh_devices()defrefresh_devices(self):刷新已连接设备列表resultsubprocess.run(adb devices,shellTrue,capture_outputTrue,textTrue)linesresult.stdout.strip().split(\n)current_idsset()forlineinlines[1:]:# 跳过第一行if\tdeviceinline:device_idline.split(\t)[0].strip()current_ids.add(device_id)ifdevice_idnotinself.devices:# 新设备获取基本信息self.devices[device_id]self._get_device_info(device_id)# 移除已断开的设备disconnectedset(self.devices.keys())-current_idsfordevice_idindisconnected:print(f设备断开{device_id})delself.devices[device_id]returnlist(current_ids)def_get_device_info(self,device_id):获取设备基本信息defadb_shell(cmd):resultsubprocess.run(fadb -s{device_id}shell{cmd},shellTrue,capture_outputTrue,textTrue)returnresult.stdout.strip()info{id:device_id,model:adb_shell(getprop ro.product.model),android_version:adb_shell(getprop ro.build.version.release),status:idle,current_task:None,connected_at:datetime.now().isoformat()}# 获取屏幕分辨率size_outputadb_shell(wm size)importrematchre.search(r(\d)x(\d),size_output)ifmatch:info[screen_width]int(match.group(1))info[screen_height]int(match.group(2))print(f新设备接入{device_id}({info[model]}, Android{info[android_version]}))returninfodefget_idle_device(self):获取一台空闲设备fordevice_id,infoinself.devices.items():ifinfo.get(status)idle:returndevice_idreturnNonedefset_device_status(self,device_id,status,taskNone):更新设备状态ifdevice_idinself.devices:self.devices[device_id][status]status self.devices[device_id][current_task]taskdefget_all_status(self):获取所有设备状态概览status_summary{}fordevice_id,infoinself.devices.items():status_summary[device_id]{model:info.get(model),status:info.get(status),task:info.get(current_task)}returnstatus_summary# 初始化设备管理器managerDeviceManager()print(f当前已连接{len(manager.devices)}台设备)print(json.dumps(manager.get_all_status(),ensure_asciiFalse,indent2))二、任务队列与分发importqueueimportthreadingclassTaskQueue:任务队列支持多设备并发执行def__init__(self,device_manager:DeviceManager):self.queuequeue.Queue()self.device_managerdevice_manager self.results[]self.lockthreading.Lock()self.runningFalsedefadd_task(self,task_func,task_id,**kwargs):添加任务到队列self.queue.put({id:task_id,func:task_func,kwargs:kwargs})defworker(self,device_id):工作线程从队列取任务执行whileself.running:try:taskself.queue.get(timeout5)exceptqueue.Empty:continuetask_idtask[id]print(f设备{device_id}开始执行任务{task_id})self.device_manager.set_device_status(device_id,busy,task_id)start_timetime.time()try:resulttask[func](device_iddevice_id,**task[kwargs])statussuccessexceptExceptionase:resultstr(e)statusfailedprint(f设备{device_id}任务{task_id}失败{e})elapsedround(time.time()-start_time,2)withself.lock:self.results.append({task_id:task_id,device_id:device_id,status:status,result:result,elapsed_seconds:elapsed,completed_at:datetime.now().isoformat()})self.device_manager.set_device_status(device_id,idle)self.queue.task_done()print(f设备{device_id}完成任务{task_id}{elapsed}秒)defrun(self):启动任务分发self.runningTrue# 为每台设备创建一个工作线程threads[]fordevice_idinself.device_manager.devices.keys():tthreading.Thread(targetself.worker,args(device_id,),daemonTrue)t.start()threads.append(t)print(f设备{device_id}工作线程已启动)# 等待所有任务完成self.queue.join()self.runningFalseprint(f\n所有任务完成)successsum(1forrinself.resultsifr[status]success)print(f成功{success}/{len(self.results)})returnself.results实战示例多机并行采集defcollect_douyin_account_data(device_id,account_url):在指定设备上采集抖音账号数据print(f[{device_id}] 开始采集{account_url})# ADB操作函数defrun(cmd):subprocess.run(fadb -s{device_id}{cmd},shellTrue,capture_outputTrue)deftap(x,y):run(fshell input tap{x}{y})time.sleep(0.5)defscreenshot(path):run(fshell screencap -p /sdcard/screen.png)run(fpull /sdcard/screen.png{path})# 打开抖音run(shell am start -n com.ss.android.ugc.aweme/.main.MainActivity)time.sleep(3)# ... 导航到账号页面、采集数据# 具体操作根据实际抖音APP界面调整screenshot(fresult_{device_id}_{int(time.time())}.png)return{device_id:device_id,account:account_url,status:collected}# 准备任务列表account_urls[https://www.douyin.com/user/account1,https://www.douyin.com/user/account2,https://www.douyin.com/user/account3,# ... 更多账号]# 创建任务队列task_queueTaskQueue(manager)fori,urlinenumerate(account_urls):task_queue.add_task(collect_douyin_account_data,task_idftask_{i},account_urlurl)# 启动多机并行执行resultstask_queue.run()# 保存结果importopenpyxl wbopenpyxl.Workbook()wswb.active ws.append([任务ID,设备ID,状态,耗时(秒),完成时间])forrinresults:ws.append([r[task_id],r[device_id],r[status],r[elapsed_seconds],r[completed_at]])wb.save(多机采集结果.xlsx)三、设备状态监控importosdefmonitor_devices(interval30):定期监控设备状态保存监控日志log_filefdevice_monitor_{datetime.now().strftime(%Y%m%d)}.logwhileTrue:manager.refresh_devices()statusmanager.get_all_status()# 写入日志withopen(log_file,a,encodingutf-8)asf:timestampdatetime.now().strftime(%H:%M:%S)f.write(f[{timestamp}] 设备数量{len(status)}\n)fordevice_id,infoinstatus.items():f.write(f{device_id}:{info[model]}-{info[status]}\n)# 检查掉线设备iflen(status)0:print(⚠️ 所有设备已断开连接)# 发送告警通知time.sleep(interval)# 在后台线程中运行监控monitor_threadthreading.Thread(targetmonitor_devices,daemonTrue)monitor_thread.start()TEMU店群矩阵自动化运营核价报活动四、踩坑记录坑1USB Hub质量至关重要便宜的USB Hub供电不稳定手机充电状态下运行ADB会频繁断连。要用工业级有源USB Hub每个口独立供电推荐7口以上带适配器的型号。坑2多设备并发时ADB命令冲突多线程同时运行ADB命令时要确保每条命令都带-s device_id指定设备否则会操作错设备。坑3线程数与设备数要匹配每台设备对应一个工作线程线程数不要超过设备数。超过设备数的线程会空转消耗资源。坑4设备温度过高手机持续运行自动化任务会发热热保护会降频甚至关机。定期每2-3小时让设备休息15-20分钟或给设备加散热片。坑5无线ADB稳定性差ADB WiFi连接在网络波动时容易断开长时间运行的任务建议用USB有线连接。WiFi连接只适合临时操作或单次任务。总结多机协同的核心要素稳定的USB供电有源Hub ADB命令严格指定设备ID 线程池管理任务 设备状态监控。规模超过5台设备后任务队列 日志监控就变得必不可少。署名林焱