Skip to content

U
ubains-module-test
提交 d0757333 authored 作者: 陈泽健's avatar 陈泽健
浏览文件
操作

fix(device-sim): 修复批量启动SQLAlchemy连接池跨事件循环绑定导致大面积失败 

根因: 全局 async engine 的 AsyncAdaptedQueuePool 内部 asyncio.Queue 在启动时
绑定主事件循环, worker 线程新循环 await 该池报 " Queue is bound to a different
上级 a2c721e2
显示空白字符变更
内嵌 并排
正在显示 包含 105 行增加 和 114 行删除
backend/app/services/device_sim_service.py
浏览文件 @ d0757333
......@@ -11,6 +11,7 @@
import json
import logging
import queue as thread_queue
import threading
import asyncio
from datetime import datetime, timedelta
......@@ -49,16 +50,17 @@ class QueueMessage:
timestamp: datetime
# 全局消息队列（最大 10000 条）
_message_queue: asyncio.Queue = None
# 全局消息队列（最大 10000 条，线程安全）
_message_queue: thread_queue.Queue = None
_queue_stats = {"enqueued": 0, "processed": 0, "dropped": 0}
_queue_stats_lock = threading.Lock()
def get_message_queue() -> asyncio.Queue:
"""获取消息队列实例"""
def get_message_queue() -> thread_queue.Queue:
"""获取消息队列实例（线程安全）"""
global _message_queue
if _message_queue is None:
_message_queue = asyncio.Queue(maxsize=10000)
_message_queue = thread_queue.Queue(maxsize=10000)
return _message_queue
......@@ -72,9 +74,11 @@ def enqueue_message(msg: QueueMessage) -> bool:
queue = get_message_queue()
try:
queue.put_nowait(msg)
with _queue_stats_lock:
_queue_stats["enqueued"] += 1
return True
except asyncio.QueueFull:
except thread_queue.Full:
with _queue_stats_lock:
_queue_stats["dropped"] += 1
logger.warning(f"消息队列已满，丢弃消息: device_id={msg.device_id}")
return False
......@@ -1067,7 +1071,8 @@ class DeviceSimService:
dict: {"success_count": int, "failed_count": int, "failed_ids": list, "message": str}
"""
from concurrent.futures import ThreadPoolExecutor
from app.database import async_session_maker
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from app.config import settings
if not device_ids:
return {
......@@ -1080,33 +1085,89 @@ class DeviceSimService:
# 去重，避免同一设备重复启动
unique_ids = list(dict.fromkeys(device_ids))
# 关键：不能复用全局 async_session_maker/engine！
# 全局 engine 的连接池内部是 asyncio.Queue（AsyncAdaptedQueuePool），
# 在应用启动时（主事件循环 init_db）被首次使用后即绑定到主循环。
# worker 线程如果用自己的新事件循环去 await 这个池子，会报
# "Queue is bound to a different event loop"，导致批量启动大面积失败。
# 因此每个 worker 线程必须使用独立 engine + 独立事件循环，
# 线程内复用（ThreadPoolExecutor 会复用线程），批结束时统一释放。
_thread_db = threading.local() # 每个线程的 (loop, engine, maker)
_created_db_ctxs = [] # 记录本次批量启动创建的所有上下文
_db_ctx_lock = threading.Lock()
def _get_thread_db_ctx():
"""获取当前线程独立的事件循环 + 数据库引擎（首次创建，之后复用）"""
ctx = getattr(_thread_db, "ctx", None)
if ctx is None:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
engine = create_async_engine(
settings.DATABASE_URL,
future=True,
connect_args={"check_same_thread": False} if "sqlite" in settings.DATABASE_URL else {},
pool_pre_ping="sqlite" in settings.DATABASE_URL,
pool_recycle=3600,
# 每个线程串行使用其引擎，pool_size=1 足够，避免连接数爆炸
pool_size=1,
max_overflow=0,
)
maker = async_sessionmaker(
engine, class_=AsyncSession, expire_on_commit=False, autoflush=False
)
ctx = {"loop": loop, "engine": engine, "maker": maker}
_thread_db.ctx = ctx
with _db_ctx_lock:
_created_db_ctxs.append(ctx)
else:
# 线程被复用：重新把该线程所属循环设为当前循环
asyncio.set_event_loop(ctx["loop"])
return ctx
def _release_thread_db_ctxs():
"""批量启动结束后，在各自循环上释放引擎并关闭循环（线程池线程已退出）"""
with _db_ctx_lock:
ctxs = _created_db_ctxs[:]
_created_db_ctxs.clear()
for ctx in ctxs:
loop, engine = ctx["loop"], ctx["engine"]
try:
if not loop.is_closed():
loop.run_until_complete(engine.dispose())
except Exception as e:
logger.warning(f"释放批量启动数据库引擎失败: {e}")
finally:
try:
loop.close()
except Exception:
pass
def _start_one_sync(device_id: str) -> bool:
"""
在独立线程中启动单个设备（同步包装器）
每个线程创建独立的 asyncio 事件循环和数据库会话，
实现真正的并行执行，避免同步 HTTP 调用阻塞事件循环。
每个线程持有一个独立的事件循环和数据库引擎（见 _get_thread_db_ctx），
整个线程生命周期内复用，规避全局连接池跨事件循环的绑定问题。
"""
new_loop = asyncio.new_event_loop()
asyncio.set_event_loop(new_loop)
ctx = _get_thread_db_ctx()
loop = ctx["loop"]
try:
async def _run():
async with async_session_maker() as session:
async with ctx["maker"]() as session:
svc = DeviceSimService(session)
result = await svc.start_simulator(device_id)
await session.commit()
return result
return new_loop.run_until_complete(_run())
return loop.run_until_complete(_run())
except Exception as e:
logger.warning(f"批量启动设备失败: {device_id}, error={e}")
return False
finally:
new_loop.close()
loop = asyncio.get_event_loop()
semaphore = asyncio.Semaphore(max_concurrency)
# 使用线程池 + 信号量控制并发
try:
with ThreadPoolExecutor(max_workers=max_concurrency, thread_name_prefix="sim_start") as executor:
async def _start_one(device_id: str) -> Tuple[str, bool]:
async with semaphore:
......@@ -1114,6 +1175,9 @@ class DeviceSimService:
return device_id, result
results = await asyncio.gather(*[_start_one(did) for did in unique_ids])
finally:
# 无论成败都释放线程独立引擎，避免连接泄漏
_release_thread_db_ctxs()
success_count = 0
failed_ids = []
......@@ -1349,7 +1413,9 @@ async def message_consumer_loop():
"""
消息消费循环（在主线程异步循环中运行）
从消息队列中取出消息，写入数据库并触发 WebSocket 推送。
从线程安全的消息队列中取出消息，写入数据库并触发 WebSocket 推送。
使用 poll 模式（get_nowait + sleep）替代阻塞 await queue.get()，
因为队列是 thread_queue.Queue，不支持 asyncio 的 await。
"""
from app.database import async_session_maker
......@@ -1357,9 +1423,13 @@ async def message_consumer_loop():
while True:
try:
# 从队列获取消息（阻塞等待）
# 从队列获取消息（非阻塞 poll）
queue = get_message_queue()
msg: QueueMessage = await queue.get()
try:
msg: QueueMessage = queue.get_nowait()
except thread_queue.Empty:
await asyncio.sleep(0.05)
continue
# 创建数据库 session
async with async_session_maker() as db:
......@@ -1388,6 +1458,7 @@ async def message_consumer_loop():
msg.device_type = sim.device_type
await db.commit()
with _queue_stats_lock:
_queue_stats["processed"] += 1
except Exception as e:
......@@ -1419,6 +1490,7 @@ async def message_consumer_loop():
def get_queue_stats() -> dict:
"""获取队列统计信息"""
queue = get_message_queue()
with _queue_stats_lock:
return {
"queue_size": queue.qsize(),
"enqueued": _queue_stats["enqueued"],
......
deploy_to_560.py deleted 100644 → 0
浏览文件 @ a2c721e2
#!/usr/bin/env python3
"""
部署脚本 - 更新服务器 192.168.5.60 的前端和后端代码
"""
import paramiko
import sys
import time
def deploy(restart_backend=True, restart_frontend=True):
host = "192.168.5.60"
username = "ubains"
password = "Ubains@123"
deploy_dir = "/data/third_party/plat-auto-test"
commands = [
# 拉取最新代码
f"cd {deploy_dir} && git pull origin platform-auto-test",
]
if restart_backend:
commands.extend([
# 重启后端容器（应用 Python 代码变更）
"docker compose -f /data/third_party/plat-auto-test/deploy/docker-compose.yml restart app",
"sleep 5", # 等待容器启动
"docker compose -f /data/third_party/plat-auto-test/deploy/docker-compose.yml ps app",
])
if restart_frontend:
commands.extend([
# 重启前端容器（如果前端有变更）
"docker compose -f /data/third_party/plat-auto-test/deploy/docker-compose.yml restart nginx",
])
print(f"连接服务器 {host}...")
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
try:
ssh.connect(host, username=username, password=password, timeout=30)
print("连接成功")
for cmd in commands:
print(f"\n执行: {cmd}")
stdin, stdout, stderr = ssh.exec_command(cmd, timeout=120)
out = stdout.read().decode('utf-8')
err = stderr.read().decode('utf-8')
if out:
print(out)
if err and "warning" not in err.lower():
print(f"[stderr] {err}")
print("\n" + "=" * 50)
print("部署完成!")
print("=" * 50)
print("访问地址:")
print(" 前端: http://192.168.5.60")
print(" API 文档: http://192.168.5.60/docs")
print(" 健康检查: http://192.168.5.60/health")
except Exception as e:
print(f"错误: {e}")
sys.exit(1)
finally:
ssh.close()
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description="部署到 192.168.5.60")
parser.add_argument("--backend-only", action="store_true", help="只重启后端")
parser.add_argument("--frontend-only", action="store_true", help="只重启前端")
parser.add_argument("--no-restart", action="store_true", help="只拉取代码，不重启容器")
args = parser.parse_args()
if args.no_restart:
deploy(restart_backend=False, restart_frontend=False)
elif args.backend_only:
deploy(restart_backend=True, restart_frontend=False)
elif args.frontend_only:
deploy(restart_backend=False, restart_frontend=True)
else:
deploy()
\ No newline at end of file
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论