在Quart端点中调用aiohttp抛出远程主机强制关闭连接错误

问题现象 #

在Quart端点中使用aiohttp发起GET/POST请求时，即便遵循aiohttp官方教程实现，仍会抛出ConnectionResetError: [WinError 10054] An existing connection was forcibly closed by the remote host错误，完整异常栈如下：

Exception in callback _ProactorBasePipeTransport._call_connection_lost(None)
handle: <Handle _ProactorBasePipeTransport._call_connection_lost(None) created at C:\Program Files\WindowsApps\PythonSoftwareFoundation.Python.3.10_3.10.2800.0_x64__qbz5n2kfra8p0\lib\asyncio\proactor_events.py:109>
source_traceback: Object created at (most recent call last):
  File "C:\Users\devuser\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.10_qbz5n2kfra8p0\LocalCache\local-packages\Python310\site-packages\aiohttp\client.py", line 560, in _request
    await resp.start(conn)
  File "C:\Users\devuser\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.10_qbz5n2kfra8p0\LocalCache\local-packages\Python310\site-packages\aiohttp\client_reqrep.py", line 917, in start
    payload.on_eof(self._response_eof)
  File "C:\Users\devuser\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.10_qbz5n2kfra8p0\LocalCache\local-packages\Python310\site-packages\aiohttp\streams.py", line 167, in on_eof
    callback()
  File "C:\Users\devuser\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.10_qbz5n2kfra8p0\LocalCache\local-packages\Python310\site-packages\aiohttp\client_reqrep.py", line 952, in _response_eof
    self._connection.release()
  File "C:\Users\devuser\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.10_qbz5n2kfra8p0\LocalCache\local-packages\Python310\site-packages\aiohttp\connector.py", line 178, in release
    self._connector._release(
  File "C:\Users\devuser\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.10_qbz5n2kfra8p0\LocalCache\local-packages\Python310\site-packages\aiohttp\connector.py", line 663, in _release
    protocol.close()
  File "C:\Users\devuser\AppData\Local\Packages\PythonSoftwareFoundation.Python.3.10_qbz5n2kfra8p0\LocalCache\local-packages\Python310\site-packages\aiohttp\client_proto.py", line 63, in close
    transport.close()
  File "C:\Program Files\WindowsApps\PythonSoftwareFoundation.Python.3.10_3.10.2800.0_x64__qbz5n2kfra8p0\lib\asyncio\proactor_events.py", line 109, in close
    self._loop.call_soon(self._call_connection_lost, None)
Traceback (most recent call last):
  File "C:\Program Files\WindowsApps\PythonSoftwareFoundation.Python.3.10_3.10.2800.0_x64__qbz5n2kfra8p0\lib\asyncio\events.py", line 80, in _run
    self._context.run(self._callback, *self._args)
  File "C:\Program Files\WindowsApps\PythonSoftwareFoundation.Python.3.10_3.10.2800.0_x64__qbz5n2kfra8p0\lib\asyncio\proactor_events.py", line 165, in _call_connection_lost
    self._sock.shutdown(socket.SHUT_RDWR)
ConnectionResetError: [WinError 10054] An existing connection was forcibly closed by the remote host

推测是aiohttp与Quart的交互存在异常，相关代码如下：

异步GET请求方法 #

async def _call_api_get(self, url : str, json_data : dict = None) -> ApiResponse:
    try:
        async with aiohttp.ClientSession() as session:
            async with session.get(url, json=json_data) as resp:
                body = await resp.json() \
                    if resp.content_type == 'application/json' \
                    else await resp.text()
                api_return = ApiResponse(
                    status_code = resp.status,
                    body = body,
                    content_type = resp.content_type)

    except ConnectionResetError as ex:
        print("ConnectionResetError:", ex)

    except Exception as ex:
        api_return = ApiResponse(exception_msg = ex)

    return api_return

Quart端点方法 #

async def my_endpoint(self) -> Response:

    auth_url = 'http://www.google.co.uk'

    api_return_body : dict = {"key": "value"}

    api_response = await self._call_api_get(auth_url,
                                            json_data=auth_request)

已参考过同类问题，但仍未解决。

解决方案 #

1. 复用ClientSession而非每次请求创建 #

当前代码每次请求都新建ClientSession，频繁建立/销毁连接易触发远端连接关闭策略。建议全局复用一个ClientSession实例：

import aiohttp
from quart import Quart

app = Quart(__name__)
# 应用初始化时创建全局session
app.client_session = aiohttp.ClientSession()

# 应用关闭时销毁session
@app.teardown_appcontext
async def close_session(exception):
    await app.client_session.close()

# 修改请求方法
async def _call_api_get(self, url : str, json_data : dict = None) -> ApiResponse:
    api_return = None
    try:
        async with app.client_session.get(url, json=json_data) as resp:
            body = await resp.json() if resp.content_type == 'application/json' else await resp.text()
            api_return = ApiResponse(
                status_code=resp.status,
                body=body,
                content_type=resp.content_type)
    except ConnectionResetError as ex:
        print("ConnectionResetError:", ex)
        api_return = ApiResponse(exception_msg=str(ex))
    except Exception as ex:
        api_return = ApiResponse(exception_msg=str(ex))
    return api_return

2. 调整aiohttp连接池配置 #

通过配置连接池参数，避免连接被过早回收：

# 创建session时指定连接池配置
timeout = aiohttp.ClientTimeout(total=60)
connector = aiohttp.TCPConnector(
    limit=100,  # 连接池总大小
    limit_per_host=10,  # 单主机连接数限制
    keepalive_timeout=30  # 连接保持时长
)
app.client_session = aiohttp.ClientSession(timeout=timeout, connector=connector)

3. 切换Windows环境下的asyncio事件循环 #

Windows默认的ProactorEventLoop可能存在兼容性问题，切换为SelectorEventLoop：

import asyncio
from quart import Quart

asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
app = Quart(__name__)

4. 确保完整读取响应内容 #

未完整读取响应就释放连接可能触发错误，需确保响应内容读取完成后再结束上下文：

import json

async def _call_api_get(self, url : str, json_data : dict = None) -> ApiResponse:
    api_return = None
    try:
        async with app.client_session.get(url, json=json_data) as resp:
            # 先读取完整响应数据
            raw_data = await resp.read()
            body = json.loads(raw_data) if resp.content_type == 'application/json' else raw_data.decode('utf-8')
            api_return = ApiResponse(
                status_code=resp.status,
                body=body,
                content_type=resp.content_type)
    except ConnectionResetError as ex:
        print("ConnectionResetError:", ex)
        api_return = ApiResponse(exception_msg=str(ex))
    except Exception as ex:
        api_return = ApiResponse(exception_msg=str(ex))
    return api_return

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