|
@@ -1,15 +1,28 @@
|
|
|
"""
|
|
"""
|
|
|
数据库连接管理模块
|
|
数据库连接管理模块
|
|
|
"""
|
|
"""
|
|
|
|
|
+import asyncio
|
|
|
import aiomysql
|
|
import aiomysql
|
|
|
import redis.asyncio as aioredis
|
|
import redis.asyncio as aioredis
|
|
|
-from typing import Optional
|
|
|
|
|
|
|
+from typing import Optional, Callable, Awaitable
|
|
|
from contextlib import asynccontextmanager
|
|
from contextlib import asynccontextmanager
|
|
|
from config import settings
|
|
from config import settings
|
|
|
import logging
|
|
import logging
|
|
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
|
|
|
|
+MYSQL_RETRY_EXCEPTIONS = (
|
|
|
|
|
+ aiomysql.OperationalError,
|
|
|
|
|
+ aiomysql.InterfaceError,
|
|
|
|
|
+ aiomysql.InternalError,
|
|
|
|
|
+ ConnectionError,
|
|
|
|
|
+ ConnectionResetError,
|
|
|
|
|
+ ConnectionAbortedError,
|
|
|
|
|
+ BrokenPipeError,
|
|
|
|
|
+ TimeoutError,
|
|
|
|
|
+ asyncio.TimeoutError,
|
|
|
|
|
+)
|
|
|
|
|
+
|
|
|
|
|
|
|
|
class DatabaseManager:
|
|
class DatabaseManager:
|
|
|
"""数据库连接管理器(MySQL + Redis)"""
|
|
"""数据库连接管理器(MySQL + Redis)"""
|
|
@@ -17,11 +30,26 @@ class DatabaseManager:
|
|
|
def __init__(self):
|
|
def __init__(self):
|
|
|
self.mysql_pool: Optional[aiomysql.Pool] = None
|
|
self.mysql_pool: Optional[aiomysql.Pool] = None
|
|
|
self.redis_client: Optional[aioredis.Redis] = None
|
|
self.redis_client: Optional[aioredis.Redis] = None
|
|
|
|
|
+ self._mysql_lock = asyncio.Lock()
|
|
|
|
|
+ self._redis_lock = asyncio.Lock()
|
|
|
|
|
+ self._last_mysql_healthcheck: float = 0.0
|
|
|
|
|
|
|
|
async def connect(self):
|
|
async def connect(self):
|
|
|
"""初始化数据库连接"""
|
|
"""初始化数据库连接"""
|
|
|
try:
|
|
try:
|
|
|
- # 连接MySQL
|
|
|
|
|
|
|
+ await self._create_mysql_pool()
|
|
|
|
|
+ await self._create_redis_client()
|
|
|
|
|
+ except Exception as e:
|
|
|
|
|
+ logger.error(f"❌ 数据库连接失败: {e}")
|
|
|
|
|
+ raise
|
|
|
|
|
+
|
|
|
|
|
+ async def _create_mysql_pool(self):
|
|
|
|
|
+ async with self._mysql_lock:
|
|
|
|
|
+ if self.mysql_pool:
|
|
|
|
|
+ self.mysql_pool.close()
|
|
|
|
|
+ await self.mysql_pool.wait_closed()
|
|
|
|
|
+ self.mysql_pool = None
|
|
|
|
|
+
|
|
|
self.mysql_pool = await aiomysql.create_pool(
|
|
self.mysql_pool = await aiomysql.create_pool(
|
|
|
host=settings.MYSQL_HOST,
|
|
host=settings.MYSQL_HOST,
|
|
|
port=settings.MYSQL_PORT,
|
|
port=settings.MYSQL_PORT,
|
|
@@ -32,33 +60,43 @@ class DatabaseManager:
|
|
|
minsize=5,
|
|
minsize=5,
|
|
|
maxsize=settings.MYSQL_POOL_SIZE,
|
|
maxsize=settings.MYSQL_POOL_SIZE,
|
|
|
autocommit=True,
|
|
autocommit=True,
|
|
|
- echo=False
|
|
|
|
|
|
|
+ echo=False,
|
|
|
|
|
+ connect_timeout=settings.MYSQL_CONNECT_TIMEOUT,
|
|
|
|
|
+ # 注意:aiomysql 不支持 read_timeout 参数,已移除
|
|
|
|
|
+ pool_recycle=settings.MYSQL_POOL_RECYCLE,
|
|
|
|
|
+ )
|
|
|
|
|
+ self._last_mysql_healthcheck = 0.0
|
|
|
|
|
+ logger.info(
|
|
|
|
|
+ f"✅ MySQL连接池已创建: {settings.MYSQL_HOST}:{settings.MYSQL_PORT}/{settings.MYSQL_DATABASE}"
|
|
|
)
|
|
)
|
|
|
- logger.info(f"✅ MySQL连接池已创建: {settings.MYSQL_HOST}:{settings.MYSQL_PORT}/{settings.MYSQL_DATABASE}")
|
|
|
|
|
|
|
+
|
|
|
|
|
+ async def _create_redis_client(self):
|
|
|
|
|
+ async with self._redis_lock:
|
|
|
|
|
+ if self.redis_client:
|
|
|
|
|
+ await self.redis_client.close()
|
|
|
|
|
+ self.redis_client = None
|
|
|
|
|
|
|
|
- # 连接Redis
|
|
|
|
|
- # 根据SSL配置选择协议(rediss:// 或 redis://)
|
|
|
|
|
redis_protocol = "rediss" if settings.REDIS_SSL else "redis"
|
|
redis_protocol = "rediss" if settings.REDIS_SSL else "redis"
|
|
|
- redis_url = f"{redis_protocol}://:{settings.REDIS_PASSWORD}@{settings.REDIS_HOST}:{settings.REDIS_PORT}/{settings.REDIS_DB}"
|
|
|
|
|
|
|
+ redis_url = (
|
|
|
|
|
+ f"{redis_protocol}://:{settings.REDIS_PASSWORD}@"
|
|
|
|
|
+ f"{settings.REDIS_HOST}:{settings.REDIS_PORT}/{settings.REDIS_DB}"
|
|
|
|
|
+ )
|
|
|
|
|
|
|
|
self.redis_client = await aioredis.from_url(
|
|
self.redis_client = await aioredis.from_url(
|
|
|
redis_url,
|
|
redis_url,
|
|
|
encoding="utf-8",
|
|
encoding="utf-8",
|
|
|
decode_responses=True,
|
|
decode_responses=True,
|
|
|
max_connections=settings.REDIS_MAX_CONNECTIONS,
|
|
max_connections=settings.REDIS_MAX_CONNECTIONS,
|
|
|
- socket_connect_timeout=10, # 增加连接超时到10秒
|
|
|
|
|
- socket_timeout=10, # 增加操作超时到10秒
|
|
|
|
|
- retry_on_timeout=True, # 超时自动重试
|
|
|
|
|
- health_check_interval=30 # 健康检查间隔30秒
|
|
|
|
|
|
|
+ socket_connect_timeout=10,
|
|
|
|
|
+ socket_timeout=30, # 🔥 增加超时时间,避免短时网络波动断开连接
|
|
|
|
|
+ retry_on_timeout=True,
|
|
|
|
|
+ health_check_interval=30,
|
|
|
|
|
+ socket_keepalive=True, # 🔥 启用 TCP keepalive,保持长连接(Redis 3.2+ 默认启用)
|
|
|
)
|
|
)
|
|
|
-
|
|
|
|
|
- # 测试Redis连接
|
|
|
|
|
await self.redis_client.ping()
|
|
await self.redis_client.ping()
|
|
|
- logger.info(f"✅ Redis连接成功: {settings.REDIS_HOST}:{settings.REDIS_PORT}/{settings.REDIS_DB}")
|
|
|
|
|
-
|
|
|
|
|
- except Exception as e:
|
|
|
|
|
- logger.error(f"❌ 数据库连接失败: {e}")
|
|
|
|
|
- raise
|
|
|
|
|
|
|
+ logger.info(
|
|
|
|
|
+ f"✅ Redis连接成功: {settings.REDIS_HOST}:{settings.REDIS_PORT}/{settings.REDIS_DB}"
|
|
|
|
|
+ )
|
|
|
|
|
|
|
|
async def disconnect(self):
|
|
async def disconnect(self):
|
|
|
"""关闭数据库连接"""
|
|
"""关闭数据库连接"""
|
|
@@ -66,18 +104,58 @@ class DatabaseManager:
|
|
|
if self.mysql_pool:
|
|
if self.mysql_pool:
|
|
|
self.mysql_pool.close()
|
|
self.mysql_pool.close()
|
|
|
await self.mysql_pool.wait_closed()
|
|
await self.mysql_pool.wait_closed()
|
|
|
|
|
+ self.mysql_pool = None
|
|
|
logger.info("✅ MySQL连接池已关闭")
|
|
logger.info("✅ MySQL连接池已关闭")
|
|
|
|
|
|
|
|
if self.redis_client:
|
|
if self.redis_client:
|
|
|
await self.redis_client.close()
|
|
await self.redis_client.close()
|
|
|
|
|
+ self.redis_client = None
|
|
|
logger.info("✅ Redis连接已关闭")
|
|
logger.info("✅ Redis连接已关闭")
|
|
|
|
|
|
|
|
except Exception as e:
|
|
except Exception as e:
|
|
|
logger.error(f"❌ 关闭数据库连接失败: {e}")
|
|
logger.error(f"❌ 关闭数据库连接失败: {e}")
|
|
|
|
|
|
|
|
|
|
+ async def _ensure_mysql_pool(self):
|
|
|
|
|
+ """确保MySQL连接池可用(定期健康检查,保持长连接)"""
|
|
|
|
|
+ if not self.mysql_pool:
|
|
|
|
|
+ await self._create_mysql_pool()
|
|
|
|
|
+ return
|
|
|
|
|
+
|
|
|
|
|
+ loop = asyncio.get_running_loop()
|
|
|
|
|
+ now = loop.time()
|
|
|
|
|
+ # 🔥 健康检查间隔:只在超过间隔时检查,避免频繁检查
|
|
|
|
|
+ if now - self._last_mysql_healthcheck < settings.MYSQL_HEALTHCHECK_INTERVAL:
|
|
|
|
|
+ return
|
|
|
|
|
+
|
|
|
|
|
+ # 🔥 只做健康检查,不重建连接(保持长连接)
|
|
|
|
|
+ try:
|
|
|
|
|
+ async with self.mysql_pool.acquire() as conn:
|
|
|
|
|
+ await conn.ping()
|
|
|
|
|
+ self._last_mysql_healthcheck = now
|
|
|
|
|
+ # 健康检查成功,连接正常,不重建
|
|
|
|
|
+ except Exception as exc:
|
|
|
|
|
+ # 🔥 只有在健康检查失败时才重建连接(连接已断开)
|
|
|
|
|
+ logger.warning(f"⚠️ MySQL健康检查失败,连接可能已断开,准备重建: {exc}")
|
|
|
|
|
+ await self._create_mysql_pool()
|
|
|
|
|
+ self._last_mysql_healthcheck = loop.time()
|
|
|
|
|
+
|
|
|
|
|
+ async def _run_with_retry(self, operation: Callable[[], Awaitable], description: str):
|
|
|
|
|
+ """执行MySQL操作并在连接异常时重试一次"""
|
|
|
|
|
+ for attempt in range(2):
|
|
|
|
|
+ try:
|
|
|
|
|
+ await self._ensure_mysql_pool()
|
|
|
|
|
+ return await operation()
|
|
|
|
|
+ except MYSQL_RETRY_EXCEPTIONS as exc:
|
|
|
|
|
+ if attempt == 1:
|
|
|
|
|
+ logger.error(f"❌ {description} 失败,已重试: {exc}")
|
|
|
|
|
+ raise
|
|
|
|
|
+ logger.warning(f"⚠️ {description} 连接异常,尝试重建连接: {exc}")
|
|
|
|
|
+ await self._create_mysql_pool()
|
|
|
|
|
+
|
|
|
@asynccontextmanager
|
|
@asynccontextmanager
|
|
|
async def get_mysql_connection(self):
|
|
async def get_mysql_connection(self):
|
|
|
"""获取MySQL连接(上下文管理器)"""
|
|
"""获取MySQL连接(上下文管理器)"""
|
|
|
|
|
+ await self._ensure_mysql_pool()
|
|
|
if not self.mysql_pool:
|
|
if not self.mysql_pool:
|
|
|
raise RuntimeError("MySQL连接池未初始化")
|
|
raise RuntimeError("MySQL连接池未初始化")
|
|
|
|
|
|
|
@@ -87,21 +165,36 @@ class DatabaseManager:
|
|
|
|
|
|
|
|
async def execute_query(self, query: str, params: dict = None):
|
|
async def execute_query(self, query: str, params: dict = None):
|
|
|
"""执行查询(SELECT)"""
|
|
"""执行查询(SELECT)"""
|
|
|
- async with self.get_mysql_connection() as cursor:
|
|
|
|
|
- await cursor.execute(query, params or {})
|
|
|
|
|
- return await cursor.fetchall()
|
|
|
|
|
|
|
+ params = params or {}
|
|
|
|
|
+
|
|
|
|
|
+ async def _operation():
|
|
|
|
|
+ async with self.get_mysql_connection() as cursor:
|
|
|
|
|
+ await cursor.execute(query, params)
|
|
|
|
|
+ return await cursor.fetchall()
|
|
|
|
|
+
|
|
|
|
|
+ return await self._run_with_retry(_operation, "执行查询")
|
|
|
|
|
|
|
|
async def execute_one(self, query: str, params: dict = None):
|
|
async def execute_one(self, query: str, params: dict = None):
|
|
|
"""执行查询并返回一条记录"""
|
|
"""执行查询并返回一条记录"""
|
|
|
- async with self.get_mysql_connection() as cursor:
|
|
|
|
|
- await cursor.execute(query, params or {})
|
|
|
|
|
- return await cursor.fetchone()
|
|
|
|
|
|
|
+ params = params or {}
|
|
|
|
|
+
|
|
|
|
|
+ async def _operation():
|
|
|
|
|
+ async with self.get_mysql_connection() as cursor:
|
|
|
|
|
+ await cursor.execute(query, params)
|
|
|
|
|
+ return await cursor.fetchone()
|
|
|
|
|
+
|
|
|
|
|
+ return await self._run_with_retry(_operation, "执行单条查询")
|
|
|
|
|
|
|
|
async def execute_update(self, query: str, params: dict = None):
|
|
async def execute_update(self, query: str, params: dict = None):
|
|
|
"""执行更新(INSERT/UPDATE/DELETE)"""
|
|
"""执行更新(INSERT/UPDATE/DELETE)"""
|
|
|
- async with self.get_mysql_connection() as cursor:
|
|
|
|
|
- await cursor.execute(query, params or {})
|
|
|
|
|
- return cursor.lastrowid
|
|
|
|
|
|
|
+ params = params or {}
|
|
|
|
|
+
|
|
|
|
|
+ async def _operation():
|
|
|
|
|
+ async with self.get_mysql_connection() as cursor:
|
|
|
|
|
+ await cursor.execute(query, params)
|
|
|
|
|
+ return cursor.lastrowid
|
|
|
|
|
+
|
|
|
|
|
+ return await self._run_with_retry(_operation, "执行更新")
|
|
|
|
|
|
|
|
def get_redis(self) -> aioredis.Redis:
|
|
def get_redis(self) -> aioredis.Redis:
|
|
|
"""获取Redis客户端"""
|
|
"""获取Redis客户端"""
|