refactor(proxy): 将 proxy resolver 同步阻塞操作异步化,避免阻塞事件循环

- 为 resolve_proxy_info、resolve_delegate_config、build_proxy_url、
  get_system_proxy_config、build_post_kwargs、build_stream_kwargs 新增
  _async 异步版本,通过 asyncio.to_thread 在工作线程中执行同步 DB 查询
- 为 _proxy_node_cache 和 _system_proxy_cache 添加 threading.Lock 保护
  多线程并发读写安全
- 大 payload 的 gzip 压缩超过 64KB 阈值时走线程池,小 payload 仍在事件
  循环中同步执行以避免不必要的线程调度开销
- hub_transport 的 frame 压缩同样增加异步版本
- 删除已无调用者的同步方法 create_client_with_proxy,将其逻辑内联至
  get_upstream_client 并改为异步
- 更新所有 handler/executor/failover 调用点使用新的异步 API
- 补充 async 版本的单元测试
This commit is contained in:
fawney19
2026-03-12 13:59:21 +08:00
parent 66fec80e79
commit 0112ab752b
11 changed files with 309 additions and 162 deletions

View File

@@ -930,11 +930,11 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
from src.services.proxy_node.resolver import (
get_proxy_label,
resolve_effective_proxy,
resolve_proxy_info,
resolve_proxy_info_async,
)
effective_proxy = resolve_effective_proxy(provider.proxy, getattr(key, "proxy", None))
ctx.proxy_info = resolve_proxy_info(effective_proxy)
ctx.proxy_info = await resolve_proxy_info_async(effective_proxy)
proxy_label = get_proxy_label(ctx.proxy_info)
logger.debug(
@@ -946,10 +946,13 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
# simulate streaming to the client (sync -> stream bridge).
if not upstream_is_stream:
from src.clients.http_client import HTTPClientPool
from src.services.proxy_node.resolver import build_post_kwargs, resolve_delegate_config
from src.services.proxy_node.resolver import (
build_post_kwargs_async,
resolve_delegate_config_async,
)
request_timeout_sync = provider.request_timeout or config.http_request_timeout
delegate_cfg = resolve_delegate_config(effective_proxy)
delegate_cfg = await resolve_delegate_config_async(effective_proxy)
http_client = await HTTPClientPool.get_upstream_client(
delegate_cfg,
proxy_config=effective_proxy,
@@ -957,7 +960,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
)
try:
_pkw = build_post_kwargs(
_pkw = await build_post_kwargs_async(
delegate_cfg,
url=url,
headers=provider_headers,
@@ -997,7 +1000,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
ctx.provider_request_headers = provider_headers
# retry once
_pkw = build_post_kwargs(
_pkw = await build_post_kwargs_async(
delegate_cfg,
url=url,
headers=provider_headers,
@@ -1147,9 +1150,12 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
# 获取 HTTP 客户端支持代理配置Key 级别优先于 Provider 级别)
# 使用连接池复用客户端,避免每次流式请求都新建 TCP/TLS 连接
from src.clients.http_client import HTTPClientPool
from src.services.proxy_node.resolver import build_stream_kwargs, resolve_delegate_config
from src.services.proxy_node.resolver import (
build_stream_kwargs_async,
resolve_delegate_config_async,
)
delegate_cfg = resolve_delegate_config(effective_proxy)
delegate_cfg = await resolve_delegate_config_async(effective_proxy)
http_client = await HTTPClientPool.get_upstream_client(
delegate_cfg,
proxy_config=effective_proxy,
@@ -1164,7 +1170,7 @@ class ChatHandlerBase(BaseMessageHandler, ABC):
async def _connect_and_prefetch() -> None:
"""建立连接并预读首字节(受整体超时控制)"""
nonlocal byte_iterator, prefetched_chunks, response_ctx
_skw = build_stream_kwargs(
_skw = await build_stream_kwargs_async(
delegate_cfg,
url=url,
headers=provider_headers,

View File

@@ -517,11 +517,11 @@ class ChatSyncExecutor:
from src.services.proxy_node.resolver import (
get_proxy_label,
resolve_effective_proxy,
resolve_proxy_info,
resolve_proxy_info_async,
)
_effective_proxy = resolve_effective_proxy(provider.proxy, getattr(key, "proxy", None))
ctx.sync_proxy_info = resolve_proxy_info(_effective_proxy)
ctx.sync_proxy_info = await resolve_proxy_info_async(_effective_proxy)
_proxy_label = get_proxy_label(ctx.sync_proxy_info)
provider_type = str(getattr(provider, "provider_type", "") or "").lower()
@@ -538,16 +538,16 @@ class ChatSyncExecutor:
from src.clients.http_client import HTTPClientPool
from src.config.settings import config
from src.services.proxy_node.resolver import (
build_post_kwargs,
build_stream_kwargs,
resolve_delegate_config,
build_post_kwargs_async,
build_stream_kwargs_async,
resolve_delegate_config_async,
)
# 非流式请求使用 http_request_timeout 作为整体超时
# 优先使用 Provider 配置,否则使用全局配置
request_timeout = provider.request_timeout or config.http_request_timeout
delegate_cfg = resolve_delegate_config(_effective_proxy)
delegate_cfg = await resolve_delegate_config_async(_effective_proxy)
http_client = await HTTPClientPool.get_upstream_client(
delegate_cfg,
proxy_config=_effective_proxy,
@@ -559,7 +559,7 @@ class ChatSyncExecutor:
resp: httpx.Response | None = None
if not upstream_is_stream:
try:
_pkw = build_post_kwargs(
_pkw = await build_post_kwargs_async(
delegate_cfg,
url=url,
headers=provider_hdrs,
@@ -584,7 +584,7 @@ class ChatSyncExecutor:
)
try:
_stream_args = build_stream_kwargs(
_stream_args = await build_stream_kwargs_async(
delegate_cfg,
url=url,
headers=provider_hdrs,

View File

@@ -455,19 +455,22 @@ class CliStreamMixin:
# 解析有效代理Key 级别优先于 Provider 级别)
from src.services.proxy_node.resolver import get_proxy_label as _gpl
from src.services.proxy_node.resolver import resolve_effective_proxy as _rep
from src.services.proxy_node.resolver import resolve_proxy_info as _rpi
from src.services.proxy_node.resolver import resolve_proxy_info_async as _rpi_async
effective_proxy = _rep(provider.proxy, getattr(key, "proxy", None))
ctx.proxy_info = _rpi(effective_proxy)
ctx.proxy_info = await _rpi_async(effective_proxy)
# If upstream is forced to non-stream mode, we execute a sync request and then
# simulate streaming to the client (sync -> stream bridge).
if not upstream_is_stream:
from src.clients.http_client import HTTPClientPool
from src.services.proxy_node.resolver import build_post_kwargs, resolve_delegate_config
from src.services.proxy_node.resolver import (
build_post_kwargs_async,
resolve_delegate_config_async,
)
request_timeout_sync = provider.request_timeout or config.http_request_timeout
delegate_cfg = resolve_delegate_config(effective_proxy)
delegate_cfg = await resolve_delegate_config_async(effective_proxy)
http_client = await HTTPClientPool.get_upstream_client(
delegate_cfg,
proxy_config=effective_proxy,
@@ -475,7 +478,7 @@ class CliStreamMixin:
)
try:
_pkw = build_post_kwargs(
_pkw = await build_post_kwargs_async(
delegate_cfg,
url=url,
headers=provider_headers,
@@ -517,7 +520,7 @@ class CliStreamMixin:
ctx.provider_request_headers = provider_headers
# retry once
_pkw = build_post_kwargs(
_pkw = await build_post_kwargs_async(
delegate_cfg,
url=url,
headers=provider_headers,
@@ -663,9 +666,12 @@ class CliStreamMixin:
# 获取 HTTP 客户端支持代理配置Key 级别优先于 Provider 级别)
# 使用连接池复用客户端,避免每次流式请求都新建 TCP/TLS 连接
from src.clients.http_client import HTTPClientPool
from src.services.proxy_node.resolver import build_stream_kwargs, resolve_delegate_config
from src.services.proxy_node.resolver import (
build_stream_kwargs_async,
resolve_delegate_config_async,
)
delegate_cfg = resolve_delegate_config(effective_proxy)
delegate_cfg = await resolve_delegate_config_async(effective_proxy)
http_client = await HTTPClientPool.get_upstream_client(
delegate_cfg,
proxy_config=effective_proxy,
@@ -680,7 +686,7 @@ class CliStreamMixin:
async def _connect_and_prefetch() -> None:
"""建立连接并预读首字节(受整体超时控制)"""
nonlocal byte_iterator, prefetched_chunks, response_ctx
_skw = build_stream_kwargs(
_skw = await build_stream_kwargs_async(
delegate_cfg,
url=url,
headers=provider_headers,

View File

@@ -272,11 +272,11 @@ class CliSyncMixin:
from src.services.proxy_node.resolver import (
get_proxy_label,
resolve_effective_proxy,
resolve_proxy_info,
resolve_proxy_info_async,
)
_effective_proxy = resolve_effective_proxy(provider.proxy, getattr(key, "proxy", None))
sync_proxy_info = resolve_proxy_info(_effective_proxy)
sync_proxy_info = await resolve_proxy_info_async(_effective_proxy)
_proxy_label = get_proxy_label(sync_proxy_info)
logger.info(
@@ -291,16 +291,16 @@ class CliSyncMixin:
# 注意:使用 get_proxy_client 复用连接池,不再每次创建新客户端
from src.clients.http_client import HTTPClientPool
from src.services.proxy_node.resolver import (
build_post_kwargs,
build_stream_kwargs,
resolve_delegate_config,
build_post_kwargs_async,
build_stream_kwargs_async,
resolve_delegate_config_async,
)
# 非流式请求使用 http_request_timeout 作为整体超时
# 优先使用 Provider 配置,否则使用全局配置
request_timeout = provider.request_timeout or config.http_request_timeout
delegate_cfg = resolve_delegate_config(_effective_proxy)
delegate_cfg = await resolve_delegate_config_async(_effective_proxy)
http_client = await HTTPClientPool.get_upstream_client(
delegate_cfg,
proxy_config=_effective_proxy,
@@ -312,7 +312,7 @@ class CliSyncMixin:
resp: httpx.Response | None = None
if not upstream_is_stream:
try:
_pkw = build_post_kwargs(
_pkw = await build_post_kwargs_async(
delegate_cfg,
url=url,
headers=provider_headers,
@@ -337,7 +337,7 @@ class CliSyncMixin:
)
try:
_stream_args = build_stream_kwargs(
_stream_args = await build_stream_kwargs_async(
delegate_cfg,
url=url,
headers=provider_headers,

View File

@@ -21,10 +21,11 @@ from src.config import config
from src.core.logger import logger
from src.services.provider.fingerprint import KNOWN_IMPERSONATE_PROFILES
from src.services.proxy_node.resolver import (
build_proxy_url,
build_proxy_url_async,
compute_proxy_cache_key,
get_system_proxy_config,
get_system_proxy_config_async,
make_proxy_param,
resolve_delegate_config_async,
)
from src.utils.ssl_utils import get_ssl_context, get_ssl_context_for_profile
@@ -222,12 +223,9 @@ class HTTPClientPool:
"""
# 无特定代理时,回退到系统默认代理
if not proxy_config:
proxy_config = get_system_proxy_config()
proxy_config = await get_system_proxy_config_async()
# tunnel 模式检查tunnel 节点走专用的 TunnelTransport 客户端
from src.services.proxy_node.resolver import resolve_delegate_config
delegate_cfg = resolve_delegate_config(proxy_config)
delegate_cfg = await resolve_delegate_config_async(proxy_config)
if delegate_cfg and delegate_cfg.get("tunnel"):
return await cls._get_tunnel_client(delegate_cfg["node_id"])
@@ -262,7 +260,7 @@ class HTTPClientPool:
await cls._evict_lru_proxy_client()
# 添加代理配置
proxy_url = build_proxy_url(proxy_config) if proxy_config else None
proxy_url = await build_proxy_url_async(proxy_config) if proxy_config else None
# curl_cffi Transport: real TLS fingerprint impersonation.
# Supports:
@@ -424,57 +422,6 @@ class HTTPClientPool:
finally:
await client.aclose()
@classmethod
def create_client_with_proxy(
cls,
proxy_config: dict[str, Any] | None = None,
timeout: httpx.Timeout | None = None,
tls_profile: str | None = None,
**kwargs: Any,
) -> httpx.AsyncClient:
"""
创建带代理配置的HTTP客户端
⚠️ 性能警告:此方法每次都创建新客户端,推荐使用 get_proxy_client() 复用连接。
Args:
proxy_config: 代理配置字典,包含 url, username, password
timeout: 超时配置
**kwargs: 其他 httpx.AsyncClient 配置参数
Returns:
配置好的 httpx.AsyncClient 实例(调用者需要负责关闭)
"""
client_config: dict[str, Any] = {
"http2": config.enable_http2,
"verify": get_ssl_context_for_profile(tls_profile),
"follow_redirects": True,
}
if timeout:
client_config["timeout"] = timeout
else:
client_config["timeout"] = httpx.Timeout(
connect=config.http_connect_timeout,
read=config.http_read_timeout,
write=config.http_write_timeout,
pool=config.http_pool_timeout,
)
# 无特定代理时,回退到系统默认代理(与 get_proxy_client 行为一致)
if proxy_config is None:
proxy_config = get_system_proxy_config()
# 添加代理配置
proxy_url = build_proxy_url(proxy_config) if proxy_config else None
proxy_param = make_proxy_param(proxy_url)
if proxy_param:
client_config["proxy"] = proxy_param
logger.debug("创建带代理的HTTP客户端(一次性): {}", proxy_config.get("url", "unknown"))
client_config.update(kwargs)
return httpx.AsyncClient(**client_config) # type: ignore[arg-type]
@classmethod
async def _reset_default_client(cls) -> bool:
"""Atomically replace the shared default client with a fresh instance.
@@ -551,7 +498,7 @@ class HTTPClientPool:
return True
if not proxy_config:
proxy_config = get_system_proxy_config()
proxy_config = await get_system_proxy_config_async()
base_cache_key = compute_proxy_cache_key(proxy_config)
if base_cache_key == "__no_proxy__":
@@ -616,11 +563,37 @@ class HTTPClientPool:
"""
if delegate_cfg and delegate_cfg.get("tunnel"):
return await cls._get_tunnel_client(delegate_cfg["node_id"], timeout=timeout)
return cls.create_client_with_proxy(
proxy_config=proxy_config,
timeout=timeout,
tls_profile=tls_profile,
client_config: dict[str, Any] = {
"http2": config.enable_http2,
"verify": get_ssl_context_for_profile(tls_profile),
"follow_redirects": True,
}
if timeout:
client_config["timeout"] = timeout
else:
client_config["timeout"] = httpx.Timeout(
connect=config.http_connect_timeout,
read=config.http_read_timeout,
write=config.http_write_timeout,
pool=config.http_pool_timeout,
)
resolved_proxy_config = proxy_config
if resolved_proxy_config is None:
resolved_proxy_config = await get_system_proxy_config_async()
proxy_url = (
await build_proxy_url_async(resolved_proxy_config) if resolved_proxy_config else None
)
proxy_param = make_proxy_param(proxy_url)
if proxy_param:
client_config["proxy"] = proxy_param
logger.debug(
"创建带代理的HTTP客户端(一次性): {}",
resolved_proxy_config.get("url", "unknown") if resolved_proxy_config else "unknown",
)
return httpx.AsyncClient(**client_config)
@classmethod
async def _get_tunnel_client(

View File

@@ -153,7 +153,7 @@ class FailoverEngine:
async def _rotate_upstream_client(self, candidate: ProviderCandidate) -> bool:
from src.clients.http_client import HTTPClientPool
from src.services.proxy_node.resolver import (
resolve_delegate_config,
resolve_delegate_config_async,
resolve_effective_proxy,
)
@@ -161,7 +161,7 @@ class FailoverEngine:
getattr(candidate.provider, "proxy", None),
getattr(candidate.key, "proxy", None),
)
delegate_cfg = resolve_delegate_config(effective_proxy)
delegate_cfg = await resolve_delegate_config_async(effective_proxy)
return await HTTPClientPool.reset_upstream_client(
delegate_cfg, proxy_config=effective_proxy
)

View File

@@ -28,6 +28,7 @@ if TYPE_CHECKING:
_TUNNEL_COMPRESS_MIN_SIZE = 512
_TUNNEL_ASYNC_COMPRESS_THRESHOLD = 64 * 1024
_RECONNECT_DELAYS_SECONDS: tuple[float, ...] = (0.0, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0)
_HEARTBEAT_DEDUP_TTL_SECONDS = 600
_LOOP_WATCHDOG_INTERVAL_SECONDS = 1.0
@@ -619,7 +620,7 @@ class HubConnectionManager:
body_data = body or b""
if body_data:
body_payload, body_flags = _compress_frame_payload(body_data)
body_payload, body_flags = await _compress_frame_payload_async(body_data)
else:
body_payload, body_flags = body_data, 0
body_flags |= FrameFlags.END_STREAM
@@ -783,6 +784,12 @@ def _compress_frame_payload(data: bytes) -> tuple[bytes, int]:
return data, 0
async def _compress_frame_payload_async(data: bytes) -> tuple[bytes, int]:
if len(data) < _TUNNEL_ASYNC_COMPRESS_THRESHOLD:
return _compress_frame_payload(data)
return await asyncio.to_thread(_compress_frame_payload, data)
def _decompress_frame_payload(frame: Frame) -> bytes:
if frame.is_gzip:
return gzip.decompress(frame.payload)

View File

@@ -7,9 +7,11 @@
from __future__ import annotations
import asyncio
import gzip
import hashlib
import json
import threading
import time
from typing import Any
from urllib.parse import quote, urlparse
@@ -24,18 +26,22 @@ from src.core.logger import logger
# ProxyNode 信息缓存(降低高频 DB 查询开销)
# ---------------------------------------------------------------------------
_proxy_node_cache: dict[str, tuple[dict[str, Any] | None, float]] = {}
_proxy_node_cache_lock = threading.Lock()
_PROXY_NODE_CACHE_TTL_SECONDS = 15.0
_PROXY_NODE_CACHE_NEGATIVE_TTL_SECONDS = 5.0 # 不可用节点使用更短的 TTL加速恢复感知
_PROXY_NODE_CACHE_MAX_SIZE = 256
# payload 超过此阈值时 build_*_kwargs_async 才走 to_thread
# 避免小 payload 承担不必要的线程调度开销
_ASYNC_PAYLOAD_THRESHOLD = 64 * 1024
def _get_proxy_node_info(node_id: str) -> dict[str, Any] | None:
"""
读取 ProxyNode 信息(带内存 TTL 缓存)
NOTE: 使用同步 DB sessioncreate_session在 async 上下文中会短暂阻塞
事件循环。60s TTL 缓存覆盖绝大多数请求,阻塞仅发生在缓存未命中时
若后续 delegate 模式导致调用频率显著上升,应考虑改为 run_in_executor 包装。
通过 asyncio.to_thread 在工作线程中运行,使用 _proxy_node_cache_lock 保护
缓存的并发读写安全
Returns:
aether-proxy 节点: {"ip": str, "port": int, "name": str, ...}
@@ -43,20 +49,16 @@ def _get_proxy_node_info(node_id: str) -> dict[str, Any] | None:
不存在/非在线: None
"""
now = time.time()
cached = _proxy_node_cache.get(node_id)
if cached:
value, expires_at = cached
if now < expires_at:
return value
# 防止无效 node_id 导致缓存无限膨胀:淘汰最旧的条目而非全部清除
if len(_proxy_node_cache) >= _PROXY_NODE_CACHE_MAX_SIZE:
# 按过期时间排序,删除最旧的 25%
evict_count = _PROXY_NODE_CACHE_MAX_SIZE // 4
sorted_keys = sorted(_proxy_node_cache, key=lambda k: _proxy_node_cache[k][1])
for k in sorted_keys[:evict_count]:
del _proxy_node_cache[k]
# 快速路径:缓存命中
with _proxy_node_cache_lock:
cached = _proxy_node_cache.get(node_id)
if cached:
value, expires_at = cached
if now < expires_at:
return value
# 缓存未命中,查询 DB
from src.database import create_session
from src.models.database import ProxyNode, ProxyNodeStatus
@@ -64,71 +66,77 @@ def _get_proxy_node_info(node_id: str) -> dict[str, Any] | None:
try:
node = db.query(ProxyNode).filter(ProxyNode.id == node_id).first()
if not node:
_proxy_node_cache[node_id] = (None, now + _PROXY_NODE_CACHE_NEGATIVE_TTL_SECONDS)
return None
# tunnel 模式节点:统一以 DB 的 tunnel_connected/status 为准(由 Hub 广播维护)
if node.tunnel_mode and not node.is_manual:
result: dict[str, Any] | None = None
ttl = _PROXY_NODE_CACHE_NEGATIVE_TTL_SECONDS
elif node.tunnel_mode and not node.is_manual:
if node.status != ProxyNodeStatus.ONLINE or not bool(node.tunnel_connected):
_proxy_node_cache[node_id] = (
None,
now + _PROXY_NODE_CACHE_NEGATIVE_TTL_SECONDS,
)
return None
value: dict[str, Any] = {
"name": node.name,
"ip": node.ip,
"port": node.port,
"tunnel_mode": True,
"tunnel_connected": True,
}
_proxy_node_cache[node_id] = (value, now + _PROXY_NODE_CACHE_TTL_SECONDS)
return value
# 手动节点 / 非 tunnel 节点:仍依赖 DB status
if node.status != ProxyNodeStatus.ONLINE:
_proxy_node_cache[node_id] = (None, now + _PROXY_NODE_CACHE_NEGATIVE_TTL_SECONDS)
return None
if node.is_manual:
value = {
result = None
ttl = _PROXY_NODE_CACHE_NEGATIVE_TTL_SECONDS
else:
result = {
"name": node.name,
"ip": node.ip,
"port": node.port,
"tunnel_mode": True,
"tunnel_connected": True,
}
ttl = _PROXY_NODE_CACHE_TTL_SECONDS
elif node.status != ProxyNodeStatus.ONLINE:
result = None
ttl = _PROXY_NODE_CACHE_NEGATIVE_TTL_SECONDS
elif node.is_manual:
result = {
"is_manual": True,
"name": node.name,
"proxy_url": node.proxy_url,
"username": node.proxy_username,
"password": node.proxy_password,
}
ttl = _PROXY_NODE_CACHE_TTL_SECONDS
else:
value = {
result = {
"name": node.name,
"ip": node.ip,
"port": node.port,
"tunnel_mode": bool(node.tunnel_mode),
"tunnel_connected": bool(node.tunnel_connected),
}
_proxy_node_cache[node_id] = (value, now + _PROXY_NODE_CACHE_TTL_SECONDS)
return value
ttl = _PROXY_NODE_CACHE_TTL_SECONDS
finally:
db.close()
# 写回缓存(持锁做淘汰+写入,使用新时间戳以排除 DB 查询耗时)
write_now = time.time()
with _proxy_node_cache_lock:
if len(_proxy_node_cache) >= _PROXY_NODE_CACHE_MAX_SIZE:
evict_count = _PROXY_NODE_CACHE_MAX_SIZE // 4
sorted_keys = sorted(_proxy_node_cache, key=lambda k: _proxy_node_cache[k][1])
for k in sorted_keys[:evict_count]:
del _proxy_node_cache[k]
_proxy_node_cache[node_id] = (result, write_now + ttl)
return result
# ---------------------------------------------------------------------------
# 系统默认代理
# ---------------------------------------------------------------------------
_system_proxy_cache: tuple[dict[str, Any] | None, float] | None = None
_system_proxy_cache_lock = threading.Lock()
_SYSTEM_PROXY_CACHE_TTL = 60.0
def invalidate_proxy_node_cache(node_id: str) -> None:
"""主动清除指定节点的信息缓存tunnel 断开时调用,避免使用过期的连接状态)"""
_proxy_node_cache.pop(node_id, None)
with _proxy_node_cache_lock:
_proxy_node_cache.pop(node_id, None)
def invalidate_system_proxy_cache() -> None:
"""手动失效系统代理缓存(在删除节点等操作后调用)"""
global _system_proxy_cache
_system_proxy_cache = None
with _system_proxy_cache_lock:
_system_proxy_cache = None
def get_system_proxy_config() -> dict[str, Any] | None:
@@ -140,10 +148,13 @@ def get_system_proxy_config() -> dict[str, Any] | None:
"""
global _system_proxy_cache
now = time.time()
if _system_proxy_cache:
value, expires_at = _system_proxy_cache
if now < expires_at:
return value
# 快速路径:缓存命中
with _system_proxy_cache_lock:
if _system_proxy_cache:
value, expires_at = _system_proxy_cache
if now < expires_at:
return value
from src.database import create_session
from src.services.system.config import SystemConfigService
@@ -155,15 +166,21 @@ def get_system_proxy_config() -> dict[str, Any] | None:
result: dict[str, Any] | None = {"node_id": node_id.strip(), "enabled": True}
else:
result = None
_system_proxy_cache = (result, now + _SYSTEM_PROXY_CACHE_TTL)
return result
except Exception as exc:
logger.warning("获取系统默认代理配置失败: {}", exc)
_system_proxy_cache = (None, now + _SYSTEM_PROXY_CACHE_TTL)
return None
result = None
finally:
db.close()
with _system_proxy_cache_lock:
_system_proxy_cache = (result, now + _SYSTEM_PROXY_CACHE_TTL)
return result
async def get_system_proxy_config_async() -> dict[str, Any] | None:
"""异步读取系统默认代理配置,避免在事件循环中执行同步 DB 查询。"""
return await asyncio.to_thread(get_system_proxy_config)
# ---------------------------------------------------------------------------
# 代理 URL 认证注入
@@ -472,6 +489,13 @@ def build_proxy_url(proxy_config: dict[str, Any]) -> str | None:
return proxy_url
async def build_proxy_url_async(proxy_config: dict[str, Any] | None) -> str | None:
"""异步构建代理 URL避免 ProxyNode 查询阻塞事件循环。"""
if not proxy_config:
return None
return await asyncio.to_thread(build_proxy_url, proxy_config)
# ---------------------------------------------------------------------------
# 代理信息追踪(日志/usage
# ---------------------------------------------------------------------------
@@ -528,6 +552,11 @@ def resolve_proxy_info(proxy_config: dict[str, Any] | None) -> dict[str, Any] |
return None
async def resolve_proxy_info_async(proxy_config: dict[str, Any] | None) -> dict[str, Any] | None:
"""异步解析代理摘要信息,避免事件循环被同步代理解析阻塞。"""
return await asyncio.to_thread(resolve_proxy_info, proxy_config)
def get_proxy_label(proxy_info: dict[str, Any] | None) -> str:
"""从 proxy_info 中提取简短的代理标签(用于日志)"""
if not proxy_info:
@@ -612,6 +641,13 @@ def resolve_delegate_config(proxy_config: dict[str, Any] | None) -> dict[str, An
return None
async def resolve_delegate_config_async(
proxy_config: dict[str, Any] | None,
) -> dict[str, Any] | None:
"""异步解析 tunnel 代理配置,避免同步 DB 查询阻塞事件循环。"""
return await asyncio.to_thread(resolve_delegate_config, proxy_config)
# ---------------------------------------------------------------------------
# 统一上游请求参数构建
# ---------------------------------------------------------------------------
@@ -642,6 +678,25 @@ def _maybe_compress_payload(
return json_bytes, normalized_headers
async def _maybe_compress_payload_async(
payload: Any,
headers: dict[str, str],
client_content_encoding: str | None = None,
) -> tuple[bytes, dict[str, str]]:
"""异步版 _maybe_compress_payload仅在 payload 超过阈值时走线程池。"""
json_bytes = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
normalized_headers = {k: v for k, v in headers.items() if k.lower() != "content-encoding"}
if is_gzip_content_encoding(client_content_encoding):
if len(json_bytes) >= _ASYNC_PAYLOAD_THRESHOLD:
compressed = await asyncio.to_thread(gzip.compress, json_bytes, 6)
else:
compressed = gzip.compress(json_bytes, compresslevel=6)
return compressed, {**normalized_headers, "Content-Encoding": "gzip"}
return json_bytes, normalized_headers
def build_post_kwargs(
_delegate_cfg: dict[str, Any] | None = None,
*,
@@ -673,6 +728,28 @@ def build_post_kwargs(
}
async def build_post_kwargs_async(
_delegate_cfg: dict[str, Any] | None = None,
*,
url: str,
headers: dict[str, str],
payload: Any,
timeout: float,
client_content_encoding: str | None = None,
refresh_auth: bool = False,
) -> dict[str, Any]:
"""构建 POST kwargs大 payload 的序列化/压缩走线程池避免阻塞事件循环。"""
content, final_headers = await _maybe_compress_payload_async(
payload, headers, client_content_encoding
)
return {
"url": url,
"content": content,
"headers": final_headers,
"timeout": httpx.Timeout(timeout),
}
def build_stream_kwargs(
_delegate_cfg: dict[str, Any] | None = None,
*,
@@ -704,3 +781,27 @@ def build_stream_kwargs(
if timeout is not None:
kwargs["timeout"] = httpx.Timeout(timeout)
return kwargs
async def build_stream_kwargs_async(
_delegate_cfg: dict[str, Any] | None = None,
*,
url: str,
headers: dict[str, str],
payload: Any,
timeout: float | None = None,
client_content_encoding: str | None = None,
) -> dict[str, Any]:
"""构建 stream kwargs大 payload 的序列化/压缩走线程池避免阻塞事件循环。"""
content, final_headers = await _maybe_compress_payload_async(
payload, headers, client_content_encoding
)
kwargs: dict[str, Any] = {
"method": "POST",
"url": url,
"content": content,
"headers": final_headers,
}
if timeout is not None:
kwargs["timeout"] = httpx.Timeout(timeout)
return kwargs

View File

@@ -203,7 +203,7 @@ class RequestExecutor:
# 非流式请求:标记为 success 状态
from src.services.proxy_node.resolver import (
resolve_effective_proxy,
resolve_proxy_info,
resolve_proxy_info_async,
)
_eff_proxy = resolve_effective_proxy(
@@ -214,7 +214,7 @@ class RequestExecutor:
"model_name": model_name,
"api_format": api_format,
}
_pi = resolve_proxy_info(_eff_proxy)
_pi = await resolve_proxy_info_async(_eff_proxy)
if _pi:
_extra["proxy"] = _pi
RequestCandidateService.mark_candidate_success(

View File

@@ -199,7 +199,10 @@ class TaskErrorOperationsService:
- "raise": raise the underlying exception
"""
from src.core.api_format.conversion.exceptions import FormatConversionError
from src.services.proxy_node.resolver import resolve_effective_proxy, resolve_proxy_info
from src.services.proxy_node.resolver import (
resolve_effective_proxy,
resolve_proxy_info_async,
)
from src.services.request.executor import ExecutionError
# 提前解析代理信息,写入候选记录的 extra_data用于链路追踪展示
@@ -207,7 +210,7 @@ class TaskErrorOperationsService:
getattr(candidate.provider, "proxy", None),
getattr(candidate.key, "proxy", None),
)
_proxy_info = resolve_proxy_info(_eff_proxy)
_proxy_info = await resolve_proxy_info_async(_eff_proxy)
_proxy_extra: dict[str, Any] | None = {"proxy": _proxy_info} if _proxy_info else None
if not isinstance(exec_err, ExecutionError):