Class GuestSessionPool

app/utils/guest_session_pool.py:187–610

匿名访客会话池，支持最小负载获取与失败替换。

class GuestSessionPool:
    """匿名访客会话池，支持最小负载获取与失败替换。"""

    def __init__(self, pool_size: int = 3):
        self.pool_size = max(1, pool_size)
        self._lock = Lock()
        self._sessions: Dict[str, GuestSession] = {}
        self._maintenance_task: Optional[asyncio.Task] = None
        self._http_client: Optional[httpx.AsyncClient] = None
        self._client_lock = asyncio.Lock()
        self._capacity_lock = asyncio.Lock()
        self._background_tasks: Set[asyncio.Task] = set()
        self._cleanup_parallelism = GUEST_CLEANUP_PARALLELISM
        self._maintenance_interval = GUEST_POOL_MAINTENANCE_INTERVAL_SECONDS

    async def _get_http_client(self) -> httpx.AsyncClient:
        """获取可复用的 HTTP 客户端，减少频繁建连开销。"""
        if self._http_client is not None:
            return self._http_client

        async with self._client_lock:
            if self._http_client is None:
                self._http_client = _build_async_client()
        return self._http_client

    async def _close_http_client(self):
        """关闭可复用的 HTTP 客户端。"""
        async with self._client_lock:
            client = self._http_client
            self._http_client = None

        if client is not None:
            await client.aclose()

    def _track_background_task(self, coro) -> asyncio.Task:
        """跟踪后台任务，避免清理阻塞前台重试路径。"""
        task = asyncio.create_task(coro)
        self._background_tasks.add(task)

        def _on_done(done_task: asyncio.Task):
            self._background_tasks.discard(done_task)
            try:
                done_task.result()
            except asyncio.CancelledError:
                pass
            except Exception as exc:
                logger.warning(f"⚠️ 匿名会话后台任务异常: {exc}")

        task.add_done_callback(_on_done)
        return task

    async def _wait_background_tasks(self):
        """等待当前已注册的后台任务结束。"""
        pending = list(self._background_tasks)
        if pending:
            await asyncio.gather(*pending, return_exceptions=True)

    async def _delete_sessions_concurrently(self, sessions: List[GuestSession]):