feat: 优化连接相关配置
This commit is contained in:
martsforever
+22
-22
@@ -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
|
||||
)
|
||||
|
||||
@@ -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}")
|
||||
|
||||
|
||||
+38
-16
@@ -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)
|
||||
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)
|
||||
|
||||
Reference in New Issue
Block a user