perf(http): 异步 HTTP 引入共享 AsyncHTTPTransport，复用 TCP/TLS 握手

AsyncRequestUtils 使用按事件循环弱引用持有的共享 AsyncHTTPTransport 作为底层连接池与 TLS 会话；每次请求创建轻量 AsyncClient 承载本次 cookie jar、timeout、follow_redirects，用完即销毁。共享 transport 由 _NonClosingTransportProxy 包装后注入 AsyncClient，吞掉 AsyncClient 退出时向底层 transport 传播的 __aexit__/aclose，使底层连接池跨调用持久，从而真正复用 TCP/TLS 握手。设计要点： - 共享 transport 桶按 (proxy, verify, max_keepalive_connections, max_connections, keepalive_expiry) 区分；每事件循环 32 桶 LRU 上限，超出后异步关闭最久未用桶；关闭 task 由模块级强引用集合持有以兼容 Python 3.11+ 的任务 GC 行为。 - 通过 FastAPI lifespan shutdown 调用 aclose_shared_async_transports 集中释放底层 transport，避免 ResourceWarning。 - AsyncRequestUtils.request 走三条 path：用户自管 client / 共享 transport + per-call AsyncClient / 兜底临时 client。三条路径 cookie 语义一致；后两条因 per-call AsyncClient 生命周期局限于单次调用，天然不积累 Set-Cookie，避免跨调用 jar 演化串扰。 - _make_request 对幂等方法（GET/HEAD/OPTIONS）在 RemoteProtocolError / ReadError / WriteError 时单次重试，容忍 keep-alive stale 连接命中；非幂等方法不重试，但记录 debug 日志。 - get_stream 使用 httpx.AsyncClient.stream() 标准流式 API，与 request 共用三条 path 的 client 选择逻辑；幂等单次重试； yield 体异常透传给 stream 的 __aexit__。公共 API 表面零变动。插件可通过 max_keepalive_connections / max_connections / keepalive_expiry 三个 limits 参数为自己定制连接池容量与握手有效期。 TMDB 真实压测（10 部美剧 × 每部 50 集，1020 请求）： 61.96s → 18.15s（3.41×），单请求 p95 149.6ms → 38.1ms。
2026-09-08 08:57:09 +08:00 · 2026-05-11 08:46:40 +08:00
parent 888ccfcfc2
commit c745616495
2 changed files with 595 additions and 212 deletions
@@ -13,6 +13,7 @@ from app.startup.plugins_initializer import init_plugins, stop_plugins, sync_plu
 from app.startup.routers_initializer import init_routers
 from app.startup.scheduler_initializer import stop_scheduler, init_scheduler, init_plugin_scheduler
 from app.startup.workflow_initializer import init_workflow, stop_workflow
+from app.utils.http import aclose_shared_async_transports


 async def init_extra():
@@ -83,3 +84,5 @@ async def lifespan(app: FastAPI):
        stop_plugins()
        # 停止模块
        await stop_modules()
+        # 关闭共享的异步 HTTP 连接池，释放底层连接资源
+        await aclose_shared_async_transports()
@@ -1,6 +1,10 @@
+import asyncio
+import collections
 import re
 import sys
-from contextlib import contextmanager, asynccontextmanager
+import threading
+import weakref
+from contextlib import AsyncExitStack, contextmanager, asynccontextmanager
 from pathlib import Path
 from typing import Any, Optional, Tuple, Union

@@ -18,25 +22,169 @@ from app.log import logger
 urllib3.disable_warnings(InsecureRequestWarning)


+class _NonClosingTransportProxy(httpx.AsyncBaseTransport):
+    """
+    包装共享底层 transport，转发请求但吞掉 __aexit__/aclose 调用。
+    防止 per-call AsyncClient 在 async with 退出时把底层连接池一并清空。
+    底层 transport 的真正关闭由 aclose_shared_async_transports() 统一管理。
+    """
+
+    __slots__ = ("_wrapped",)
+
+    def __init__(self, wrapped: httpx.AsyncBaseTransport):
+        self._wrapped = wrapped
+
+    async def __aenter__(self):  # pragma: no cover - 简单转发
+        return self
+
+    async def __aexit__(self, exc_type=None, exc_value=None, traceback=None) -> None:
+        # 故意 no-op：不向底层 transport 传播 __aexit__，避免连接池被清空
+        return None
+
+    async def aclose(self) -> None:
+        # 故意 no-op：调用方显式 aclose 也不影响共享池
+        return None
+
+    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
+        return await self._wrapped.handle_async_request(request)
+
+
+_SharedTransportKey = Tuple[
+    Optional[str],          # proxy
+    Union[bool, str],       # verify
+    int,                    # max_keepalive_connections
+    int,                    # max_connections
+    int,                    # keepalive_expiry
+]
+# 共享底层 transport 桶，按事件循环和配置区分，支持 LRU 淘汰
+_shared_async_transports: weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, collections.OrderedDict[_SharedTransportKey, httpx.AsyncHTTPTransport]] = weakref.WeakKeyDictionary()
+# 不同线程各自驱动的事件循环并发首次写入外层弱字典时，需要互斥保护
+_shared_async_transports_lock = threading.Lock()
+# 每个事件循环允许的最大共享 transport 桶数；超出后按 LRU 淘汰最久未用桶。
+_MAX_SHARED_TRANSPORTS_PER_LOOP = 32
+# 默认的最大 keep-alive 连接数
+_DEFAULT_MAX_KEEPALIVE_CONNECTIONS = 20
+# 默认的最大连接数（包括 keep-alive 和非 keep-alive 连接）
+_DEFAULT_MAX_CONNECTIONS = 40
+# 默认的 keep-alive 连接过期时间（秒）
+_DEFAULT_KEEPALIVE_EXPIRY = 30
+# 持有 LRU 淘汰后正在异步关闭的 transport task，避免 fire-and-forget 被 GC 警告
+_pending_eviction_tasks: set[asyncio.Task] = set()
+
+
+def _get_shared_async_transport(
+    proxy: Optional[str],
+    verify: Union[bool, str],
+    max_keepalive_connections: int,
+    max_connections: int,
+    keepalive_expiry: int,
+) -> Optional[httpx.AsyncHTTPTransport]:
+    """
+    返回与当前事件循环绑定的共享 AsyncHTTPTransport（底层连接池）；首次按需创建。
+    没有运行中的事件循环或循环已关闭时返回 None，由调用方走临时客户端兜底。
+
+    Transport 只持有连接池、SSL、代理；cookies/timeout/follow_redirects 等
+    会话级状态由调用方在外层 AsyncClient(transport=...) 实例化时单独配置，
+    每次调用用完即销毁，因此天然无 jar 累积串扰。
+    """
+    try:
+        loop = asyncio.get_running_loop()
+    except RuntimeError:
+        return None
+    if loop.is_closed():
+        return None
+
+    with _shared_async_transports_lock:
+        per_loop = _shared_async_transports.get(loop)
+        if per_loop is None:
+            per_loop = collections.OrderedDict()
+            _shared_async_transports[loop] = per_loop
+
+    key: _SharedTransportKey = (
+        proxy,
+        verify,
+        max_keepalive_connections,
+        max_connections,
+        keepalive_expiry,
+    )
+    transport = per_loop.get(key)
+    if transport is not None:
+        per_loop.move_to_end(key)  # LRU 触摸
+        return transport
+
+    # 首次见到这个配置，创建新的共享 transport 桶
+    transport = httpx.AsyncHTTPTransport(
+        proxy=proxy,
+        verify=verify,
+        limits=httpx.Limits(
+            max_keepalive_connections=max_keepalive_connections,
+            max_connections=max_connections,
+            keepalive_expiry=keepalive_expiry,
+        ),
+    )
+    per_loop[key] = transport
+
+    # LRU 淘汰：超出上限时关闭并移除最久未用桶
+    while len(per_loop) > _MAX_SHARED_TRANSPORTS_PER_LOOP:
+        evicted_key, evicted_transport = per_loop.popitem(last=False)
+        try:
+            task = loop.create_task(evicted_transport.aclose())
+            # 强引用避免 task 仅被 loop 弱持有而触发 "Task was destroyed but pending"
+            _pending_eviction_tasks.add(task)
+            task.add_done_callback(_pending_eviction_tasks.discard)
+        except Exception as e:  # pragma: no cover - 防御性
+            logger.debug(f"LRU 淘汰共享 transport 时调度关闭失败: {e!r}")
+
+    return transport
+
+
+async def aclose_shared_async_transports() -> None:
+    """
+    关闭当前事件循环下所有共享 AsyncHTTPTransport，释放底层连接池。
+    建议在应用关闭流程（如 FastAPI shutdown 事件）中调用，避免 ResourceWarning。
+    """
+    try:
+        loop = asyncio.get_running_loop()
+    except RuntimeError:
+        return
+    # 弹出而非 get+clear，避免外层 dict 残留空 OrderedDict 占位
+    with _shared_async_transports_lock:
+        per_loop = _shared_async_transports.pop(loop, None)
+    if not per_loop:
+        return
+    transports = list(per_loop.values())
+    per_loop.clear()
+    # 并行关闭：每个 transport 的 TLS close_notify 各占一个 RTT，
+    # 顺序等待会线性放大 shutdown 耗时；return_exceptions 让单点失败
+    # 不影响其他 transport 的释放
+    results = await asyncio.gather(
+        *(t.aclose() for t in transports), return_exceptions=True
+    )
+    for result in results:
+        if isinstance(result, BaseException):
+            logger.debug(f"关闭共享 AsyncHTTPTransport 失败: {result!r}")
+
+
 def _url_decode_if_latin(original: str) -> str:
    """
-    解码URL编码的字符串，只解码文本，二进程数据保持不变
+    解码URL编码的字符串，只解码文本，二进制数据保持不变
    :param original: URL编码字符串
    :return: 解码后的字符串或原始二进制数据
    """
    try:
        # 先解码
-        decoded = unquote(original, encoding='latin-1')
+        decoded = unquote(original, encoding="latin-1")
        # 再完整编码
-        fully_encoded = quote(decoded, safe='')
+        fully_encoded = quote(decoded, safe="")
        # 验证
-        decoded_again = unquote(fully_encoded, encoding='latin-1')
+        decoded_again = unquote(fully_encoded, encoding="latin-1")
        if decoded_again == decoded:
            return decoded
    except Exception as e:
        logger.error(f"latin-1解码URL编码失败：{e}")
    return original

+
 def cookie_parse(cookies_str: str, array: bool = False) -> Union[list, dict]:
    """
    解析cookie，转化为字典或者数组
@@ -101,7 +249,8 @@ class RequestUtils:
    HTTP请求工具类，提供同步HTTP请求的基本功能
    """

-    def __init__(self,
+    def __init__(
+        self,
        headers: dict = None,
        ua: str = None,
        cookies: Union[str, dict] = None,
@@ -110,7 +259,8 @@ class RequestUtils:
        timeout: int = None,
        referer: str = None,
        content_type: str = None,
-                 accept_type: str = None):
+        accept_type: str = None,
+    ):
        """
        :param headers: 请求头部信息
        :param ua: User-Agent字符串
@@ -138,7 +288,7 @@ class RequestUtils:
                "User-Agent": ua,
                "Content-Type": content_type,
                "Accept": accept_type,
-                "referer": referer
+                "referer": referer,
            }
        if cookies:
            if isinstance(cookies, str):
@@ -167,7 +317,9 @@ class RequestUtils:
                except Exception as e:
                    logger.debug(f"关闭响应失败: {e}")

-    def request(self, method: str, url: str, raise_exception: bool = False, **kwargs) -> Optional[Response]:
+    def request(
+        self, method: str, url: str, raise_exception: bool = False, **kwargs
+    ) -> Optional[Response]:
        """
        发起HTTP请求
        :param method: HTTP方法，如 get, post, put 等
@@ -191,7 +343,11 @@ class RequestUtils:
            return req_method(method, url, **kwargs)
        except requests.exceptions.RequestException as e:
            # 获取更详细的错误信息
-            error_msg = str(e) if str(e) else f"未知网络错误 (URL: {url}, Method: {method.upper()})"
+            error_msg = (
+                str(e)
+                if str(e)
+                else f"未知网络错误 (URL: {url}, Method: {method.upper()})"
+            )
            logger.debug(f"请求失败: {error_msg}")
            if raise_exception:
                raise
@@ -219,7 +375,9 @@ class RequestUtils:
            if response is not None:
                response.close()

-    def post(self, url: str, data: Any = None, json: dict = None, **kwargs) -> Optional[Response]:
+    def post(
+        self, url: str, data: Any = None, json: dict = None, **kwargs
+    ) -> Optional[Response]:
        """
        发送POST请求
        :param url: 请求的URL
@@ -240,14 +398,16 @@ class RequestUtils:
        """
        return self.request(method="put", url=url, data=data, **kwargs)

-    def get_res(self,
+    def get_res(
+        self,
        url: str,
        params: dict = None,
        data: Any = None,
        json: dict = None,
        allow_redirects: bool = True,
        raise_exception: bool = False,
-                **kwargs) -> Optional[Response]:
+        **kwargs,
+    ) -> Optional[Response]:
        """
        发送GET请求并返回响应对象
        :param url: 请求的URL
@@ -260,14 +420,16 @@ class RequestUtils:
        :return: HTTP响应对象，若发生RequestException则返回None
        :raises: requests.exceptions.RequestException 仅raise_exception为True时会抛出
        """
-        return self.request(method="get",
+        return self.request(
+            method="get",
            url=url,
            params=params,
            data=data,
            json=json,
            allow_redirects=allow_redirects,
            raise_exception=raise_exception,
-                            **kwargs)
+            **kwargs,
+        )

    @contextmanager
    def get_stream(self, url: str, params: dict = None, **kwargs):
@@ -277,7 +439,7 @@ class RequestUtils:
        :param params: 请求的参数
        :param kwargs: 其他请求参数
        """
-        kwargs['stream'] = True
+        kwargs["stream"] = True
        response = self.request(method="get", url=url, params=params, **kwargs)
        try:
            yield response
@@ -285,7 +447,8 @@ class RequestUtils:
            if response is not None:
                response.close()

-    def post_res(self,
+    def post_res(
+        self,
        url: str,
        data: Any = None,
        params: dict = None,
@@ -293,7 +456,8 @@ class RequestUtils:
        files: Any = None,
        json: dict = None,
        raise_exception: bool = False,
-                 **kwargs) -> Optional[Response]:
+        **kwargs,
+    ) -> Optional[Response]:
        """
        发送POST请求并返回响应对象
        :param url: 请求的URL
@@ -307,7 +471,8 @@ class RequestUtils:
        :return: HTTP响应对象，若发生RequestException则返回None
        :raises: requests.exceptions.RequestException 仅raise_exception为True时会抛出
        """
-        return self.request(method="post",
+        return self.request(
+            method="post",
            url=url,
            data=data,
            params=params,
@@ -315,9 +480,11 @@ class RequestUtils:
            files=files,
            json=json,
            raise_exception=raise_exception,
-                            **kwargs)
+            **kwargs,
+        )

-    def put_res(self,
+    def put_res(
+        self,
        url: str,
        data: Any = None,
        params: dict = None,
@@ -325,7 +492,8 @@ class RequestUtils:
        files: Any = None,
        json: dict = None,
        raise_exception: bool = False,
-                **kwargs) -> Optional[Response]:
+        **kwargs,
+    ) -> Optional[Response]:
        """
        发送PUT请求并返回响应对象
        :param url: 请求的URL
@@ -339,7 +507,8 @@ class RequestUtils:
        :return: HTTP响应对象，若发生RequestException则返回None
        :raises: requests.exceptions.RequestException 仅raise_exception为True时会抛出
        """
-        return self.request(method="put",
+        return self.request(
+            method="put",
            url=url,
            data=data,
            params=params,
@@ -347,15 +516,18 @@ class RequestUtils:
            files=files,
            json=json,
            raise_exception=raise_exception,
-                            **kwargs)
+            **kwargs,
+        )

-    def delete_res(self,
+    def delete_res(
+        self,
        url: str,
        data: Any = None,
        params: dict = None,
        allow_redirects: bool = True,
        raise_exception: bool = False,
-                   **kwargs) -> Optional[Response]:
+        **kwargs,
+    ) -> Optional[Response]:
        """
        发送DELETE请求并返回响应对象
        :param url: 请求的URL
@@ -367,13 +539,15 @@ class RequestUtils:
        :return: HTTP响应对象，若发生RequestException则返回None
        :raises: requests.exceptions.RequestException 仅raise_exception为True时会抛出
        """
-        return self.request(method="delete",
+        return self.request(
+            method="delete",
            url=url,
            data=data,
            params=params,
            allow_redirects=allow_redirects,
            raise_exception=raise_exception,
-                            **kwargs)
+            **kwargs,
+        )

    def get_json(self, url: str, params: dict = None, **kwargs) -> Optional[dict]:
        """
@@ -397,7 +571,9 @@ class RequestUtils:
            if response is not None:
                response.close()

-    def post_json(self, url: str, data: Any = None, json: dict = None, **kwargs) -> Optional[dict]:
+    def post_json(
+        self, url: str, data: Any = None, json: dict = None, **kwargs
+    ) -> Optional[dict]:
        """
        发送POST请求并返回JSON数据，自动关闭连接
        :param url: 请求的URL
@@ -441,15 +617,26 @@ class RequestUtils:
                try:
                    max_age = int(directive.split("=")[1])
                except Exception as e:
-                    logger.debug(f"Invalid max-age directive in Cache-Control header: {directive}, {e}")
-            elif directive in {"no-cache", "private", "public", "no-store", "must-revalidate"}:
+                    logger.debug(
+                        f"Invalid max-age directive in Cache-Control header: {directive}, {e}"
+                    )
+            elif directive in {
+                "no-cache",
+                "private",
+                "public",
+                "no-store",
+                "must-revalidate",
+            }:
                cache_directive = directive

        return cache_directive, max_age

    @staticmethod
-    def generate_cache_headers(etag: Optional[str], cache_control: Optional[str] = "public",
-                               max_age: Optional[int] = 86400) -> dict:
+    def generate_cache_headers(
+        etag: Optional[str],
+        cache_control: Optional[str] = "public",
+        max_age: Optional[int] = 86400,
+    ) -> dict:
        """
        生成 HTTP 响应的 ETag 和 Cache-Control 头
        :param etag: 响应的 ETag 值。如果为 None，则不添加 ETag 头部。
@@ -472,8 +659,11 @@ class RequestUtils:
        return cache_headers

    @staticmethod
-    def detect_encoding_from_html_response(response: Response,
-                                           performance_mode: bool = False, confidence_threshold: float = 0.8):
+    def detect_encoding_from_html_response(
+        response: Response,
+        performance_mode: bool = False,
+        confidence_threshold: float = 0.8,
+    ):
        """
        根据HTML响应内容探测编码信息

@@ -498,7 +688,9 @@ class RequestUtils:
                    return "utf-8"

                # 3. 如果是 HTML 响应体，检查其中的 <meta charset="..."> 标签
-                if re.search(r"charset=[\"']?utf-8[\"']?", response.text, re.IGNORECASE):
+                if re.search(
+                    r"charset=[\"']?utf-8[\"']?", response.text, re.IGNORECASE
+                ):
                    return "utf-8"

                # 4. 尝试从 response headers 中获取编码信息
@@ -517,7 +709,9 @@ class RequestUtils:
                    return "utf-8"

                # 3. 如果是 HTML 响应体，检查其中的 <meta charset="..."> 标签
-                if re.search(r"charset=[\"']?utf-8[\"']?", response.text, re.IGNORECASE):
+                if re.search(
+                    r"charset=[\"']?utf-8[\"']?", response.text, re.IGNORECASE
+                ):
                    return "utf-8"
                # 4. 使用 chardet 库进一步分析内容
                detection = chardet.detect(response.content)
@@ -533,8 +727,11 @@ class RequestUtils:
            return fallback_encoding or "utf-8"

    @staticmethod
-    def get_decoded_html_content(response: Response,
-                                 performance_mode: bool = False, confidence_threshold: float = 0.8) -> str:
+    def get_decoded_html_content(
+        response: Response,
+        performance_mode: bool = False,
+        confidence_threshold: float = 0.8,
+    ) -> str:
        """
        获取HTML响应的解码文本内容

@@ -548,9 +745,12 @@ class RequestUtils:
                return ""
            if response.content:
                # 1. 获取编码信息
-                encoding = (RequestUtils.detect_encoding_from_html_response(response, performance_mode,
-                                                                            confidence_threshold)
-                            or response.apparent_encoding)
+                encoding = (
+                    RequestUtils.detect_encoding_from_html_response(
+                        response, performance_mode, confidence_threshold
+                    )
+                    or response.apparent_encoding
+                )
                # 2. 根据解析得到的编码进行解码
                try:
                    # 尝试用推测的编码解码
@@ -572,7 +772,8 @@ class AsyncRequestUtils:
    异步HTTP请求工具类，提供异步HTTP请求的基本功能
    """

-    def __init__(self,
+    def __init__(
+        self,
        headers: dict = None,
        ua: str = None,
        cookies: Union[str, dict] = None,
@@ -582,8 +783,12 @@ class AsyncRequestUtils:
        referer: str = None,
        content_type: str = None,
        accept_type: str = None,
-                 verify: bool = False,
-                 follow_redirects: bool = True):
+        verify: Union[bool, str] = False,
+        follow_redirects: bool = True,
+        max_keepalive_connections: int = _DEFAULT_MAX_KEEPALIVE_CONNECTIONS,
+        max_connections: int = _DEFAULT_MAX_CONNECTIONS,
+        keepalive_expiry: int = _DEFAULT_KEEPALIVE_EXPIRY,
+    ):
        """
        :param headers: 请求头部信息
        :param ua: User-Agent字符串
@@ -596,12 +801,18 @@ class AsyncRequestUtils:
        :param accept_type: Accept头部信息，默认为 "application/json"
        :param verify: 是否校验证书
        :param follow_redirects: 客户端默认是否跟随重定向
+        :param max_keepalive_connections: 共享 AsyncHTTPTransport 的最大 keep-alive 连接数
+        :param max_connections: 共享 AsyncHTTPTransport 的最大连接数
+        :param keepalive_expiry: 共享 AsyncHTTPTransport 的 keep-alive 连接过期时间（秒）
        """
        self._proxies = self._convert_proxies_for_httpx(proxies)
        self._client = client
        self._timeout = timeout or 20
        self._verify = verify
        self._follow_redirects = follow_redirects
+        self._max_keepalive_connections = max_keepalive_connections
+        self._max_connections = max_connections
+        self._keepalive_expiry = keepalive_expiry
        if not content_type:
            content_type = "application/x-www-form-urlencoded; charset=UTF-8"
        if headers:
@@ -672,7 +883,9 @@ class AsyncRequestUtils:
                except Exception as e:
                    logger.debug(f"关闭异步响应失败: {e}")

-    async def request(self, method: str, url: str, raise_exception: bool = False, **kwargs) -> Optional[httpx.Response]:
+    async def request(
+        self, method: str, url: str, raise_exception: bool = False, **kwargs
+    ) -> Optional[httpx.Response]:
        """
        发起异步HTTP请求
        :param method: HTTP方法，如 get, post, put 等
@@ -682,33 +895,104 @@ class AsyncRequestUtils:
        :return: HTTP响应对象
        :raises: httpx.RequestError 仅raise_exception为True时会抛出
        """
-        if self._client is None:
-            # 创建临时客户端
+        # 运行时 self._cookies 只能是 dict | None（cookie_parse 默认 array=False 返回 dict）
+        cookies_dict: Optional[dict] = self._cookies if isinstance(self._cookies, dict) else None
+
+        if self._client is not None:
+            # 用户自管 client 时，把实例级 cookies 注入到本次 per-request kwargs，
+            # 既能复用用户 client，又不让 instance cookies 被静默丢弃。
+            # 调用方若显式传 kwargs["cookies"]，则以其为准（setdefault 不覆盖）。
+            if cookies_dict is not None:
+                kwargs.setdefault("cookies", cookies_dict)
+            return await self._make_request(
+                self._client, method, url, raise_exception, **kwargs
+            )
+
+        # 共享底层 transport（连接池+TLS 复用），每次请求创建轻量 AsyncClient。
+        # AsyncClient 持有的 cookie jar 仅存活于本次请求 lifecycle，
+        # 既复用握手又彻底避免 jar 跨调用累积。
+        transport = _get_shared_async_transport(
+            proxy=self._proxies,
+            verify=self._verify,
+            max_keepalive_connections=self._max_keepalive_connections,
+            max_connections=self._max_connections,
+            keepalive_expiry=self._keepalive_expiry,
+        )
+        if transport is not None:
+            # 用 _NonClosingTransportProxy 包装共享 transport，吞掉 AsyncClient.__aexit__
+            # 传播下来的 transport.__aexit__，避免每次 async with 退出都把共享连接池清空。
+            async with httpx.AsyncClient(
+                transport=_NonClosingTransportProxy(transport),
+                timeout=httpx.Timeout(self._timeout),
+                follow_redirects=self._follow_redirects,
+                cookies=cookies_dict,
+            ) as client:
+                return await self._make_request(
+                    client, method, url, raise_exception, **kwargs
+                )
+
+        # 兜底：没有运行中的事件循环时，临时客户端走完即关
        async with httpx.AsyncClient(
            proxy=self._proxies,
            timeout=self._timeout,
            verify=self._verify,
            follow_redirects=self._follow_redirects,
-                    cookies=self._cookies  # 在创建客户端时传入Cookie
+            cookies=cookies_dict,
        ) as client:
-                return await self._make_request(client, method, url, raise_exception, **kwargs)
-        else:
-            return await self._make_request(self._client, method, url, raise_exception, **kwargs)
+            return await self._make_request(
+                client, method, url, raise_exception, **kwargs
+            )

-    async def _make_request(self, client: httpx.AsyncClient, method: str, url: str, raise_exception: bool = False,
-                            **kwargs) -> Optional[httpx.Response]:
+    async def _make_request(
+        self,
+        client: httpx.AsyncClient,
+        method: str,
+        url: str,
+        raise_exception: bool = False,
+        **kwargs,
+    ) -> Optional[httpx.Response]:
        """
        执行实际的异步请求
        """
        kwargs.setdefault("headers", self._headers)
-        # Cookie已经在AsyncClient创建时设置，不要在request时再设置，否则会被覆盖
-        # kwargs.setdefault("cookies", self._cookies)
+        # 共享池下 client 自带默认 timeout，这里用每请求 timeout 覆盖以尊重实例配置
+        kwargs.setdefault("timeout", self._timeout)
+        # Cookie 在 request() 入口已按 path 处理：
+        # - path A（用户自管 client）：kwargs["cookies"] 已注入
+        # - path B/C（新建 AsyncClient）：构造时已绑定 cookies
+        # 这里不重复 setdefault，避免覆盖各 path 的设定
+
+        method_upper = method.upper()
+        # 仅对幂等方法做 stale-pool 竞态重试：复用了刚被对端 FIN 的 keep-alive 连接时，
+        # 实际请求通常未到服务端，httpx 自身不重试，这里兜底一次。
+        is_idempotent = method_upper in ("GET", "HEAD", "OPTIONS")
+        stale_conn_errs = (httpx.RemoteProtocolError, httpx.ReadError, httpx.WriteError)

        try:
            return await client.request(method, url, **kwargs)
+        except stale_conn_errs as e:
+            if is_idempotent:
+                logger.debug(f"keep-alive 连接已失效，幂等方法重试一次: {e!r}")
+                try:
+                    return await client.request(method, url, **kwargs)
+                except httpx.RequestError as e2:
+                    error_msg = (
+                        str(e2) or f"未知网络错误 (URL: {url}, Method: {method_upper})"
+                    )
+                    logger.debug(f"重试后异步请求仍失败: {error_msg}")
+                    if raise_exception:
+                        raise
+                    return None
+            # 非幂等方法（POST/PUT/PATCH/DELETE 等）不重试以避免重复副作用，
+            # 但仍记录调试日志，避免静默失败掩盖问题
+            error_msg = str(e) or f"未知网络错误 (URL: {url}, Method: {method_upper})"
+            logger.debug(f"异步请求失败(非幂等不重试): {error_msg}")
+            if raise_exception:
+                raise
+            return None
        except httpx.RequestError as e:
            # 获取更详细的错误信息
-            error_msg = str(e) if str(e) else f"未知网络错误 (URL: {url}, Method: {method.upper()})"
+            error_msg = str(e) or f"未知网络错误 (URL: {url}, Method: {method_upper})"
            logger.debug(f"异步请求失败: {error_msg}")
            if raise_exception:
                raise
@@ -736,7 +1020,9 @@ class AsyncRequestUtils:
            if response is not None:
                await response.aclose()

-    async def post(self, url: str, data: Any = None, json: dict = None, **kwargs) -> Optional[httpx.Response]:
+    async def post(
+        self, url: str, data: Any = None, json: dict = None, **kwargs
+    ) -> Optional[httpx.Response]:
        """
        发送异步POST请求
        :param url: 请求的URL
@@ -745,9 +1031,13 @@ class AsyncRequestUtils:
        :param kwargs: 其他请求参数，如headers, cookies, proxies等
        :return: HTTP响应对象，若发生RequestError则返回None
        """
-        return await self.request(method="post", url=url, data=data, json=json, **kwargs)
+        return await self.request(
+            method="post", url=url, data=data, json=json, **kwargs
+        )

-    async def put(self, url: str, data: Any = None, **kwargs) -> Optional[httpx.Response]:
+    async def put(
+        self, url: str, data: Any = None, **kwargs
+    ) -> Optional[httpx.Response]:
        """
        发送异步PUT请求
        :param url: 请求的URL
@@ -757,14 +1047,16 @@ class AsyncRequestUtils:
        """
        return await self.request(method="put", url=url, data=data, **kwargs)

-    async def get_res(self,
+    async def get_res(
+        self,
        url: str,
        params: dict = None,
        data: Any = None,
        json: dict = None,
        allow_redirects: bool = True,
        raise_exception: bool = False,
-                      **kwargs) -> Optional[httpx.Response]:
+        **kwargs,
+    ) -> Optional[httpx.Response]:
        """
        发送异步GET请求并返回响应对象
        :param url: 请求的URL
@@ -777,32 +1069,105 @@ class AsyncRequestUtils:
        :return: HTTP响应对象，若发生RequestError则返回None
        :raises: httpx.RequestError 仅raise_exception为True时会抛出
        """
-        return await self.request(method="get",
+        return await self.request(
+            method="get",
            url=url,
            params=params,
            data=data,
            json=json,
            follow_redirects=allow_redirects,
            raise_exception=raise_exception,
-                                  **kwargs)
+            **kwargs,
+        )

    @asynccontextmanager
-    async def get_stream(self, url: str, params: dict = None, **kwargs):
+    async def get_stream(
+        self,
+        url: str,
+        params: dict = None,
+        raise_exception: bool = False,
+        **kwargs,
+    ):
        """
-        获取异步流式响应的上下文管理器，适用于大文件下载
+        获取异步流式响应的上下文管理器，适用于大文件下载。
+        使用 httpx.AsyncClient.stream() 标准流式 API，避免把响应体一次性读入内存。
+
        :param url: 请求的URL
        :param params: 请求的参数
-        :param kwargs: 其他请求参数
+        :param raise_exception: 是否在发生异常时抛出，否则吞掉并 yield None
+        :param kwargs: 其他请求参数（headers, cookies 等）
+        :return: 上下文管理器，进入后 yield httpx.Response（出错时 yield None）
        """
-        kwargs['stream'] = True
-        response = await self.request(method="get", url=url, params=params, **kwargs)
-        try:
-            yield response
-        finally:
-            if response is not None:
-                await response.aclose()
+        cookies_dict: Optional[dict] = self._cookies if isinstance(self._cookies, dict) else None
+        kwargs.setdefault("headers", self._headers)

-    async def post_res(self,
+        # 与 _make_request 保持一致：复用 keep-alive 时偶遇对端 FIN 的连接，
+        # 流式 GET 是幂等的，单次重试即可
+        stale_conn_errs = (httpx.RemoteProtocolError, httpx.ReadError, httpx.WriteError)
+
+        async with AsyncExitStack() as stack:
+            # 选 client：复用与 request() 相同的三条 path 逻辑
+            if self._client is not None:
+                client = self._client
+                if cookies_dict is not None:
+                    kwargs.setdefault("cookies", cookies_dict)
+            else:
+                transport = _get_shared_async_transport(
+                    proxy=self._proxies,
+                    verify=self._verify,
+                    max_keepalive_connections=self._max_keepalive_connections,
+                    max_connections=self._max_connections,
+                    keepalive_expiry=self._keepalive_expiry,
+                )
+                if transport is not None:
+                    client = await stack.enter_async_context(
+                        httpx.AsyncClient(
+                            transport=_NonClosingTransportProxy(transport),
+                            timeout=httpx.Timeout(self._timeout),
+                            follow_redirects=self._follow_redirects,
+                            cookies=cookies_dict,
+                        )
+                    )
+                else:
+                    client = await stack.enter_async_context(
+                        httpx.AsyncClient(
+                            proxy=self._proxies,
+                            timeout=self._timeout,
+                            verify=self._verify,
+                            follow_redirects=self._follow_redirects,
+                            cookies=cookies_dict,
+                        )
+                    )
+
+            try:
+                response = await stack.enter_async_context(
+                    client.stream("GET", url, params=params, **kwargs)
+                )
+            except stale_conn_errs as e:
+                logger.debug(f"流式 keep-alive 连接已失效，重试一次: {e!r}")
+                try:
+                    response = await stack.enter_async_context(
+                        client.stream("GET", url, params=params, **kwargs)
+                    )
+                except httpx.RequestError as e2:
+                    logger.debug(f"重试后异步流式请求仍失败: {e2!r}")
+                    if raise_exception:
+                        raise
+                    yield None
+                    return
+            except httpx.RequestError as e:
+                logger.debug(f"异步流式请求失败: {e!r}")
+                if raise_exception:
+                    raise
+                yield None
+                return
+
+            # AsyncExitStack 反向 unwind：先关 stream，再关 owned client；
+            # yield 体内的异常由标准 async with 协议透传给各 __aexit__
+            yield response
+
+    async def post_res(
+        self,
        url: str,
        data: Any = None,
        params: dict = None,
@@ -810,7 +1175,8 @@ class AsyncRequestUtils:
        files: Any = None,
        json: dict = None,
        raise_exception: bool = False,
-                       **kwargs) -> Optional[httpx.Response]:
+        **kwargs,
+    ) -> Optional[httpx.Response]:
        """
        发送异步POST请求并返回响应对象
        :param url: 请求的URL
@@ -824,7 +1190,8 @@ class AsyncRequestUtils:
        :return: HTTP响应对象，若发生RequestError则返回None
        :raises: httpx.RequestError 仅raise_exception为True时会抛出
        """
-        return await self.request(method="post",
+        return await self.request(
+            method="post",
            url=url,
            data=data,
            params=params,
@@ -832,9 +1199,11 @@ class AsyncRequestUtils:
            files=files,
            json=json,
            raise_exception=raise_exception,
-                                  **kwargs)
+            **kwargs,
+        )

-    async def put_res(self,
+    async def put_res(
+        self,
        url: str,
        data: Any = None,
        params: dict = None,
@@ -842,7 +1211,8 @@ class AsyncRequestUtils:
        files: Any = None,
        json: dict = None,
        raise_exception: bool = False,
-                      **kwargs) -> Optional[httpx.Response]:
+        **kwargs,
+    ) -> Optional[httpx.Response]:
        """
        发送异步PUT请求并返回响应对象
        :param url: 请求的URL
@@ -856,7 +1226,8 @@ class AsyncRequestUtils:
        :return: HTTP响应对象，若发生RequestError则返回None
        :raises: httpx.RequestError 仅raise_exception为True时会抛出
        """
-        return await self.request(method="put",
+        return await self.request(
+            method="put",
            url=url,
            data=data,
            params=params,
@@ -864,15 +1235,18 @@ class AsyncRequestUtils:
            files=files,
            json=json,
            raise_exception=raise_exception,
-                                  **kwargs)
+            **kwargs,
+        )

-    async def delete_res(self,
+    async def delete_res(
+        self,
        url: str,
        data: Any = None,
        params: dict = None,
        allow_redirects: bool = True,
        raise_exception: bool = False,
-                         **kwargs) -> Optional[httpx.Response]:
+        **kwargs,
+    ) -> Optional[httpx.Response]:
        """
        发送异步DELETE请求并返回响应对象
        :param url: 请求的URL
@@ -884,13 +1258,15 @@ class AsyncRequestUtils:
        :return: HTTP响应对象，若发生RequestError则返回None
        :raises: httpx.RequestError 仅raise_exception为True时会抛出
        """
-        return await self.request(method="delete",
+        return await self.request(
+            method="delete",
            url=url,
            data=data,
            params=params,
            follow_redirects=allow_redirects,
            raise_exception=raise_exception,
-                                  **kwargs)
+            **kwargs,
+        )

    async def get_json(self, url: str, params: dict = None, **kwargs) -> Optional[dict]:
        """
@@ -914,7 +1290,9 @@ class AsyncRequestUtils:
            if response is not None:
                await response.aclose()

-    async def post_json(self, url: str, data: Any = None, json: dict = None, **kwargs) -> Optional[dict]:
+    async def post_json(
+        self, url: str, data: Any = None, json: dict = None, **kwargs
+    ) -> Optional[dict]:
        """
        发送异步POST请求并返回JSON数据，自动关闭连接
        :param url: 请求的URL
@@ -925,7 +1303,9 @@ class AsyncRequestUtils:
        """
        if json is None:
            json = {}
-        response = await self.request(method="post", url=url, data=data, json=json, **kwargs)
+        response = await self.request(
+            method="post", url=url, data=data, json=json, **kwargs
+        )
        try:
            if response:
                try: