diff --git a/app/utils/db_utils.py b/app/utils/db_utils.py index 3ad777d..ba50d00 100644 --- a/app/utils/db_utils.py +++ b/app/utils/db_utils.py @@ -1,36 +1,36 @@ -import asyncio -import sys -from contextlib import asynccontextmanager -from typing import Annotated, AsyncContextManager +from typing import Annotated from fastapi.params import Depends -from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver -from sqlalchemy import AsyncAdaptedQueuePool, text -from sqlalchemy.ext.asyncio import AsyncEngine +from sqlalchemy import text +from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker -from sqlmodel.ext.asyncio.session import AsyncSession from app.config.env import env -from sqlmodel import create_engine DATABASE_URL = f"mysql+asyncmy://{env.db_username}:{env.db_password}@{env.db_host}:{env.db_port}/{env.db_database}?charset=utf8mb4" -# 创建异步引擎实例，用于异步操作数据库 -async_engine = AsyncEngine(create_engine( +# 建议直接使用这个函数，不要嵌套 create_engine(AsyncAdaptedQueuePool) +async_engine = create_async_engine( DATABASE_URL, - poolclass=AsyncAdaptedQueuePool, # 使用异步适配的队列池 - pool_size=5, # 连接池保持的连接数 - max_overflow=10, # 允许超过pool_size的最大连接数 - pool_timeout=30, # 获取连接的超时时间(秒) - pool_recycle=180, # 连接回收时间(秒) - pool_pre_ping=True, # 启用连接有效性检测 - echo=True, # 启用SQL语句日志输出，便于开发调试 - future=True, # 启用SQLAlchemy 2.0风格的未来模式API -)) + # 连接池保持的连接数 + pool_size=5, + # 允许超过pool_size的最大连接数 + max_overflow=10, + # 获取连接的超时时间(秒) + pool_timeout=30, + # 连接回收时间(秒) + pool_recycle=60, # 针对 Docker NAT 环境优化 + # 启用连接有效性检测 + pool_pre_ping=True, # 解决 Lost connection 的核心 + # 启用SQL语句日志输出，便于开发调试 + echo=True, + # 启用SQLAlchemy 2.0风格的未来模式API + future=True, +) -# 创建一个会话工厂函数 +# 使用这个 Session 工厂 async_session = sessionmaker( - bind=async_engine, + async_engine, class_=AsyncSession, expire_on_commit=False ) diff --git a/app/utils/milvus_utils.py b/app/utils/milvus_utils.py index 7727aae..74549b9 100644 --- a/app/utils/milvus_utils.py +++ b/app/utils/milvus_utils.py @@ -63,7 +63,7 @@ class MilvusService: # 检查Milvus连接是否正常 async def check_milvus_connection(self): - await asyncio.sleep(0.1) + await asyncio.sleep(0.5) await self.async_search(KnowledgeQueryParam(question="hello", kb_code="")) print("✅ Milvus connection successful：", f"Milvus://{env.milvus_username}:{env.milvus_password}@{env.milvus_uri}/{env.llama_index_database}/{env.llama_index_collection}") diff --git a/app/utils/redis_utils.py b/app/utils/redis_utils.py index 05727c5..5f20334 100644 --- a/app/utils/redis_utils.py +++ b/app/utils/redis_utils.py @@ -1,63 +1,81 @@ import json import uuid +import logging from contextlib import asynccontextmanager -from typing import Annotated, AsyncContextManager +from typing import Annotated, AsyncContextManager, Any, Callable import redis.asyncio as redis from fastapi import Depends from app.config.env import env +# 设置日志，方便调试 +logger = logging.getLogger(__name__) -class RedisUtils(): + +class RedisUtils: def __init__(self): - self.redis_pool = None + self.redis_pool: redis.ConnectionPool | None = None - # 检查 Redis 连接，先创建连接池管理对象 + # 1. 优化连接池初始化：加入心跳和保活机制 async def check_redis_connection(self): + """初始化连接池并执行读写测试""" self.redis_pool = redis.ConnectionPool( host=env.redis_host, port=env.redis_port, password=env.redis_password, db=env.redis_db, encoding="utf-8", - decode_responses=True + decode_responses=True, + # --- 解决 10054 报错的核心配置 --- + health_check_interval=5, # 每5秒自动发送PING，保持连接活跃 + socket_keepalive=True, # 启用TCP层面的保活探测 + socket_connect_timeout=3, # 连接超时设置 + retry_on_timeout=True, # 遇到超时自动重试一次 ) + # 立即对redis做一次读写测试 async with self.get_redis_connection() as redis_client: try: await redis_client.ping() random_id = str(uuid.uuid4()) - await redis_client.set("__init__setup__", random_id) - cache_random_id = await redis_client.get("__init__setup__") + await redis_client.set("__init__setup__", random_id, ex=10) # 设置10秒过期 + cache_random_id = await redis_client.get("__init__setup__") + if cache_random_id != random_id: - raise Exception("Redis 连接失败，读写测试不一致") + raise Exception("Redis 读写测试一致性校验失败") + print("✅ Redis connection successful：", f"redis://{env.redis_password}@${env.redis_host}:{env.redis_port}/{env.redis_db}") except Exception as e: - print(f"❌ Database connection failed") + logger.error(f"❌ Redis connection failed: {e}") raise e - # 获取一个Redis连接客户端 + # 2. 获取连接：利用连接池自动管理，无需手动关闭 client @asynccontextmanager async def get_redis_connection(self) -> AsyncContextManager[redis.Redis]: - if not self.redis_pool: - raise Exception("Redis 连接池未初始化") + if self.redis_pool is None: + await self.check_redis_connection() + + # 异步连接池中，Redis 实例创建很轻量 + # 上下文管理器结束后，它会自动将连接归还给 pool，而不是物理断开 redis_client = redis.Redis(connection_pool=self.redis_pool) try: yield redis_client - finally: - await redis_client.close() + except Exception as e: + logger.error(f"Redis Connection Error: {e}") + raise - # 清理Redis连接池 + # 3. 清理连接池 async def close_redis_connection(self): if self.redis_pool: await self.redis_pool.disconnect() + print("🚀 Redis connection pool closed.") redis_utils = RedisUtils() -# 作用：用于在接口中注入得到会话实例对象session，在接口执行完毕之后，自动执行close动作关闭会话 +# --- FastAPI 依赖注入 --- async def get_redis_client() -> redis.Redis: async with redis_utils.get_redis_connection() as redis_client: yield redis_client @@ -66,8 +84,12 @@ async def get_redis_client() -> redis.Redis: RedisClientDep = Annotated[redis.Redis, Depends(get_redis_client)] -# 从redis中获取key的缓存，如果没有值则执行默认值获取函数，并保存到redis中 async def get_redis_cache(key: str, default_value_getter): + """ + 从redis中获取key的缓存，如果没有值则执行默认值获取函数，并保存到redis中 + @param key 缓存的key + @param default_value_getter 如果缓存值不存在的情况下则调用异步函数 default_value_getter 来获取默认值，这个函数的返回值必须是一个字典 dict + """ async with redis_utils.get_redis_connection() as redis_client: # default_value_getter返回的字段可能嵌套多层对象，这里改成用json字符串缓存 json_string = await redis_client.get(key)