|
@@ -166,8 +166,15 @@ class DoubaoConnectionPool:
|
|
|
await self.idle_connections.put(connection)
|
|
await self.idle_connections.put(connection)
|
|
|
logger.debug(f"连接已归还,当前空闲连接: {self.idle_connections.qsize()}")
|
|
logger.debug(f"连接已归还,当前空闲连接: {self.idle_connections.qsize()}")
|
|
|
else:
|
|
else:
|
|
|
- # 连接不健康,重新创建
|
|
|
|
|
- logger.warning("连接不健康,重新创建...")
|
|
|
|
|
|
|
+ # 连接不健康,先关闭旧连接
|
|
|
|
|
+ reason = "致命错误" if hasattr(connection, 'has_fatal_error') and connection.has_fatal_error else "健康检查失败"
|
|
|
|
|
+ logger.warning(f"连接不健康 ({reason}),销毁并重新创建...")
|
|
|
|
|
+ try:
|
|
|
|
|
+ await connection.close()
|
|
|
|
|
+ except:
|
|
|
|
|
+ pass
|
|
|
|
|
+
|
|
|
|
|
+ # 重新创建连接
|
|
|
new_conn = await self._create_connection(-1)
|
|
new_conn = await self._create_connection(-1)
|
|
|
if new_conn:
|
|
if new_conn:
|
|
|
await self.idle_connections.put(new_conn)
|
|
await self.idle_connections.put(new_conn)
|
|
@@ -180,6 +187,11 @@ class DoubaoConnectionPool:
|
|
|
async def _is_connection_healthy(self, connection) -> bool:
|
|
async def _is_connection_healthy(self, connection) -> bool:
|
|
|
"""检查连接健康状态"""
|
|
"""检查连接健康状态"""
|
|
|
try:
|
|
try:
|
|
|
|
|
+ # 检查是否有致命错误
|
|
|
|
|
+ if hasattr(connection, 'has_fatal_error') and connection.has_fatal_error:
|
|
|
|
|
+ logger.warning(f"连接有致命错误: {connection.session_id}")
|
|
|
|
|
+ return False
|
|
|
|
|
+
|
|
|
# 检查连接是否还活着
|
|
# 检查连接是否还活着
|
|
|
if not connection or not connection.is_connected():
|
|
if not connection or not connection.is_connected():
|
|
|
return False
|
|
return False
|
|
@@ -214,6 +226,43 @@ class DoubaoConnectionPool:
|
|
|
f"总计: {idle_count + active_count}/{self.pool_size}"
|
|
f"总计: {idle_count + active_count}/{self.pool_size}"
|
|
|
)
|
|
)
|
|
|
|
|
|
|
|
|
|
+ # 🔧 主动清理空闲连接池中的失效连接
|
|
|
|
|
+ unhealthy_count = 0
|
|
|
|
|
+ temp_connections = []
|
|
|
|
|
+
|
|
|
|
|
+ # 取出所有空闲连接检查
|
|
|
|
|
+ while not self.idle_connections.empty():
|
|
|
|
|
+ try:
|
|
|
|
|
+ conn = await asyncio.wait_for(self.idle_connections.get(), timeout=1)
|
|
|
|
|
+ temp_connections.append(conn)
|
|
|
|
|
+ except asyncio.TimeoutError:
|
|
|
|
|
+ break
|
|
|
|
|
+
|
|
|
|
|
+ # 检查并重新放回健康的连接
|
|
|
|
|
+ for conn in temp_connections:
|
|
|
|
|
+ if await self._is_connection_healthy(conn):
|
|
|
|
|
+ await self.idle_connections.put(conn)
|
|
|
|
|
+ else:
|
|
|
|
|
+ unhealthy_count += 1
|
|
|
|
|
+ logger.warning(f"清理不健康的空闲连接: {conn.session_id}")
|
|
|
|
|
+ try:
|
|
|
|
|
+ await conn.close()
|
|
|
|
|
+ except:
|
|
|
|
|
+ pass
|
|
|
|
|
+
|
|
|
|
|
+ # 创建新连接补充
|
|
|
|
|
+ new_conn = await self._create_connection(-1)
|
|
|
|
|
+ if new_conn:
|
|
|
|
|
+ await self.idle_connections.put(new_conn)
|
|
|
|
|
+ self.connection_health[id(new_conn)] = {
|
|
|
|
|
+ 'created_at': time.time(),
|
|
|
|
|
+ 'last_used': time.time(),
|
|
|
|
|
+ 'request_count': 0
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ if unhealthy_count > 0:
|
|
|
|
|
+ logger.info(f"🔧 清理了 {unhealthy_count} 个不健康的连接并重新创建")
|
|
|
|
|
+
|
|
|
except Exception as e:
|
|
except Exception as e:
|
|
|
logger.error(f"健康检查循环错误: {e}")
|
|
logger.error(f"健康检查循环错误: {e}")
|
|
|
|
|
|