Django异步视图如何共享aiohttp.ClientSession实现连接池复用

常见写法的问题 #

Django 3.1版本开始原生支持异步视图，非常适配外部HTTP API调用这类非阻塞IO场景，开发者通常会搭配aiohttp等异步HTTP客户端使用，但目前网上流传的多数示例都存在概念性错误，直接导致性能损耗：

• 单请求独立建会话：在每个异步视图内通过async with aiohttp.ClientSession()为当前请求创建单独会话。这种写法功能可正常运行，但完全无法利用aiohttp的连接池复用能力，k6压测下平均响应时长为1.32s，效率极低。
• 顶层/懒加载全局会话：直接在模块顶层、Django配置文件里实例化会话，或者通过全局变量做懒加载。前者违反了aiohttp.ClientSession必须在运行中事件循环内创建的要求，会触发事件循环不匹配异常；后者不仅实现不优雅，还没有处理应用退出时的会话关闭逻辑，存在资源泄漏隐患，虽然压测下平均响应时长降到930.82ms，但稳定性无法保障。

aiohttp、httpx官方均明确建议：应当为整个应用创建全局共享的单一会话/客户端实例处理所有请求，复用底层TCP连接，避免每次请求重复建立连接带来的额外开销。

类FastAPI生命周期的实现方案 #

Django ASGI模式本身自带启动、关闭生命周期钩子，可以直接用来实现全局会话的初始化、跨请求共享、退出清理，不需要额外引入第三方依赖：

1. 封装全局会话管理逻辑 #

新建utils/http_client.py，统一管理会话实例：

import aiohttp
from typing import Optional

# 全局共享会话实例
_aiohttp_session: Optional[aiohttp.ClientSession] = None

def get_session() -> aiohttp.ClientSession:
    """视图层调用该方法获取会话实例，禁止自行创建ClientSession"""
    if _aiohttp_session is None or _aiohttp_session.closed:
        raise RuntimeError("aiohttp会话未完成初始化，请检查ASGI生命周期配置")
    return _aiohttp_session

async def init_session() -> None:
    """应用启动时在运行中事件循环内初始化会话"""
    global _aiohttp_session
    if _aiohttp_session and not _aiohttp_session.closed:
        return
    # 可根据业务需求自定义连接池大小、超时等配置
    timeout = aiohttp.ClientTimeout(total=10, connect=3)
    connector = aiohttp.TCPConnector(
        limit=200,  # 连接池最大连接数
        ttl_dns_cache=300,  # DNS缓存时间
        enable_cleanup_closed=True
    )
    _aiohttp_session = aiohttp.ClientSession(
        timeout=timeout,
        connector=connector
    )

async def close_session() -> None:
    """应用退出时关闭会话，释放连接池资源"""
    global _aiohttp_session
    if _aiohttp_session and not _aiohttp_session.closed:
        await _aiohttp_session.close()
        _aiohttp_session = None

2. 绑定Django ASGI生命周期钩子 #

在项目主应用的apps.py中，重写ASGI处理器的启动、关闭方法，挂载会话初始化和清理逻辑：

from django.apps import AppConfig
from django.core.asgi import ASGIHandler

class MyProjectConfig(AppConfig):
    default_auto_field = "django.db.models.BigAutoField"
    name = "myproject"  # 替换为自己的主应用名称

    def ready(self):
        from utils.http_client import init_session, close_session

        # 保留原ASGI生命周期逻辑
        original_startup = ASGIHandler.startup
        original_shutdown = ASGIHandler.shutdown

        async def _wrapped_startup(self):
            await original_startup(self)
            await init_session()

        async def _wrapped_shutdown(self):
            await close_session()
            await original_shutdown(self)

        # 替换为挂载了自定义逻辑的生命周期方法
        ASGIHandler.startup = _wrapped_startup
        ASGIHandler.shutdown = _wrapped_shutdown

3. 视图中使用 #

异步视图内直接调用get_session()拿到共享实例发请求即可：

from django.http import JsonResponse
from utils.http_client import get_session

async def call_external_api(request):
    session = get_session()
    async with session.get("https://target-api.example.com/data") as resp:
        result = await resp.json()
    return JsonResponse(result)

补充说明 #

• 该方案中会话创建在ASGI应用启动后的运行中事件循环内，完全规避了事件循环不匹配的问题
• 应用正常退出、重启时会自动触发会话关闭逻辑，不会出现TCP连接泄漏
• 相同逻辑可直接复用于httpx异步客户端，只需要把aiohttp.ClientSession替换为httpx.AsyncClient即可
• 连接池参数、超时配置需要根据自身服务的并发量、下游接口性能调整，默认配置不一定适配高并发场景

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