如何解决Python中aiohttp单例Session的并发重建冲突问题？

核心问题 #

你的代码并发冲突的根源在于：

• 锁未共享：create_session和get_session里的asyncio.Lock()都是局部变量，每个方法调用都会生成新锁，完全无法阻止多个请求同时进入临界区修改session。
• 锁释放逻辑错误：get_session中如果进入重建session的分支，lock.release()不会执行，导致锁永久持有，后续请求全部阻塞。
• 私有属性依赖不安全：直接访问session._connector._conns判断是否有活跃请求不可靠，私有属性可能随aiohttp版本变更，且无法覆盖所有请求处理中的场景（比如请求已发起但连接尚未建立）。
• 初始化异步不安全：__init__中用loop.create_task创建session，可能导致get_session被调用时session还未初始化完成。

修正后的代码 #

import asyncio
import datetime
import hashlib
import aiohttp
import logging

logger = logging.getLogger(__name__)
LOGIN = "your_login"
PASSWORD = "your_password"
URL = "your_base_url"

class Session:
    def __init__(self):
        self.session = None
        self.last_created = 0.0
        # 用实例级锁，确保所有方法共享同一锁
        self.lock = asyncio.Lock()
        # 初始化时触发session创建
        asyncio.create_task(self._initialize_session())

    async def _initialize_session(self):
        """初始化session（仅在启动时调用）"""
        await self._create_new_session()

    async def _create_new_session(self):
        """创建新session并登录，内部方法，需在锁保护下调用"""
        # 先创建新session并完成登录，再替换旧session，避免旧session被正在使用时关闭
        new_session = aiohttp.ClientSession(
            connector=aiohttp.TCPConnector(limit=50), 
            trust_env=True
        )
        try:
            payload = {
                "login": LOGIN,
                "password": hashlib.md5(PASSWORD.encode()).hexdigest()
            }
            async with new_session.post(f"{URL}/login", json=payload) as response:
                data = await response.json()
                if data['code'] != "0":
                    await new_session.close()
                    logger.error("Login failed, could not create session")
                    return False
        except Exception as e:
            await new_session.close()
            logger.error(f"Error creating session: {str(e)}")
            return False
        
        # 登录成功后，替换旧session并关闭
        old_session = self.session
        self.session = new_session
        self.last_created = datetime.datetime.now().timestamp()
        logger.debug(f"Created new session at {datetime.datetime.fromtimestamp(self.last_created)}")
        
        if old_session is not None:
            # 关闭旧session，aiohttp会自动等待所有活跃请求完成后再关闭
            await old_session.close()
            logger.debug("Closed old session")
        return True

    async def get_session(self):
        """获取可用的session，自动处理过期重建"""
        async with self.lock:
            current_time = datetime.datetime.now().timestamp()
            # 检查session是否未初始化或已过期（15分钟=900秒）
            if self.session is None or (current_time - self.last_created > 900):
                logger.debug(f"Session expired or not initialized, creating new one")
                # 尝试创建新session，失败则返回旧session（如果存在）
                if not await self._create_new_session():
                    return self.session
            return self.session

关键修改说明 #

1. 共享实例锁：将lock改为类的实例属性，所有方法共用同一锁，确保临界区操作互斥。
2. 安全的session替换逻辑：先创建并登录新session，成功后再替换旧session，最后关闭旧session——aiohttp的ClientSession.close()会等待所有活跃请求完成后再关闭，避免正在执行的请求被中断。
3. 锁的正确使用：用async with self.lock替代手动acquire()和release()，自动处理锁的释放，避免分支逻辑导致的锁泄漏。
4. 移除私有属性依赖：不再依赖_connector._conns判断活跃请求，改为基于时间戳判断过期，结合session替换时的安全关闭逻辑，确保旧请求不受影响。
5. 可靠的初始化：用_initialize_session确保启动时session被正确初始化，避免空session问题。

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