diff --git a/fam-core/src/fam_core/app.py b/fam-core/src/fam_core/app.py index 5428abb..9fe350d 100644 --- a/fam-core/src/fam_core/app.py +++ b/fam-core/src/fam_core/app.py @@ -5,6 +5,8 @@ FAM-Core 主应用 - Flask 单进程 """ import os import sys +import time +import threading from flask import Flask, jsonify # 确保包路径 @@ -64,14 +66,31 @@ except Exception as e: @app.route('/api/status', methods=['GET']) def status(): - """系统状态""" + """系统状态(检查线程实际存活)""" return jsonify({ - "scheduler_running": _scheduler._running if _scheduler else False, - "dispatcher_running": _dispatcher._running if _dispatcher else False, - "poller_running": _poller._running if _poller else False, + "scheduler_running": _scheduler.is_alive() if _scheduler else False, + "dispatcher_running": _dispatcher.is_alive() if _dispatcher else False, + "poller_running": _poller.is_alive() if _poller else False, }), 200 +def _watchdog_run(): + """看门狗:每 60s 检查线程存活,崩溃自动重启""" + logger.info("Watchdog 线程启动,检查间隔 60s") + while True: + time.sleep(60) + for comp, name in [(_scheduler, 'Scheduler'), (_dispatcher, 'Dispatcher'), (_poller, 'Poller')]: + if comp and hasattr(comp, 'check_and_restart'): + try: + comp.check_and_restart() + except Exception as e: + logger.error(f"Watchdog 重启 {name} 失败: {e}", exc_info=True) + + +_watchdog_thread = threading.Thread(target=_watchdog_run, daemon=True, name='watchdog') +_watchdog_thread.start() + + if __name__ == '__main__': cfg = load_config() port = cfg.get('server', {}).get('port', 8000) diff --git a/fam-core/src/fam_core/dispatcher/dispatcher.py b/fam-core/src/fam_core/dispatcher/dispatcher.py index 9270e25..4c88409 100644 --- a/fam-core/src/fam_core/dispatcher/dispatcher.py +++ b/fam-core/src/fam_core/dispatcher/dispatcher.py @@ -9,8 +9,9 @@ Dispatcher - 30s 轮询 PENDING 任务,上传视频至 Edge 异步队列 4. Dispatcher 标记任务为 PROCESSING(已派发,等待 Poller 拉取结果) 5. Poller 线程定期从 Edge /api/edge/results 拉取结果,写库后标记 SUCCESS -退避重试: min(60 * (retry_count + 1) * 2, 600) 秒 +退避重试: min(30 * (retry_count + 1), 300) 秒 分块级重试: 每块最多重试 3 次 +看门狗: 线程崩溃后自动重启 """ import os import io @@ -51,8 +52,8 @@ class Dispatcher: self._thread = None def _calculate_backoff(self, retry_count): - """退避策略: min(60 * (retry_count + 1) * 2, 600)""" - return min(60 * (retry_count + 1) * 2, 600) + """退避策略: min(30 * (retry_count + 1), 300)""" + return min(30 * (retry_count + 1), 300) def _should_retry(self, task): """检查任务是否可以重试""" @@ -275,7 +276,7 @@ class Dispatcher: except Exception as e: logger.error(f"僵尸任务回收失败: {e}", exc_info=True) - tasks = db_layer.get_pending_tasks(limit=10) + tasks = db_layer.get_pending_tasks(limit=1) for task in tasks: if self._should_retry(task): try: @@ -301,6 +302,17 @@ class Dispatcher: self._thread = threading.Thread(target=self._run, daemon=True, name='dispatcher') self._thread.start() + def is_alive(self): + """线程是否存活""" + return self._thread is not None and self._thread.is_alive() + + def check_and_restart(self): + """看门狗:线程崩溃后自动重启""" + if self._running and not self.is_alive(): + logger.warning("Dispatcher 线程已死亡,正在重启...") + self._thread = threading.Thread(target=self._run, daemon=True, name='dispatcher') + self._thread.start() + def stop(self): """停止下发线程""" self._running = False diff --git a/fam-core/src/fam_core/poller/poller.py b/fam-core/src/fam_core/poller/poller.py index 62cf0ca..b6350ad 100644 --- a/fam-core/src/fam_core/poller/poller.py +++ b/fam-core/src/fam_core/poller/poller.py @@ -121,6 +121,17 @@ class Poller: self._thread = threading.Thread(target=self._run, daemon=True, name='poller') self._thread.start() + def is_alive(self): + """线程是否存活""" + return self._thread is not None and self._thread.is_alive() + + def check_and_restart(self): + """看门狗:线程崩溃后自动重启""" + if self._running and not self.is_alive(): + logger.warning("Poller 线程已死亡,正在重启...") + self._thread = threading.Thread(target=self._run, daemon=True, name='poller') + self._thread.start() + def stop(self): """停止拉取线程""" self._running = False diff --git a/fam-core/src/fam_core/scheduler/scheduler.py b/fam-core/src/fam_core/scheduler/scheduler.py index 6c91499..f76e88c 100644 --- a/fam-core/src/fam_core/scheduler/scheduler.py +++ b/fam-core/src/fam_core/scheduler/scheduler.py @@ -104,6 +104,17 @@ class TaskScheduler: self._thread = threading.Thread(target=self._run, daemon=True, name='task-scheduler') self._thread.start() + def is_alive(self): + """线程是否存活""" + return self._thread is not None and self._thread.is_alive() + + def check_and_restart(self): + """看门狗:线程崩溃后自动重启""" + if self._running and not self.is_alive(): + logger.warning("Scheduler 线程已死亡,正在重启...") + self._thread = threading.Thread(target=self._run, daemon=True, name='task-scheduler') + self._thread.start() + def stop(self): """停止调度线程""" self._running = False