如何在上下文管理器的循环中异步执行代码？Playwright异步问题排查

你的问题核心是在for循环里逐个await异步方法，导致任务串行执行，所以打印语句会按顺序同步输出，没有真正实现异步并行。asyncio.gather确实是实现并行的正确方式，只要合理控制上下文管理器的范围，完全不会出现提前退出的问题，下面是具体解决方案：

问题根源 #

如果你的代码是类似下面这种写法，每个循环迭代都用await调用异步截图方法，本质是串行执行——必须等当前任务完成，才会进入下一次循环：

async def capture_screenshots(self, urls):
    async with async_playwright() as p:
        browser = await p.chromium.launch()
        for url in urls:
            # 每次循环都await，任务串行执行
            await self._single_screenshot(browser, url)
        await browser.close()

正确实现方式 #

把所有异步任务先收集起来，用asyncio.gather并行执行，同时确保Playwright的上下文（browser、playwright实例）在所有任务完成后再关闭：

完整示例代码 #

import asyncio
from playwright.async_api import async_playwright

class AsyncScreenshotter:
    # 单个URL的截图逻辑，抽成独立异步方法
    async def _capture_single(self, browser, url):
        page = await browser.new_page()
        await page.goto(url, wait_until="networkidle")
        print(f"完成截图: {url}")
        await page.screenshot(path=f"{url.split('//')[-1].replace('/', '_')}.png")
        await page.close()

    async def batch_capture(self, urls):
        # Playwright上下文管理器覆盖所有任务的生命周期
        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=True)
            # 收集所有异步任务，此时任务还未执行
            tasks = [self._capture_single(browser, url) for url in urls]
            # 并行执行所有任务，gather会等待所有任务完成才继续
            await asyncio.gather(*tasks)
            # 所有任务完成后再关闭浏览器
            await browser.close()

# 测试执行
async def main():
    screenshotter = AsyncScreenshotter()
    test_urls = [
        "https://example.com",
        "https://google.com",
        "https://github.com"
    ]
    await screenshotter.batch_capture(test_urls)

if __name__ == "__main__":
    asyncio.run(main())

关键说明 #

1. 上下文安全：asyncio.gather会阻塞直到所有传入的任务全部完成，所以await browser.close()和Playwright上下文管理器的退出，都要等到所有截图任务执行完毕，完全不会出现提前退出的问题。
2. 资源隔离：每个任务独立创建和关闭page，避免多个任务共用同一个page导致的冲突。
3. 异常处理：如果需要容忍个别任务失败，可在gather中添加return_exceptions=True参数，这样单个任务抛出异常不会中断其他任务的执行，还能收集到异常信息：

   results = await asyncio.gather(*tasks, return_exceptions=True)
   # 遍历results，处理成功结果或异常
   for res in results:
       if isinstance(res, Exception):
           print(f"任务失败: {res}")
   

内容的提问来源于stack exchange，提问作者Dukhi Atma