基于异步编程与Playwright的高效自动化任务处理与状态监控系统构建

1. 项目概述：为什么我们需要异步的Playwright？

如果你正在处理需要同时打开几十个网页、监控多个后台任务状态，或者构建一个需要实时响应用户交互的自动化系统，那么传统的同步Playwright脚本可能会让你感到力不从心。页面加载的等待、网络请求的延迟、元素定位的耗时，这些I/O密集型操作在同步模式下会形成阻塞，让你的脚本效率低下，资源利用率也大打折扣。这正是“高效异步任务处理与状态监控”这个主题的核心价值所在。

简单来说，这个项目就是教你如何将Playwright这个强大的浏览器自动化工具，与Python的asyncio异步编程模型深度结合，构建出一个能够并发执行大量任务、并能实时感知和响应任务状态变化的自动化系统。它不仅仅是为了“跑得更快”，更是为了构建更健壮、更智能、更能应对现代Web应用复杂性的自动化解决方案。无论是做大规模的数据采集、多账户的社交平台管理，还是复杂的Web应用E2E测试，掌握这套方法都能让你从“脚本小子”进阶为“自动化架构师”。

2. 核心思路拆解：异步架构与状态监控的设计哲学

2.1 从同步到异步的思维转变

很多从Selenium转过来的朋友，或者初次接触Playwright的开发者，很容易延续同步的思维模式：打开浏览器 -> 导航到页面 -> 等待元素 -> 执行操作 -> 关闭浏览器，一步接一步。这种模式简单直观，但在处理多个独立任务时，问题就暴露出来了：任务A在等待网络时，CPU和整个线程都在空转，任务B、C只能干等着。

异步编程的核心思想是“在等待时去做别的事”。当你的Playwright脚本在page.goto()等待页面加载，或者在page.wait_for_selector()等待一个动态元素出现时，异步事件循环（Event Loop）会挂起这个协程（Coroutine），转而去执行其他已经就绪的协程，比如处理另一个页面的点击操作，或者解析之前已经加载完成页面的数据。这种非阻塞的I/O操作，使得单个Python进程就能轻松管理数十个甚至上百个浏览器上下文（Context）或页面（Page），极大地提升了硬件资源的利用率和任务吞吐量。

2.2 状态监控：从被动等待到主动感知

在同步脚本中，“状态监控”往往等同于“轮询”（Polling）：写一个循环，每隔几秒检查一下某个元素是否存在、某个文本是否改变。这种方式不仅低效（很多检查是无效的），而且响应延迟高。

在异步架构下，我们可以实现更优雅的“响应式”状态监控。Playwright提供了丰富的事件监听器（如page.on('request'),page.on('response'),page.on('console')），结合异步编程，我们可以为每个页面或任务挂载这些监听器。当特定事件（如某个API接口返回了成功状态码、页面出现了错误提示框、控制台输出了特定日志）发生时，监听器会立即触发一个异步回调函数。这个回调函数可以分析事件数据，更新任务状态，甚至触发后续的自动化操作。这种模式将监控从“我每隔一段时间问你好了没”变成了“你好了立刻告诉我”，实时性和效率有质的飞跃。

2.3 核心组件与数据流设计

一个高效的异步Playwright系统通常包含以下几个核心组件：

1. 任务队列：一个异步安全的队列（如asyncio.Queue），用于存放待执行的任务参数。
2. 工作者池：一组并发的异步协程（Worker），每个工作者从队列中获取任务，独立执行Playwright操作。
3. 状态管理中心：一个全局可访问的数据结构（如字典或专门的状态类），用于记录每个任务的实时状态（如“等待中”、“执行中”、“成功”、“失败及原因”）。
4. 事件监听与分发器：集成在工作者内部的Playwright事件监听逻辑，负责捕获页面级事件并更新到状态管理中心。
5. 监控与报告协程：一个独立的协程，定期或由事件驱动地检查状态管理中心，生成日志、报告或触发告警。

数据流大致如下：主程序将任务推入队列 -> 空闲工作者领取任务 -> 工作者初始化Playwright页面并设置事件监听 -> 执行具体操作，监听器异步更新任务状态 -> 任务完成，工作者清理资源并标记任务状态 -> 监控协程发现任务完成，进行后续处理。

3. 环境搭建与核心工具选型

3.1 Playwright异步API与async_playwright

Playwright从一开始就为异步编程提供了原生支持。关键入口是async_playwright()异步上下文管理器。与同步的sync_playwright()不同，它返回的对象需要在async with块内使用，并且所有方法都是awaitable的。

import asyncio from playwright.async_api import async_playwright async def main(): async with async_playwright() as p: # 选择浏览器，推荐chromium，平衡了功能、性能和兼容性 browser = await p.chromium.launch(headless=False) # 调试时可关闭无头模式 context = await browser.new_context() page = await context.new_page() # 所有的导航、点击、等待都需要await await page.goto('https://example.com') await page.screenshot(path='example.png') await browser.close() asyncio.run(main())

工具选型解析：

• 浏览器选择：p.chromium是最佳默认选择。它在Playwright中支持特性最全，性能优化最好，且无需额外安装系统依赖。p.firefox和p.webkit适用于特定兼容性测试场景。
• 启动参数：headless=False在开发调试时至关重要，你能看到浏览器实际行为。在生产环境，务必设置为headless=True以节省资源。slow_mo参数可以减慢操作速度，方便观察，但会严重影响并发性能，仅用于调试。
• 上下文管理：使用async with管理async_playwright()和browser.new_context()是最佳实践，它能确保即使发生异常，浏览器和上下文资源也能被正确关闭，避免资源泄漏。

3.2 异步生态的基石：asyncio与相关库

asyncio是Python标准库，是我们构建整个异步系统的框架。除了基本的run、create_task、gather，我们还需要掌握几个关键组件：

• asyncio.Queue：这是实现生产者-消费者模式、控制并发度的核心。我们可以设置队列的最大容量，防止内存被无限增长的任务撑爆。
• asyncio.Semaphore：信号量用于限制同时执行的协程数量，是一种更轻量级的并发控制手段，特别适合限制同时打开的浏览器页面数，防止对目标服务器造成过大压力或被封禁。
• asyncio.Event或asyncio.Condition：用于协程间的通知和状态同步。例如，一个监控协程可以等待一个Event，当有任务失败时，工作者协程设置这个Event，监控协程被唤醒并处理告警。

对于更复杂的任务依赖和调度，可以考虑第三方库如celery（但需配合消息队列），但对于纯asyncio环境，aiojobs或asyncio原生组件通常足够。

3.3 状态存储与可视化选型

状态存储的选择取决于系统复杂度：

• 简单场景（内存存储）：直接使用Python字典或dataclass。优点是零延迟，实现简单。缺点是程序重启后状态丢失，且不适合分布式部署。
• 中等场景（持久化存储）：使用SQLite（通过aiosqlite库实现异步操作）或Redis（通过aioredis）。SQLite适合状态结构复杂、需要简单查询的场景；Redis适合读写频繁、需要设置过期时间或发布/订阅通知的场景。
• 复杂分布式场景：考虑使用消息队列（如RabbitMQ/Kafka）传递状态事件，配合专门的数据库（如PostgreSQL,MongoDB）进行持久化。

对于可视化，如果需要一个简单的Web仪表盘，可以使用异步Web框架如FastAPI或Sanic，创建一个实时推送任务状态列表的API端点，前端用Vue.js或React配合WebSocket或SSE（服务器发送事件）实现实时更新。

4. 实战构建：一个高效异步任务处理引擎

4.1 定义任务与状态模型

首先，我们需要明确任务是什么，以及它有哪些状态。一个好的模型是系统健壮性的基础。

from dataclasses import dataclass, field from enum import Enum from typing import Any, Optional import time class TaskStatus(Enum): PENDING = "pending" RUNNING = "running" SUCCESS = "success" FAILED = "failed" TIMEOUT = "timeout" @dataclass class AsyncTask: """异步任务数据类""" task_id: str url: str action: str # 例如：'screenshot', 'extract_data', 'monitor' parameters: dict[str, Any] = field(default_factory=dict) status: TaskStatus = TaskStatus.PENDING result: Optional[Any] = None error_message: Optional[str] = None created_at: float = field(default_factory=time.time) started_at: Optional[float] = None finished_at: Optional[float] = None def mark_running(self): self.status = TaskStatus.RUNNING self.started_at = time.time() def mark_success(self, result: Any): self.status = TaskStatus.SUCCESS self.result = result self.finished_at = time.time() def mark_failed(self, error_msg: str): self.status = TaskStatus.FAILED self.error_message = error_msg self.finished_at = time.time()

设计要点：

• task_id：必须全局唯一，可以使用uuid.uuid4().hex生成。这是追踪任务的唯一标识。
• status：使用枚举类型，比字符串更规范，避免拼写错误。
• 时间戳：created_at,started_at,finished_at对于监控任务性能、计算耗时、排查长时间运行任务至关重要。
• 操作方法：mark_running等方法封装了状态转换逻辑，保证了状态变更的一致性，未来如果需要添加日志或触发钩子函数，只需修改这里。

4.2 实现任务工作者（Worker）

工作者是执行具体Playwright操作的协程。它的核心是从队列中消费任务，管理浏览器页面的生命周期，并处理异常。

import asyncio from playwright.async_api import Page, Response import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class PlaywrightWorker: def __init__(self, worker_id: int, max_pages: int = 3): self.worker_id = worker_id self.semaphore = asyncio.Semaphore(max_pages) # 控制并发页面数 self.browser = None async def setup(self): """初始化浏览器实例，所有工作者可共享一个浏览器，但建议每个工作者独立上下文""" playwright = await async_playwright().start() # 为每个工作者创建独立的浏览器实例，避免上下文间cookie、存储污染 self.browser = await playwright.chromium.launch( headless=True, args=['--disable-blink-features=AutomationControlled'] # 隐藏自动化特征 ) logger.info(f"Worker {self.worker_id}: Browser launched.") async def _execute_task(self, task: AsyncTask, page: Page): """执行单个任务的核心逻辑""" task.mark_running() logger.info(f"Worker {self.worker_id}: Processing task {task.task_id} for {task.url}") try: # 示例：导航并截图 if task.action == 'screenshot': # 设置超时和等待策略 response = await page.goto(task.url, wait_until='networkidle', timeout=30000) if not response or not response.ok: raise Exception(f"Page load failed with status: {getattr(response, 'status', 'unknown')}") # 可能需要在加载后等待特定元素 await page.wait_for_selector('body', state='attached', timeout=5000) screenshot_path = f"screenshots/{task.task_id}.png" await page.screenshot(path=screenshot_path, full_page=True) task.mark_success({'screenshot_path': screenshot_path}) # 可以扩展更多action，如 extract_data, login, monitor 等 else: raise ValueError(f"Unknown action: {task.action}") except Exception as e: logger.error(f"Worker {self.worker_id}: Task {task.task_id} failed. Error: {e}") task.mark_failed(str(e)) # 可以在这里保存错误时的页面截图或HTML，便于调试 try: await page.screenshot(path=f"errors/{task.task_id}_error.png") except: pass async def run(self, task_queue: asyncio.Queue, state_manager: 'StateManager'): """工作者主循环""" await self.setup() while True: task = await task_queue.get() logger.debug(f"Worker {self.worker_id}: Got task {task.task_id} from queue.") async with self.semaphore: # 限制该工作者同时处理的页面数 context = await self.browser.new_context( viewport={'width': 1920, 'height': 1080}, user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ...' ) page = await context.new_page() # !!! 关键步骤：挂载事件监听器以实现状态监控 !!! self._attach_monitors(page, task, state_manager) try: await self._execute_task(task, page) finally: # 无论成功失败，都清理页面和上下文，防止内存泄漏 await page.close() await context.close() task_queue.task_done() # 将更新后的任务状态同步到状态管理器 state_manager.update_task(task) def _attach_monitors(self, page: Page, task: AsyncTask, state_manager: 'StateManager'): """为页面挂载事件监听器，实现细粒度状态监控""" # 监控网络请求失败（如404, 500） async def on_request_failed(request): failure = request.failure if failure: error_msg = f"Request failed: {request.url} - {failure}" # 可以更新任务的子状态或记录日志 state_manager.record_event(task.task_id, 'request_failed', {'url': request.url, 'error': error_msg}) page.on('requestfailed', on_request_failed) # 监控控制台错误 async def on_console(message): if message.type == 'error': error_msg = f"Console error: {message.text}" state_manager.record_event(task.task_id, 'console_error', {'text': message.text}) # 如果是关键错误，可以立即标记任务失败 # if "Fatal" in message.text: # task.mark_failed(error_msg) page.on('console', on_console) # 监控特定API响应（例如，监控一个数据接口是否返回成功） async def on_response(response: Response): if '/api/data' in response.url and response.status == 200: # 假设我们监控的API成功返回 try: data = await response.json() if data.get('code') == 0: # 根据实际API设计判断 state_manager.record_event(task.task_id, 'api_success', data) except: pass page.on('response', on_response)

工作者实现详解：

1. 并发控制：self.semaphore限制了单个工作者同时打开的页面数。这是防止内存溢出的关键，也避免对目标网站造成过大压力。通常设置为3-5个比较合理。
2. 资源隔离：每个任务使用独立的context和page。这确保了任务间的Cookie、本地存储、缓存完全隔离，互不干扰，模拟了真实用户独立会话的行为。
3. 生命周期管理：try...finally块确保了即使任务执行过程中出现异常，页面和上下文资源也一定会被关闭。这是避免资源泄漏的黄金法则。
4. 事件监听：_attach_monitors方法是状态监控的灵魂。它为非阻塞的事件（网络请求、控制台日志）绑定了异步回调函数。这些回调函数在事件发生时被事件循环调用，可以实时更新任务状态或记录日志。注意，回调函数内部的操作必须是异步的，且要快速完成，不能阻塞事件循环。

4.3 构建状态管理器（StateManager）

状态管理器负责集中存储和提供任务状态查询接口。在简单实现中，它可能只是一个内存字典的包装；在复杂系统中，它可能连接着数据库。

import asyncio from typing import Dict, List, Optional import json class StateManager: def __init__(self): self._tasks: Dict[str, AsyncTask] = {} self._task_events: Dict[str, List[dict]] = {} # 记录每个任务的关键事件 self._lock = asyncio.Lock() # 异步锁，保证状态更新的线程安全 async def update_task(self, task: AsyncTask): """更新任务状态（异步安全）""" async with self._lock: self._tasks[task.task_id] = task # 这里可以触发状态变更的钩子，例如通知WebSocket、写入数据库等 if task.status in (TaskStatus.SUCCESS, TaskStatus.FAILED, TaskStatus.TIMEOUT): logger.info(f"Task {task.task_id} finished with status: {task.status}") async def get_task(self, task_id: str) -> Optional[AsyncTask]: """获取任务状态""" async with self._lock: return self._tasks.get(task_id) async def get_all_tasks(self, status_filter: Optional[TaskStatus] = None) -> List[AsyncTask]: """获取所有任务，可选状态过滤""" async with self._lock: tasks = list(self._tasks.values()) if status_filter: tasks = [t for t in tasks if t.status == status_filter] return tasks async def record_event(self, task_id: str, event_type: str, data: dict): """记录任务运行时事件（如网络错误、API响应）""" async with self._lock: if task_id not in self._task_events: self._task_events[task_id] = [] self._task_events[task_id].append({ 'timestamp': time.time(), 'type': event_type, 'data': data }) logger.debug(f"Event recorded for task {task_id}: {event_type}") async def get_task_events(self, task_id: str) -> List[dict]: """获取任务的所有事件日志""" async with self._lock: return self._task_events.get(task_id, [])

状态管理器要点：

• 异步锁：asyncio.Lock是必须的。因为多个工作者协程会并发地调用update_task和record_event，如果没有锁，对共享字典的并发修改可能导致数据错乱或程序崩溃。
• 事件记录：_task_events记录了任务生命周期的详细日志，这对于调试复杂问题（比如“任务为什么失败了？”）有巨大帮助。你可以看到失败前最后一个网络请求是什么，控制台输出了什么错误。
• 扩展性：update_task方法内的注释指出了扩展点。当任务状态变为终态（成功/失败）时，你可以在这里触发异步通知，比如通过WebSocket推送到前端仪表盘，或者将任务结果异步写入PostgreSQL数据库。

4.4 组装引擎与主程序

最后，我们将所有组件组装起来，并创建一个主程序来调度一切。

import uuid import signal class AsyncPlaywrightEngine: def __init__(self, num_workers: int = 5, max_queue_size: int = 1000): self.num_workers = num_workers self.task_queue = asyncio.Queue(maxsize=max_queue_size) self.state_manager = StateManager() self.workers = [] self.monitor_task = None self.is_running = False async def submit_task(self, task: AsyncTask): """提交任务到队列""" # 如果队列已满，这里会等待直到有空位 await self.task_queue.put(task) await self.state_manager.update_task(task) # 初始状态为PENDING logger.info(f"Task {task.task_id} submitted.") async def start_workers(self): """启动工作者池""" for i in range(self.num_workers): worker = PlaywrightWorker(worker_id=i) self.workers.append(worker) # 将worker.run作为后台任务运行 asyncio.create_task(worker.run(self.task_queue, self.state_manager)) logger.info(f"Started {self.num_workers} workers.") async def start_monitor(self): """启动监控协程，定期检查超时任务和生成报告""" async def monitor_loop(): while self.is_running: await asyncio.sleep(30) # 每30秒检查一次 all_tasks = await self.state_manager.get_all_tasks() now = time.time() for task in all_tasks: # 检查运行超时的任务（假设超时时间为5分钟） if task.status == TaskStatus.RUNNING and task.started_at: if now - task.started_at > 300: task.mark_failed("Task execution timeout") await self.state_manager.update_task(task) logger.warning(f"Task {task.task_id} timed out.") # 生成简单报告 pending = len([t for t in all_tasks if t.status == TaskStatus.PENDING]) running = len([t for t in all_tasks if t.status == TaskStatus.RUNNING]) success = len([t for t in all_tasks if t.status == TaskStatus.SUCCESS]) failed = len([t for t in all_tasks if t.status == TaskStatus.FAILED]) logger.info(f"[Monitor] Tasks - Pending: {pending}, Running: {running}, Success: {success}, Failed: {failed}") self.monitor_task = asyncio.create_task(monitor_loop()) async def run(self): """启动引擎""" self.is_running = True await self.start_workers() await self.start_monitor() logger.info("Async Playwright Engine is now running.") async def graceful_shutdown(self, signal=None): """优雅关闭，等待队列中任务完成""" logger.info("Shutdown signal received. Draining queue...") self.is_running = False if self.monitor_task: self.monitor_task.cancel() # 等待队列中所有任务被处理完 await self.task_queue.join() logger.info("All tasks processed. Engine shutdown complete.") # 主程序示例 async def main(): engine = AsyncPlaywrightEngine(num_workers=3) # 设置信号处理，用于优雅关闭（如Ctrl+C） loop = asyncio.get_running_loop() for sig in (signal.SIGINT, signal.SIGTERM): loop.add_signal_handler(sig, lambda: asyncio.create_task(engine.graceful_shutdown())) # 启动引擎 await engine.run() # 模拟提交一批任务 sample_urls = [ 'https://httpbin.org/html', 'https://httpbin.org/json', 'https://httpbin.org/xml', # ... 更多URL ] for i, url in enumerate(sample_urls): task = AsyncTask( task_id=f"task_{i}_{uuid.uuid4().hex[:8]}", url=url, action='screenshot', parameters={'quality': 90} ) await engine.submit_task(task) await asyncio.sleep(0.5) # 控制提交速率 # 主程序可以在这里等待，或者运行其他逻辑 # 例如，运行一个简单的HTTP API服务器来查询状态 try: await asyncio.Future() # 永久等待，直到被关闭信号打断 except asyncio.CancelledError: pass finally: await engine.graceful_shutdown() if __name__ == "__main__": asyncio.run(main())

主程序解析：

1. 优雅关闭：graceful_shutdown是生产环境必备。它先设置停止标志，然后等待task_queue.join()，这意味着主程序会阻塞直到队列中所有被取出的任务都执行完毕（task_done()被调用）。这避免了强制退出导致的任务丢失。
2. 信号处理：通过asyncio.get_running_loop().add_signal_handler捕获SIGINT（Ctrl+C）和SIGTERM信号，并触发优雅关闭流程。这是编写常驻异步服务的标准做法。
3. 监控循环：monitor_loop是一个独立的后台协程。它定期检查任务状态，例如发现运行时间过长的任务并标记为超时失败。这是状态监控的主动轮询部分，是对事件监听（被动触发）的补充。
4. 任务提交：submit_task方法将任务放入队列并立即更新状态为PENDING。通过await asyncio.sleep(0.5)控制提交速率，是一种简单的限流，防止瞬间提交大量任务压垮队列或工作者。

5. 高级技巧与深度优化

5.1 性能调优：让并发飞起来

默认配置下，Playwright和异步引擎可能仍有优化空间。

• 浏览器启动参数优化：

  browser = await p.chromium.launch( headless=True, # 禁用不必要的功能以加速启动和减少内存 args=[ '--disable-gpu', '--disable-dev-shm-usage', # 在Docker等受限环境很有用 '--disable-setuid-sandbox', '--no-sandbox', # 仅在信任的环境中使用，有一定安全风险 '--disable-blink-features=AutomationControlled', '--disable-extensions', ] )

• 上下文复用与池化：对于大量相似任务（如仅cookie不同），可以考虑上下文池。预先创建一批BrowserContext，工作者从池中取用，用完归还。这比为每个任务创建/销毁上下文更快。但要注意隔离性，确保归还前清理了敏感数据。
• 连接复用与CDP会话：Playwright通过Chrome DevTools Protocol与浏览器通信。极端性能场景下，可以考虑手动管理CDP会话，但复杂度极高，一般不建议。
• 调整wait_until策略：page.goto(url, wait_until='load')比'networkidle'或'commit'更快，但页面可能还未完全渲染。根据你的任务需求选择最宽松且足够的策略。如果只是触发一个请求，'commit'可能就足够了。

5.2 稳定性保障：错误处理与重试机制

网络不稳定、网站反爬、动态内容加载失败是常态。健壮的系统必须有完善的错误处理和重试。

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type import aiohttp class RobustPlaywrightWorker(PlaywrightWorker): @retry( stop=stop_after_attempt(3), # 最多重试3次 wait=wait_exponential(multiplier=1, min=2, max=10), # 指数退避等待 retry=retry_if_exception_type((aiohttp.ClientError, TimeoutError)), # 仅对网络/超时错误重试 reraise=True # 重试次数用尽后，抛出原始异常 ) async def _load_page_with_retry(self, page: Page, url: str, timeout: int = 30000): """带重试的页面加载方法""" response = await page.goto(url, wait_until='networkidle', timeout=timeout) if not response or not response.ok: # 对于非2xx/3xx状态码，也触发重试（可选） raise aiohttp.ClientResponseError( status=response.status, message=f"HTTP {response.status} for {url}" ) return response async def _execute_task(self, task: AsyncTask, page: Page): task.mark_running() try: # 使用重试逻辑加载页面 await self._load_page_with_retry(page, task.url) # ... 后续操作 except Exception as e: logger.error(f"Task {task.task_id} failed after retries: {e}") task.mark_failed(str(e))

这里使用了tenacity库，它提供了强大且灵活的重试装饰器。wait_exponential实现了指数退避，避免在服务器临时故障时加重其负担。

5.3 动态内容应对策略

这是现代Web自动化最大的挑战之一。反爬措施、动态加载、前端框架（React, Vue, Angular）都会导致元素选择器失效。

• 智能等待与选择器策略：
  • 避免绝对定位：不要依赖xpath=//div[3]/div[5]/span[2]这种脆弱的定位。优先使用>await page.wait_for_function(""" () => { const el = document.querySelector('.my-component'); return el && el.offsetHeight > 0 && el.innerText.includes('Loaded'); } """, timeout=10000)
• 处理Shadow DOM：如果元素在Shadow Root内部，需要使用page.locator('...').shadow_root.locator('...')的链式调用。
• 应对反自动化检测：
  • args中设置'--disable-blink-features=AutomationControlled'。
  • 使用context.add_init_script注入脚本，覆盖navigator.webdriver等属性。
  • 模拟人类行为：随机延迟、移动鼠标轨迹（page.mouse.move()）、随机滚动。但要注意，过度模拟会降低性能。

5.4 资源监控与告警

一个成熟的系统需要知道自己的健康状态。

• 内存监控：使用psutil库定期检查Python进程的内存占用。如果持续增长，可能存在内存泄漏（例如，页面或上下文未正确关闭）。

  import psutil process = psutil.Process() memory_mb = process.memory_info().rss / 1024 / 1024 if memory_mb > 1024: # 超过1GB logger.warning(f"High memory usage: {memory_mb:.2f} MB")

• 队列积压告警：在监控循环中检查task_queue.qsize()。如果队列持续增长，远大于工作者处理速度，说明系统过载，需要增加工作者或暂停提交任务。
• 外部状态推送：将关键指标（任务成功率、平均耗时、队列长度）通过statsd、Prometheus客户端推送到监控系统（如Grafana），实现可视化监控和告警。

6. 常见问题排查与实战心得

6.1 问题速查表

6.2 实战心得与避坑指南

1. “无头模式”是你的朋友，也是你的敌人：生产环境务必用headless=True。但调试时，一定要用headless=False亲眼看看发生了什么。很多诡异问题（如元素点击无效）在可视化模式下运行一次就一目了然。可以使用slow_mo=100（100毫秒延迟）让操作慢下来，方便观察。

2. 选择器是门艺术，不要依赖录制工具：Playwright的代码录制功能很棒，但生成的选择器往往非常脆弱（充满div:nth-child(5)）。录制后，一定要手动优化选择器。优先使用get_by_role、get_by_text、get_by_test_id这些语义化、更稳定的定位方式。

3. 异步上下文管理是资源泄漏的重灾区：务必使用async with来管理async_playwright()、browser、context和page。如果因为逻辑复杂不能使用async with，那么必须在try...finally块中手动await page.close()和await browser.close()。一个常见的错误是只在成功路径关闭资源，异常路径却漏了。

4. 不要忽视日志，它是你唯一的“黑匣子”：为你的工作者、状态管理器、任务执行过程添加不同级别的日志（DEBUG,INFO,WARNING,ERROR）。使用logging模块配置输出到文件，并设置日志轮转。当线上任务莫名其妙失败时，详细的日志是你排查问题的唯一线索。特别是在事件监听器的回调里，记录下关键信息。

5. 分布式扩展的思路：当单机性能达到瓶颈，需要考虑分布式。核心是将任务队列和状态管理器抽离出来，成为独立服务（如使用Redis作为队列和状态存储）。每个工作节点（可以是不同的机器）从共享队列中拉取任务，执行后将状态写回共享存储。这时，主程序就变成了一个“调度器”和“监控器”。Celery或Dramatiq这类分布式任务队列库可以简化这个过程，但它们与asyncio的集成需要仔细考量，arq是一个基于asyncio和Redis的纯异步选择。

构建这样一个高效的异步Playwright任务处理与监控系统，初看有些复杂，但一旦搭建完成，它将成为一个极其强大和灵活的基础设施。你可以用它来构建爬虫集群、自动化测试平台、社交媒体管理工具等等。关键在于理解各个组件（异步编程、Playwright API、状态管理）是如何协同工作的，并从简单的版本开始，逐步迭代，增加你需要的功能。