集成aiohttp示例到自有应用时报This event loop is already running等问题

错误根因 #

• 事件循环重复调用：asyncio.run() 本身就会创建事件循环并运行传入的协程直到执行结束，你在已经处于运行状态的协程_run_async内部，又调用了loop.run_until_complete()，等于尝试在一个已经跑起来的事件循环里再次启动循环，直接触发运行时错误。
• 协程未被等待调度：asyncio.wait()和自定义的_fetch方法都是协程对象，必须用await关键字才能调度执行，你没有加await，自然会抛出协程未等待的警告。
• 多余的循环管理逻辑：你自己写的_get_loop方法、手动关闭循环的逻辑完全多余，asyncio.run()会自动处理事件循环的创建、绑定、销毁全流程，不需要手动干预。

修复后可运行代码 #

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import asyncio
import aiohttp


class FetchAsync:
    def __init__(self):
        pass

    async def _receive_via_aiohttp(self, session, url, headers):
        async with session.get(url, headers=headers) as response:
            content = await response.read()
            return response, content

    async def _fetch(self,
                     url,
                     session):
        headers = {'User-Agent': 'MyAgent'}

        # use aiohttp to get feed/xml content and response object
        response, content = await self._receive_via_aiohttp(session,
                                                            url,
                                                            headers)
        print(f"请求{url}完成，状态码：{response.status}")
        # do a lot more stuff...
        return content

    def run(self):
        # 直接用asyncio.run管理整个协程生命周期，不需要手动操作loop
        asyncio.run(self._run_async())

    async def _run_async(self):
        async with aiohttp.ClientSession() as session:
            # in real there are much more URLs
            urls = ['https://cnn.com',
                    'https://fsfe.org']

            # 创建任务列表
            tasks = [asyncio.create_task(self._fetch(url, session))
                       for url
                       in urls]

            # 直接await等待所有任务执行完成，不需要调用loop的方法
            done, pending = await asyncio.wait(tasks)
            # 后续可以遍历done获取每个任务的返回值
            for task in done:
                print(task.result())


if __name__ == '__main__':
    obj = FetchAsync()
    obj.run()

结构调整说明 #

• 删掉了冗余的_get_loop方法和手动关闭循环的逻辑，全量用asyncio.run管理事件循环生命周期
• 把任务列表改成了asyncio.create_task包裹的任务对象，调度更规范
• 直接用await asyncio.wait()等待所有任务执行完成，不需要在协程内部再调用事件循环的执行方法
• 如果你的场景需要限制并发量，后续可以用asyncio.Semaphore来做并发控制，避免请求量太高被目标站点封禁

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