From 881ea3f470fbc6573bfbf602400b411660cd662f Mon Sep 17 00:00:00 2001 From: ericwyuan Date: Thu, 20 Aug 2026 13:37:07 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20dispatcher=20=E4=B8=B2=E8=A1=8C=E4=B8=8A?= =?UTF-8?q?=E4=BC=A0=20+=20=E7=9C=8B=E9=97=A8=E7=8B=97=E8=87=AA=E5=8A=A8?= =?UTF-8?q?=E9=87=8D=E5=90=AF=20+=20=E7=BA=BF=E7=A8=8B=E7=8A=B6=E6=80=81?= =?UTF-8?q?=E7=9C=9F=E5=AE=9E=E6=A3=80=E6=B5=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Dispatcher limit=10→1: 一次只传一个视频，传完再传下一个 - 退避缩短: min(30*(n+1),300)s → 失败后更快重试（原 min(60*(n+1)*2,600)s） - /api/status: 用 thread.is_alive() 替代 _running 布尔标志 - 三组件(scheduler/dispatcher/poller)添加 is_alive()+check_and_restart() - app.py 新增 watchdog 线程: 每 60s 检测线程死亡并自动重启 --- fam-core/src/fam_core/app.py | 27 ++++++++++++++++--- .../src/fam_core/dispatcher/dispatcher.py | 20 +++++++++++--- fam-core/src/fam_core/poller/poller.py | 11 ++++++++ fam-core/src/fam_core/scheduler/scheduler.py | 11 ++++++++ 4 files changed, 61 insertions(+), 8 deletions(-) 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