如何在Python版Azure Functions中异步实例化、共享并关闭aiohttp会话

嘿，这个问题我太懂了——每次请求新建ClientSession确实挺折腾的，毕竟建立连接池、初始化这些操作开销不小，完全不符合aiohttp的最佳实践。针对Python版Azure Functions里要复用异步会话的需求，咱们可以这么一步步来搞定：

核心思路 #

Azure Functions会复用函数实例来处理后续的请求，所以我们可以借助全局变量在同一个实例的多次调用间共享ClientSession，同时要做好会话的初始化和关闭逻辑，避免资源泄漏。

具体实现代码 #

首先在你的函数文件顶部定义全局变量和辅助函数，然后修改主函数逻辑：

import aiohttp
import azure.functions as func
import asyncio
import atexit

# 全局存储会话对象，初始为None
_session: aiohttp.ClientSession | None = None

async def get_session() -> aiohttp.ClientSession:
    """懒加载获取会话，确保只创建一次"""
    global _session
    if _session is None or _session.closed:
        # 只在会话不存在/已关闭时新建
        _session = aiohttp.ClientSession()
    return _session

def close_session_on_exit():
    """函数实例回收时关闭会话，防止资源泄漏"""
    global _session
    if _session is not None and not _session.closed:
        # 异步关闭会话需要用asyncio.run同步执行
        asyncio.run(_session.close())

# 注册退出钩子，Azure回收实例时自动触发
atexit.register(close_session_on_exit)

async def main(req: func.HttpRequest) -> func.HttpResponse:
    try:
        # 获取复用的会话
        session = await get_session()
        async with session.get("你的目标请求URL") as response:
            response_content = await response.text()
            return func.HttpResponse(response_content)
    except Exception as e:
        return func.HttpResponse(body=f"请求处理出错: {str(e)}", status_code=500)

关键细节解释 #

• 懒加载初始化：不能在全局作用域直接异步创建会话（Azure Functions启动是同步的，异步代码无法在这里正确执行），所以用get_session函数在第一次异步调用时才初始化会话，后续调用直接复用。
• 会话复用逻辑：全局变量_session会在同一个函数实例的多次请求间保留，只要实例没被Azure回收，就一直复用这个会话，避免重复创建的开销。
• 资源清理：通过atexit.register注册关闭钩子，当Azure决定回收函数实例时，会自动调用close_session_on_exit来关闭会话，释放底层的连接资源，防止内存泄漏。

额外注意 #

如果你的Azure Functions用的是隔离运行时（Isolated Runtime），生命周期处理逻辑会稍有不同，但核心思路还是通过全局变量复用会话，只是关闭钩子的注册方式需要调整为适配隔离运行时的生命周期事件。

备注：内容来源于stack exchange，提问作者Konstantin