在for循环运行多个async任务时出现asyncio.run相关RuntimeError如何解决

报错原因 #

RuntimeError: asyncio.run() cannot be called from a running event loop 错误的核心是当前运行环境已经存在一个处于激活状态的asyncio事件循环，最常见的场景是你在Jupyter Notebook、IPython这类交互式环境中运行代码，这类环境本身默认就启动了事件循环，而asyncio.run()的设计用途是为普通Python脚本启动全新的事件循环，二者冲突就会抛出该错误。

另外你原始代码本身也存在逻辑问题，即便解决该报错也无法实现并发效果：你每创建一个任务就await asyncio.sleep(2)，所有任务实际是串行执行，完全没有利用并发能力；同时直接将同步搜索函数传入asyncio.create_task也不会生效，asyncio任务仅支持协程对象。

解决步骤 #

1. 修复事件循环冲突 #

• 如果你在Jupyter/IPython环境运行：直接删除asyncio.run(get_queries())，替换为await get_queries()即可直接运行
• 如果你在普通.py脚本中运行仍出现该报错：说明之前的代码逻辑残留了未关闭的事件循环，替换运行代码为：

loop = asyncio.get_event_loop()
loop.run_until_complete(get_queries())

2. 改写代码实现真正并发 #

以下是可直接运行的并发实现版本，兼容同步搜索函数，总耗时仅为单次搜索的2秒左右，而非原代码的10秒：

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor

# 模拟你的搜索客户端，可替换为实际的搜索对象
class GoogleSearch:
    def get_domain_description(self, query):
        time.sleep(2)
        return f"Google搜索{query}的结果"
class BingSearch:
    def get_domain_description(self, query):
        time.sleep(2)
        return f"Bing搜索{query}的结果"
class DuckduckgoSearch:
    def get_domain_description(self, query):
        time.sleep(2)
        return f"DuckDuckGo搜索{query}的结果"

google_search = GoogleSearch()
bing_search = BingSearch()
duckduckgo_search = DuckduckgoSearch()

queries = ['messi', 'ronaldo', 'iniesta', 'xavi', 'ramos']
engines = ['Google', 'Google', 'Bing', 'Duckduckgo', 'Google']

# 线程池用于执行同步搜索逻辑
executor = ThreadPoolExecutor(max_workers=5)

async def single_search(query, engine):
    loop = asyncio.get_running_loop()
    if engine == 'Google':
        result = await loop.run_in_executor(executor, google_search.get_domain_description, query)
    elif engine == 'Bing':
        result = await loop.run_in_executor(executor, bing_search.get_domain_description, query)
    elif engine == 'Duckduckgo':
        result = await loop.run_in_executor(executor, duckduckgo_search.get_domain_description, query)
    print(result)
    return result

async def get_queries():
    # 批量创建所有搜索任务
    tasks = [asyncio.create_task(single_search(q, e)) for q, e in zip(queries, engines)]
    # 并发等待所有任务执行完成
    await asyncio.gather(*tasks)

if __name__ == "__main__":
    # 普通脚本运行用这行
    asyncio.run(get_queries())
    # Jupyter环境运行删除上面一行，取消注释下面一行
    # await get_queries()

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