Flask复用httpx AsyncClient时出现Event loop is closed错误求助

问题 #

我有一个简单的Flask服务，接收用户请求后向API发送POST请求，希望通过httpx实现异步调用。初始代码运行正常：

@app.route('/test' , methods=['GET' , "POST"])
async def send_notification():
    tasks = []
    async with httpx.AsyncClient() as client:
        for i in range(3):
            try:
                tasks.append( send_post_request(client, url_to_send_request_to , headers , data_to_send) )
            except Exception as e:
                return "Error occured" , 200
        await asyncio.gather(*tasks)
    
async def send_post_request(client, url_to_send_request_to , headers , data_to_send):
    await client.post(url , headers=headers , data=data)

但用户请求量极大，每次请求创建新客户端会占用大量服务器内存，因此我想复用预创建的客户端（生产环境计划创建客户端池，最多维持10个客户端），修改后的代码如下：

client = httpx.AsyncClient()

@app.route('/test' , methods=['GET' , "POST"])
async def send_notification():
    tasks = []

    for i in range(3):
        try:
           tasks.append( send_post_request(client, url_to_send_request_to , headers , data_to_send) )
        except Exception as e:
            return "Error occured" , 200
    await asyncio.gather(*tasks)
        
async def send_post_request(client, url_to_send_request_to , headers , data_to_send):
    await client.post(url , headers=headers , data=data)

修改后首次请求可正常处理，但第二次请求会抛出RuntimeError: Event loop is closed错误，需要分析原因并提供解决方案。

原因分析 #

默认的Flask开发服务器是同步的，当处理异步视图时，Flask会临时创建一个一次性事件循环来处理当前请求，请求完成后就会关闭这个事件循环。

你全局初始化的httpx.AsyncClient会绑定到第一个请求的事件循环上，当第二个请求到来时，原来的事件循环已经被销毁关闭，客户端无法再使用，因此触发Event loop is closed错误。另外，httpx.AsyncClient本身不是线程安全的，在Flask默认的多线程服务器环境下共享单个客户端实例也会引发问题。

解决方案 #

1. 使用异步Flask服务器（基础方案） #

切换到支持异步的WSGI服务器（如uvicorn或hypercorn），这类服务器会维持一个长期运行的事件循环，避免每次请求创建/销毁循环的问题，同时正确管理客户端生命周期：

步骤：

• 安装依赖：

pip install uvicorn httpx flask

• 修改代码：

from flask import Flask
import httpx
import asyncio

app = Flask(__name__)
client: httpx.AsyncClient | None = None

@app.before_first_request
async def init_client():
    global client
    # 配置连接池参数，控制并发连接数
    client = httpx.AsyncClient(
        limits=httpx.Limits(max_connections=10, max_keepalive_connections=5)
    )

@app.teardown_appcontext
async def close_client(exception=None):
    global client
    if client:
        await client.aclose()

@app.route('/test', methods=['GET', 'POST'])
async def send_notification():
    tasks = []
    if not client:
        return "Client not initialized", 500
    
    for i in range(3):
        try:
            tasks.append(send_post_request(client, url_to_send_request_to, headers, data_to_send))
        except Exception as e:
            return "Error occurred", 200
    
    await asyncio.gather(*tasks)
    return "Requests sent", 200

async def send_post_request(client, url, headers, data):
    await client.post(url, headers=headers, data=data)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=5000)

2. 实现客户端池（生产环境推荐） #

如果需要更精细的并发控制，结合asyncio.Semaphore限制客户端使用数量，同时利用httpx自身的连接池优化资源：

from flask import Flask
import httpx
import asyncio
from contextlib import asynccontextmanager

app = Flask(__name__)
# 限制最大并发请求数，对应客户端池容量
semaphore = asyncio.Semaphore(10)
# 全局共享客户端，配置连接池参数
client = httpx.AsyncClient(
    limits=httpx.Limits(max_connections=10, max_keepalive_connections=10)
)

@asynccontextmanager
async def get_client():
    async with semaphore:
        yield client

@app.route('/test', methods=['GET', 'POST'])
async def send_notification():
    tasks = []
    
    for i in range(3):
        try:
            tasks.append(send_post_request(url_to_send_request_to, headers, data_to_send))
        except Exception as e:
            return "Error occurred", 200
    
    await asyncio.gather(*tasks)
    return "Requests sent", 200

async def send_post_request(url, headers, data):
    async with get_client() as client:
        await client.post(url, headers=headers, data=data)

@app.teardown_appcontext
async def close_client(exception=None):
    await client.aclose()

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=5000)

3. 兼容同步服务器的方案（仅开发/临时场景） #

如果必须使用Flask默认的同步服务器，可以为每个请求创建客户端但复用连接池，比完全创建新实例更节省资源：

@app.route('/test', methods=['GET', 'POST'])
async def send_notification():
    tasks = []
    # 复用连接池，而非每次创建全新客户端
    async with httpx.AsyncClient(
        limits=httpx.Limits(max_connections=10, max_keepalive_connections=5)
    ) as client:
        for i in range(3):
            try:
                tasks.append(send_post_request(client, url_to_send_request_to, headers, data_to_send))
            except Exception as e:
                return "Error occurred", 200
        await asyncio.gather(*tasks)
    return "Requests sent", 200

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