如何在 Django 异步视图之间共享（初始化和关闭）aiohttp.ClientSession 以使用连接池

Question

Django 从 3.1 版本开始支持异步视图，因此它非常适合对外部 HTTP API 等非阻塞调用（例如使用aiohttp）。

\n

我经常看到以下代码示例，我认为它在概念上是错误的（尽管它工作得很好）：

\n

import aiohttp\nfrom django.http import HttpRequest, HttpResponse\n\nasync def view_bad_example1(request: HttpRequest):\n    async with aiohttp.ClientSession() as session:\n        async with session.get("https://example.com/") as example_response:\n            response_text = await example_response.text()\n            return HttpResponse(response_text[:42], content_type="text/plain")\n

\n

此代码ClientSession为每个传入请求创建一个，效率低下。aiohttp然后不能使用例如连接池。

\n

\n

不要\xe2\x80\x99t为每个请求创建一个会话。您很可能需要每个应用程序有一个会话来共同执行所有请求。

\n

来源：https ://docs.aiohttp.org/en/stable/client_quickstart.html#make-a-request

\n

\n

这同样适用于 httpx：

\n

\n

另一方面，客户端实例使用 HTTP 连接池。这意味着当您向同一主机发出多个请求时，客户端将重用底层 TCP 连接，而不是为每个请求重新创建一个连接。

\n

来源：https://www.python-httpx.org/advanced/#why-use-a-client

\n

\n

有什么方法可以aiohttp.ClientSession在 Django 中全局实例化，以便可以在多个请求之间共享该实例吗？不要忘记必须ClientSession在正在运行的事件循环中创建（为什么在事件循环之外创建 ClientSession 很危险？），因此我们不能在 Django 设置中实例化它或作为模块级变量实例化它。

\n

我得到的最接近的是这段代码。但是，我认为这段代码很丑陋，并且没有解决例如关闭会话的问题。

\n

CLIENT_SESSSION = None\n\nasync def view_bad_example2(request: HttpRequest):\n    global CLIENT_SESSSION\n\n    if not CLIENT_SESSSION:\n        CLIENT_SESSSION = aiohttp.ClientSession()\n\n    example_response = await CLIENT_SESSSION.get("https://example.com/")\n    response_text = await example_response.text()\n\n    return HttpResponse(response_text[:42], content_type="text/plain")\n

\n

基本上我正在寻找来自 FastAPI 的事件的等效项，可用于在异步上下文中创建/关闭某些资源。

\n

顺便说一句，这是两个视图之间使用 k6 的性能比较：

\n

\n
• view_bad_example1：avg=1.32s min=900.86ms med=1.14s max=2.22s p(90)=2s p(95)=2.1s
\n
• view_bad_example2：avg=930.82ms min=528.28ms med=814.31ms max=1.66s p(90)=1.41s p(95)=1.52s
\n

\n

Answer 1

Django 不实现 ASGI Lifespan 协议。
参考：https: //github.com/django/django/pull/13636

斯塔莱特确实如此。FastAPI 直接使用 Starlette 的事件处理程序实现。

以下是使用 Django 实现这一目标的方法：

1. 在 Django 的子类中实现 ASGI Lifespan 协议ASGIHandler。

import django
from django.core.asgi import ASGIHandler


class MyASGIHandler(ASGIHandler):
    def __init__(self):
        super().__init__()
        self.on_shutdown = []

    async def __call__(self, scope, receive, send):
        if scope['type'] == 'lifespan':
            while True:
                message = await receive()
                if message['type'] == 'lifespan.startup':
                    # Do some startup here!
                    await send({'type': 'lifespan.startup.complete'})
                elif message['type'] == 'lifespan.shutdown':
                    # Do some shutdown here!
                    await self.shutdown()
                    await send({'type': 'lifespan.shutdown.complete'})
                    return
        await super().__call__(scope, receive, send)

    async def shutdown(self):
        for handler in self.on_shutdown:
            if asyncio.iscoroutinefunction(handler):
                await handler()
            else:
                handler()


def my_get_asgi_application():
    django.setup(set_prefix=False)
    return MyASGIHandler()

2. 替换applicationasgi.py中的。

# application = get_asgi_application()
application = my_get_asgi_application()

3. 实现一个助手get_client_session来共享实例：

import asyncio
import aiohttp
from .asgi import application

CLIENT_SESSSION = None

_lock = asyncio.Lock()


async def get_client_session():
    global CLIENT_SESSSION

    async with _lock:
        if not CLIENT_SESSSION:
            CLIENT_SESSSION = aiohttp.ClientSession()
            application.on_shutdown.append(CLIENT_SESSSION.close)

    return CLIENT_SESSSION

用法：

async def view(request: HttpRequest):
    session = await get_client_session()
    
    example_response = await session.get("https://example.com/")
    response_text = await example_response.text()

    return HttpResponse(response_text[:42], content_type="text/plain")