影刀RPA店群自动化系统：任务生命周期钩子与浏览器资源优雅回收架构

影刀RPA店群自动化系统：任务生命周期钩子与浏览器资源优雅回收架构

在这里插入图片描述

启动一个自动化任务很简单。

让它优雅地结束，把占用的资源全部还回去，才是工程化的分水岭。

在店群自动化项目跑了大半年之后，我们统计过一次资源泄漏的情况。
结果触目惊心：每运行48小时，系统中会残留大约15%的浏览器进程、20%的临时文件句柄，以及若干未释放的Redis连接。

这些问题在单个任务执行时完全看不出来。
但当任务量累积到数千次，节点内存慢慢被蚕食，最终只能靠人工定期重启续命。

我们决定从根上解决这个问题：为每一个自动化任务设计完整的生命周期钩子，确保无论任务成功、失败还是被终止，所有资源都能被回收。

一、任务生命周期：从一维到多维

大多数人理解的任务生命周期很简单：开始 → 执行 → 结束。

但在分布式自动化系统中，一个任务的实际生命周期远比这复杂。
它会经历多个阶段，每个阶段都可能发生异常退出，且资源申请分散在不同阶段。

我们重新定义了任务的七个生命周期状态：

CREATED → VALIDATED → RESOURCE_ACQUIRED → RUNNING → COMPLETED ↘ FAILED ↘ CANCELLED ``` 每个状态之间，都有明确的钩子点。 状态转换不可跳跃，资源申请与释放必须对称。 --- ## 二、钩子机制的设计：让资源管理可编程 我们在Python调度层实现了一套任务钩子框架，允许在不同的生命周期节点注册回调。 ```python from enum import Enum from typing import Callable, Dict, List import asyncio class TaskLifecycle(Enum): ON_CREATE = "on_create" ON_VALIDATE = "on_validate" ON_RESOURCE_ACQUIRED = "on_resource_acquired" ON_START = "on_start" ON_SUCCESS = "on_success" ON_FAILURE = "on_failure" ON_CANCEL = "on_cancel" ON_FINAL = "on_final" # 无论成败都会执行 class TaskHooks: def __init__(self): self._hooks: Dict[str, List[Callable]] = { state.value: [] for state in TaskLifecycle } def register(self, lifecycle: TaskLifecycle, callback: Callable): self._hooks[lifecycle.value].append(callback) async def trigger(self, lifecycle: TaskLifecycle, context: dict): for callback in self._hooks[lifecycle.value]: try: await callback(context) except Exception as e: logger.error(f"Hook {lifecycle.value} error: {e}") ``` 每个任务实例在创建时，都会挂载资源申请钩子和资源释放钩子。 **资源申请放在 `ON_RESOURCE_ACQUIRED` 阶段。** **资源释放统一放在 `ON_FINAL` 阶段。** 不管任务正常完成、失败还是被手动取消，`ON_FINAL` 都会被触发。 这保证了资源释放是必达的。 --- ## 三、浏览器资源的精细化管理 浏览器资源是最容易泄漏的。 在早期的实现里，任务结束后我们只是调用 `driver.quit()`。 但在并发场景下，有些浏览器进程并不会真正退出，子进程残留、临时文件未清理。 我们后来实现了一套更完整的浏览器实例生命周期管理。 ```python class BrowserResourceManager: def __init__(self, shop_id: str, pool): self.shop_id = shop_id self.pool = pool self.instance = None self.acquired = False async def acquire(self): self.instance = await self.pool.acquire(self.shop_id) self.acquired = True return self.instance async def release(self): if not self.acquired or not self.instance: return try: # 清理页面状态，但不关闭浏览器 await self.instance.reset_to_blank() # 归还给实例池 await self.pool.release(self.shop_id) except Exception as e: # 如果归还有问题，强制销毁并重建 logger.warning(f"Release failed for {self.shop_id}, force destroy") await self.instance.force_kill() await self.pool.rebuild_instance(self.shop_id) finally: self.acquired = False self.instance = None ``` 每个任务在 `ON_RESOURCE_ACQUIRED` 钩子中调用 `acquire`，在 `ON_FINAL` 钩子中调用 `release`。 这种对称设计，让浏览器资源泄漏率降到了零。 --- ## 四、进程与文件句柄的兜底清理 有些资源不是通过池管理的，比如： - 影刀流程启动的临时子进程 - - 任务执行中创建的临时JSON文件 - - 日志写入的文件句柄 我们为每个任务维护了一个资源清单。 任务执行过程中申请的任何外部资源，都要登记到这个清单中。 `ON_FINAL` 钩子执行时，遍历清单，逐一释放。 ```python class TaskResourceTracker: def __init__(self, task_id): self.task_id = task_id self._resources = { "processes": [], "files": [], "redis_keys": [], "temp_dirs": [] } def track_process(self, pid: int): self._resources["processes"].append(pid) def track_file(self, path: str): self._resources["files"].append(path) def track_redis_key(self, key: str): self._resources["redis_keys"].append(key) async def cleanup_all(self): for pid in self._resources["processes"]: try: proc = psutil.Process(pid) proc.terminate() proc.wait(timeout=5) except (psutil.NoSuchProcess, psutil.TimeoutExpired): try: proc.kill() except: pass for file_path in self._resources["files"]: try: Path(file_path).unlink(missing_ok=True) except Exception as e: logger.warning(f"Failed to delete {file_path}: {e}") for key in self._resources["redis_keys"]: try: await redis.delete(key) except: pass for temp_dir in self._resources["temp_dirs"]: try: shutil.rmtree(temp_dir, ignore_errors=True) except: pass self._resources = {k: [] for k in self._resources} ``` **这个机制就像一个任务结束后的“清道夫”，不管流程中间发生了什么，它都会把现场打扫干净。** --- ## 五、优雅关闭：系统级生命周期管理 单个任务的资源回收做扎实了，还需要考虑整个系统的优雅关闭。 当运维需要停机维护、或者Master节点收到操作系统关闭信号时，不能粗暴地杀掉所有进程。 正在执行的任务应该被允许在限定时间内完成，或者被打断但保证资源释放。 我们在Master和Worker节点上都注册了信号处理： ```python import signal import asyncio class GracefulShutdown: def __init__(self, scheduler, browser_pool, task_tracker): self.scheduler = scheduler self.browser_pool = browser_pool self.task_tracker = task_tracker self.shutdown_event = asyncio.Event() def register_signals(self): loop = asyncio.get_event_loop() for sig in (signal.SIGINT, signal.SIGTERM): loop.add_signal_handler(sig, self._handle_signal) def _handle_signal(self): logger.info("Received shutdown signal, starting graceful shutdown...") self.shutdown_event.set() async def shutdown(self, timeout=60): # 第一步：停止接收新任务 self.scheduler.pause() logger.info("Scheduler paused, no new tasks") # 第二步：等待当前任务完成，最长等待timeout秒 try: await asyncio.wait_for( self._wait_all_tasks(), timeout=timeout ) except asyncio.TimeoutError: logger.warning(f"Tasks not finished within {timeout}s, forcing cancellation") # 第三步：强制取消剩余任务，触发所有ON_FINAL钩子 await self.scheduler.cancel_all_running() logger.info("All tasks cancelled, hooks triggered") # 第四步：释放所有浏览器实例 await self.browser_pool.close_all() logger.info("Browser pool closed") # 第五步：清理Redis连接等 await self.task_tracker.cleanup_all() logger.info("System gracefully shut down") async def _wait_all_tasks(self): while self.scheduler.has_running_tasks(): await asyncio.sleep(1) ``` 有了这套机制，运维终于敢在白天执行部署了。 因为知道系统会自己妥善处理掉正在跑的任务，不会留下一地狼藉。 --- ## 六、容器化思维：进程级资源配额 虽然我们最终没有完全容器化（之前文章也讨论过），但容器化思维影响了资源管理策略。 我们为每个浏览器实例设置了Windows Job Object配额： - 进程内存上限：2GB - - CPU使用率上限：50%（单核等价） - - 允许的进程数上限：10（包括子进程） 当浏览器进程超出配额时，Job Object会强制终止它，而不是让整台机器受影响。 任务钩子中的 `ON_FAILURE` 会捕获这种异常，将任务标记为资源超限失败，并触发重建浏览器实例。 ```python import win32job import win32process def create_job_with_limits(memory_mb=2048, cpu_percent=50): job = win32job.CreateJobObject(None, "") info = win32job.QueryInformationJobObject(job, win32job.JobObjectExtendedLimitInformation) info['BasicLimitInformation']['LimitFlags'] = ( win32job.JOB_OBJECT_LIMIT_PROCESS_MEMORY | win32job.JOB_OBJECT_LIMIT_JOB_MEMORY | win32job.JOB_OBJECT_LIMIT_DIE_ON_UNHANDLED_EXCEPTION ) info['ProcessMemoryLimit'] = memory_mb * 1024 * 1024 info['JobMemoryLimit'] = memory_mb * 1024 * 1024 win32job.SetInformationJobObject(job, win32job.JobObjectExtendedLimitInformation, info) return job def assign_process_to_job(job_handle, pid): handle = win32api.OpenProcess(win32con.PROCESS_SET_QUOTA | win32con.PROCESS_TERMINATE, False, pid) win32job.AssignProcessToJobObject(job_handle, handle) ``` 这些配额机制确保了单个任务行为再离谱，也不会把节点拖垮。 --- ## 七、监控与长期验证 资源回收不能靠“感觉”来验证。 我们编写了每日资源泄漏检测脚本，在业务低峰期（凌晨3点）运行： - 检查所有Worker的浏览器进程数是否与实例池配置匹配 - - 检查临时目录大小，超过阈值自动清理 - - 检查孤儿Redis键（带任务前缀但无对应运行中任务的键），自动回收 这些检查结果写入Elasticsearch，形成长期趋势图。 如果某天浏览器残留数突然升高，运维能第一时间收到告警。 --- ## 八、写在最后 自动化系统做到最后，拼的不是谁功能多。 而是谁在无人值守的情况下，能更长久地稳定运行。 任务生命周期钩子和资源回收机制，就像城市的下水道系统。 平时没人注意，一旦堵了，整个城市都会瘫痪。 > 让每一个被创建的资源，都有明确的销毁路径。 > > 这不是过度设计，而是自动化工程的基本功。 --- *作者：林焱*