fix(device-sim): 消息记录架构重构 - 消息队列解耦模拟器线程与数据库

- 引入 asyncio.Queue 消息队列

- 模拟器回调只入队，不阻塞

- 消费循环写入数据库 + WebSocket 推送
Co-Authored-By: Claude <noreply@anthropic.com>

backend/app/main.py

@@ -70,11 +70,29 @@ async def lifespan(app: FastAPI):
     except Exception as e:
         logger.warning(f"默认登录模板初始化失败（非致命）: {e}")
 
+    # 启动消息消费循环
+    consumer_task = None
+    try:
+        from app.services.device_sim_service import message_consumer_loop
+        consumer_task = asyncio.create_task(message_consumer_loop())
+        logger.info("消息消费循环已启动")
+    except Exception as e:
+        logger.warning(f"消息消费循环启动失败: {e}")
+
     yield
 
     # 关闭时
     logger.info("应用正在关闭...")
 
+    # 停止消息消费循环
+    if consumer_task:
+        consumer_task.cancel()
+        try:
+            await consumer_task
+        except asyncio.CancelledError:
+            pass
+        logger.info("消息消费循环已停止")
+
 
 # 创建 FastAPI 应用实例
 app = FastAPI(

backend/app/services/device_sim_service.py

@@ -6,14 +6,16 @@
 
 作者：czj
 创建日期：2026-07-29
-最后修改：2026-07-29
+最后修改：2026-08-05
 """
 
 import json
 import logging
 import threading
+import asyncio
 from datetime import datetime, timedelta
 from typing import Optional, List, Tuple
+from dataclasses import dataclass
 
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy import select, func, delete, and_
@@ -31,6 +33,55 @@ from app.config import settings
 
 logger = logging.getLogger(__name__)
 
+# ==================== 消息队列 ====================
+
+@dataclass
+class QueueMessage:
+    """消息队列数据结构"""
+    device_id: str
+    device_name: str
+    device_type: str
+    topic: str
+    payload: dict
+    direction: str
+    status: str
+    error_message: Optional[str]
+    timestamp: datetime
+
+
+# 全局消息队列（最大 10000 条）
+_message_queue: asyncio.Queue = None
+_queue_stats = {"enqueued": 0, "processed": 0, "dropped": 0}
+
+
+def get_message_queue() -> asyncio.Queue:
+    """获取消息队列实例"""
+    global _message_queue
+    if _message_queue is None:
+        _message_queue = asyncio.Queue(maxsize=10000)
+    return _message_queue
+
+
+def enqueue_message(msg: QueueMessage) -> bool:
+    """
+    将消息入队（非阻塞，在模拟器线程中调用）
+
+    Returns:
+        bool: 是否入队成功
+    """
+    queue = get_message_queue()
+    try:
+        queue.put_nowait(msg)
+        _queue_stats["enqueued"] += 1
+        return True
+    except asyncio.QueueFull:
+        _queue_stats["dropped"] += 1
+        logger.warning(f"消息队列已满，丢弃消息: device_id={msg.device_id}")
+        return False
+
+
+# ==================== 消息推送回调 ====================
+
 # 消息推送回调函数（由 WebSocket 路由设置）
 _message_callback = None
 
@@ -549,16 +600,39 @@ class DeviceSimService:
             return success
 
     def _sync_report_log(self, kwargs: dict) -> None:
-        """同步记录上报日志（在线程中执行）"""
-        import asyncio
+        """
+        同步记录上报日志（在线程中执行）
+
+        现在只做入队操作，实际的数据库写入在消费循环中完成。
+        """
         try:
-            # 使用 asyncio.run 创建新的事件循环
-            loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(loop)
-            loop.run_until_complete(self._create_report_log(**kwargs))
-            loop.close()
+            # 从 kwargs 获取设备信息
+            device_id = kwargs.get("device_id", "")
+
+            # 从全局运行中的模拟器获取设备名称和类型
+            with _running_simulators_lock:
+                sim = _running_simulators.get(device_id)
+                device_name = sim.device_name if sim else device_id
+                device_type = sim.device_type if sim else "unknown"
+
+            # 创建队列消息
+            msg = QueueMessage(
+                device_id=device_id,
+                device_name=device_name,
+                device_type=device_type,
+                topic=kwargs.get("topic", ""),
+                payload=kwargs.get("payload", {}),
+                direction=kwargs.get("direction", "publish"),
+                status=kwargs.get("status", "success"),
+                error_message=kwargs.get("error_message"),
+                timestamp=datetime.now(),
+            )
+
+            # 入队（非阻塞）
+            enqueue_message(msg)
+
         except Exception as e:
-            logger.error(f"同步记录上报日志异常: {e}")
+            logger.error(f"消息入队异常: {e}")
 
     async def _create_report_log(self, device_id: str, topic: str,
                                   payload: dict, direction: str = "publish",
@@ -587,23 +661,12 @@ class DeviceSimService:
 
             await self.db.flush()
 
-            # 触发 WebSocket 推送回调
-            if _message_callback and sim:
-                try:
-                    await _message_callback(
-                        device_id=device_id,
-                        device_name=sim.device_name,
-                        device_type=sim.device_type,
-                        topic=topic,
-                        direction=direction,
-                        payload=payload,
-                    )
-                except Exception as e:
-                    logger.warning(f"WebSocket 推送回调异常: {e}")
-
         except Exception as e:
             logger.error(f"创建上报记录异常: {e}")
 
+        # TODO: WebSocket 推送回调在模拟器线程中执行有问题，暂时禁用
+        # 需要重构为使用消息队列或异步任务来推送
+
     async def stop_simulator(self, device_id: str) -> bool:
         """
         停止模拟设备
@@ -812,4 +875,88 @@ class DeviceSimService:
         cutoff = datetime.now() - timedelta(days=days)
         query = delete(ReportLog).where(ReportLog.created_at < cutoff)
         result = await self.db.execute(query)
-        return result.rowcount or 0
\ No newline at end of file
+        return result.rowcount or 0
+
+
+# ==================== 消息消费循环 ====================
+
+async def message_consumer_loop():
+    """
+    消息消费循环（在主线程异步循环中运行）
+
+    从消息队列中取出消息，写入数据库并触发 WebSocket 推送。
+    """
+    from app.database import async_session_maker
+
+    logger.info("消息消费循环已启动")
+
+    while True:
+        try:
+            # 从队列获取消息（阻塞等待）
+            queue = get_message_queue()
+            msg: QueueMessage = await queue.get()
+
+            # 创建数据库 session
+            async with async_session_maker() as db:
+                try:
+                    # 写入上报记录
+                    log = ReportLog(
+                        id=generate_id("log"),
+                        device_id=msg.device_id,
+                        topic=msg.topic,
+                        payload=msg.payload,
+                        direction=msg.direction,
+                        status=msg.status,
+                        error_message=msg.error_message,
+                    )
+                    db.add(log)
+
+                    # 更新设备统计
+                    query = select(DeviceSimulator).where(DeviceSimulator.id == msg.device_id)
+                    result = await db.execute(query)
+                    sim = result.scalar_one_or_none()
+                    if sim:
+                        sim.total_reports = (sim.total_reports or 0) + 1
+                        sim.last_reported_at = datetime.now()
+                        # 更新设备名称和类型（用于 WebSocket 推送）
+                        msg.device_name = sim.device_name
+                        msg.device_type = sim.device_type
+
+                    await db.commit()
+                    _queue_stats["processed"] += 1
+
+                except Exception as e:
+                    await db.rollback()
+                    logger.error(f"写入上报记录失败: {e}")
+
+            # WebSocket 推送（不阻塞消费循环）
+            if _message_callback:
+                try:
+                    await _message_callback(
+                        device_id=msg.device_id,
+                        device_name=msg.device_name,
+                        device_type=msg.device_type,
+                        topic=msg.topic,
+                        direction=msg.direction,
+                        payload=msg.payload,
+                    )
+                except Exception as e:
+                    logger.debug(f"WebSocket 推送失败: {e}")
+
+        except asyncio.CancelledError:
+            logger.info("消息消费循环已停止")
+            break
+        except Exception as e:
+            logger.error(f"消息消费循环异常: {e}")
+            await asyncio.sleep(1)  # 避免快速失败循环
+
+
+def get_queue_stats() -> dict:
+    """获取队列统计信息"""
+    queue = get_message_queue()
+    return {
+        "queue_size": queue.qsize(),
+        "enqueued": _queue_stats["enqueued"],
+        "processed": _queue_stats["processed"],
+        "dropped": _queue_stats["dropped"],
+    }
\ No newline at end of file