使用Python的asyncio向3个独立端点POST相同数据遇错误求助

错误原因 #

你遇到的RuntimeError: This event loop is already running，大概率是在Jupyter Notebook/IPython这类交互式环境中运行代码导致的——这类环境会自动启动asyncio事件循环，当你手动调用loop.run_until_complete(main())时，就会和已运行的循环冲突。如果是普通脚本环境，这段代码本身不会报错，但存在串行等待的效率问题。

方案一：修复现有代码（基于requests+run_in_executor） #

1. 补全缺失依赖与变量 #

你的代码缺少time模块导入，且url1/url2/url3、sample_data、header1/header2/header3未定义，先补上这些基础内容。

2. 替换循环启动方式 #

用Python 3.7+内置的asyncio.run()替代手动获取循环并调用run_until_complete，它会自动处理事件循环的创建与关闭，避免冲突。

3. 优化任务等待逻辑 #

原代码逐个await任务会变成串行执行，改用asyncio.gather()可以并行等待所有任务，真正发挥异步的优势。

修改后的代码：

import asyncio
import requests
import time

# 替换为你的实际配置
url1 = "http://endpoint1.example.com"
url2 = "http://endpoint2.example.com"
url3 = "http://endpoint3.example.com"
sample_data = {"key": "value"}
header1 = {"Content-Type": "application/json"}
header2 = {"Content-Type": "application/json"}
header3 = {"Content-Type": "application/json"}

async def main():
    loop = asyncio.get_event_loop()
    # 批量创建任务
    tasks = [
        loop.run_in_executor(None, lambda: requests.post(url=url1, json=sample_data, headers=header1)),
        loop.run_in_executor(None, lambda: requests.post(url=url2, json=sample_data, headers=header2)),
        loop.run_in_executor(None, lambda: requests.post(url=url3, json=sample_data, headers=header3))
    ]
    # 并行等待所有任务完成
    model1_response, model2_response, model3_response = await asyncio.gather(*tasks)
    
    # 示例：打印响应状态码
    print(f"Model1 响应状态: {model1_response.status_code}")
    print(f"Model2 响应状态: {model2_response.status_code}")
    print(f"Model3 响应状态: {model3_response.status_code}")

tm1 = time.perf_counter()
asyncio.run(main())
tm2 = time.perf_counter()
print(f'Total time elapsed: {tm2-tm1:0.2f} seconds')

方案二：改用异步HTTP库aiohttp（更地道的asyncio用法） #

requests是同步库，用run_in_executor只是把同步任务丢到线程池，并非真正的异步IO。推荐用aiohttp实现纯异步请求，效率更高：

import asyncio
import aiohttp
import time

# 替换为你的实际配置
url1 = "http://endpoint1.example.com"
url2 = "http://endpoint2.example.com"
url3 = "http://endpoint3.example.com"
sample_data = {"key": "value"}
header1 = {"Content-Type": "application/json"}
header2 = {"Content-Type": "application/json"}
header3 = {"Content-Type": "application/json"}

async def post_request(session, url, data, headers):
    async with session.post(url, json=data, headers=headers) as response:
        return await response.text(), response.status

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [
            post_request(session, url1, sample_data, header1),
            post_request(session, url2, sample_data, header2),
            post_request(session, url3, sample_data, header3)
        ]
        results = await asyncio.gather(*tasks)
        for idx, (resp_text, status) in enumerate(results, 1):
            print(f"Model{idx} 响应状态: {status}, 响应内容: {resp_text[:50]}...")

tm1 = time.perf_counter()
asyncio.run(main())
tm2 = time.perf_counter()
print(f'Total time elapsed: {tm2-tm1:0.2f} seconds')

使用前需安装aiohttp：

pip install aiohttp

交互式环境额外提示 #

如果必须在Jupyter/IPython中运行，可直接用await main()替代asyncio.run(main())——环境已存在运行的循环，直接await即可。

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