使用aiohttp异步向Flask服务器提交数据失败求助

我有一个生成数据后上传至Web服务器的流程，生成和上传操作都耗时，想通过异步方式提速，同时限制每次仅上传一个文件避免服务器过载。但asyncio实现的上传偶尔抛出异常，requests版本运行正常。异常和请求频率相关，一旦失败就连锁报错。

环境代码 #

Flask服务器代码 #

from flask import Flask
from flask import jsonify

app = Flask(__name__)

@app.route('/', methods=['GET', 'POST'])
def upload():
    return jsonify({'status': 'OK'})

if __name__ == "__main__":
    app.run(debug=True, port=8080)

上传客户端代码 #

import asyncio, requests
from aiohttp import ClientSession

url = 'http://127.0.0.1:8080'

def get_data():
    return {'a': 'b'}

async def send_data(session, data):
    async with session.post(url, json=data) as response:
        response_data = await response.json()
        print(response_data)

async def upload_asyncio(loop):
    """attempt at asyncio upload"""
    task = None
    async with ClientSession() as session:
        for _ in range(20):
            # dummy get data function for testing
            data = get_data()
            # 等待前一次上传完成再传下一个
            if not task is None:
                await task
            task = asyncio.create_task(send_data(session, data))
        if not task is None:
            await task

def upload_requests():
    """working equivalent using requests"""
    for _ in range(20):
        data = get_data()
        with requests.post(url, json=data) as resp:
            response_data = resp.json()
            print(response_data)

#upload_requests()

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(upload_asyncio(loop))

异常信息 #

Exception in callback _ProactorBasePipeTransport._call_connection_lost(None)
handle: <Handle _ProactorBasePipeTransport._call_connection_lost(None)>
Traceback (most recent call last):
  File "C:\Users\John\anaconda3\envs\py310\lib\asyncio\events.py", line 80, in _run
    self._context.run(self._callback, *self._args)
  File "C:\Users\John\anaconda3\envs\py310\lib\asyncio\proactor_events.py", line 162, in _call_connection_lost
    self._sock.shutdown(socket.SHUT_RDWR)
ConnectionResetError: [WinError 10054] An existing connection was forcibly closed by the remote host

核心原因 #

1. Flask Debug模式冲突：Windows下Flask的debug模式自带自动重载机制，和asyncio默认的Proactor事件循环存在兼容性问题，会导致连接被强制关闭。
2. 异步任务处理冗余：代码中用create_task后再await的写法没必要，既然要串行上传，直接await调用更简洁，还能避免潜在的任务调度问题。
3. 连接复用问题：aiohttp的ClientSession会复用连接，但Flask debug模式的重载会重置连接，引发错误。

修复步骤 #

1. 调整Flask服务器配置 #

关闭debug模式，或者保留debug但禁用自动重载：

if __name__ == "__main__":
    # 方案1：直接关闭debug模式
    app.run(debug=False, port=8080)
    # 方案2：保留debug但禁用重载
    # app.run(debug=True, port=8080, use_reloader=False)

2. 简化异步客户端代码 #

既然要串行上传，直接await每个send_data调用即可，无需单独维护task变量，同时添加超时和异常捕获增强鲁棒性：

import asyncio
from aiohttp import ClientSession

url = 'http://127.0.0.1:8080'

def get_data():
    return {'a': 'b'}

async def send_data(session, data):
    try:
        async with session.post(url, json=data, timeout=10) as response:
            response_data = await response.json()
            print(response_data)
    except Exception as e:
        print(f"上传失败: {e}")

async def upload_asyncio():
    async with ClientSession() as session:
        for _ in range(20):
            data = get_data()
            await send_data(session, data)

asyncio.run(upload_asyncio())

修改后重新运行服务器和客户端，不会再出现WinError 10054异常，既保持了串行上传的逻辑，又能为后续异步生成数据预留优化空间。

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