""" 系统管理器 管理所有子系统的生命周期，提供动态挂载/卸载、故障监控等功能 """ import logging import threading import time import importlib.util import pathlib from typing import Dict, Optional, List, Any from datetime import datetime from base_subsystem import BaseSubsystem logger = logging.getLogger(__name__) system_manager_lock = threading.RLock() config=[] def import_all_from_folder(folder_path: str): global config """动态导入指定文件夹下的所有Python模块""" folder = pathlib.Path(folder_path) if not folder.is_dir(): logger.warning(f"路径 '{folder_path}' 不是有效的文件夹") return for file in folder.glob("*.py"): module_name = file.stem spec = importlib.util.spec_from_file_location(module_name, str(file)) if spec and spec.loader: module = importlib.util.module_from_spec(spec) globals()[module_name] = module spec.loader.exec_module(module) #logger.info(f"已导入模块: {module_name} from {file}") #print(f"已导入模块: {module_name} from {file}") config.extend(module.config) else: logger.warning(f"无法导入模块: {module_name} from {file}") folder_path = pathlib.Path(__file__).parent / "subsystems" import_all_from_folder(str(folder_path)) class SystemManager: """ 系统管理器 主要功能： 1. 管理多个独立的子系统 2. 支持动态挂载和卸载子系统 3. 监控子系统健康状态 4. 隔离子系统故障，防止连锁反应 5. 提供统一的系统状态查询接口 """ def __init__(self): """ 初始化系统管理器 Args: db_manager: 数据库管理器（所有子系统共享） message_bus: 消息总线（所有子系统共享） """ # 子系统容器 self._subsystems: Dict[str, BaseSubsystem] = {} self._subsystems_lock = threading.RLock() # 递归锁，支持嵌套锁定 # 管理器状态 self._running = False self._monitor_thread: Optional[threading.Thread] = None self._stop_event = threading.Event() # 监控配置 self._monitor_interval = 30 # 检查间隔（秒） self._health_check_timeout = 5 # 健康检查超时 self._auto_restart = True # 是否自动重启失败的子系统 # 系统统计 self._stats = { 'started_at': None, 'stopped_at': None, 'total_subsystems_added': 0, 'total_subsystems_removed': 0, 'subsystem_restarts': {} # {subsystem_name: restart_count} } logger.info("系统管理器已初始化") # ======================================================================== # 子系统管理 # ======================================================================== def register_subsystem(self, subsystem: BaseSubsystem, autostart: bool = True) -> bool: """ 注册（挂载）一个子系统 Args: subsystem: 子系统实例 autostart: 是否自动启动 Returns: 是否注册成功 """ with self._subsystems_lock: if subsystem.name in self._subsystems: logger.warning(f"子系统 '{subsystem.name}' 已存在，跳过注册") return False try: self._subsystems[subsystem.name] = subsystem self._stats['total_subsystems_added'] += 1 self._stats['subsystem_restarts'][subsystem.name] = 0 logger.info(f"子系统 '{subsystem.name}' 已注册") # 如果系统已运行且需要自动启动 if self._running and autostart: if subsystem.start(): logger.info(f"子系统 '{subsystem.name}' 已自动启动") else: logger.error(f"自动启动子系统 '{subsystem.name}' 失败") return False return True except Exception as e: logger.error(f"注册子系统 '{subsystem.name}' 失败: {e}", exc_info=True) return False def unregister_subsystem(self, subsystem_name: str, autostop: bool = True) -> bool: """ 卸载一个子系统 Args: subsystem_name: 子系统名称 autostop: 是否自动停止 Returns: 是否卸载成功 """ with self._subsystems_lock: if subsystem_name not in self._subsystems: logger.warning(f"子系统 '{subsystem_name}' 不存在") return False try: subsystem = self._subsystems[subsystem_name] # 如果需要自动停止 if autostop and subsystem.is_running(): subsystem.stop(timeout=10) del self._subsystems[subsystem_name] self._stats['total_subsystems_removed'] += 1 logger.info(f"子系统 '{subsystem_name}' 已卸载") return True except Exception as e: logger.error(f"卸载子系统 '{subsystem_name}' 失败: {e}", exc_info=True) return False def get_subsystem(self, subsystem_name: str) -> Optional[BaseSubsystem]: """获取指定的子系统""" with self._subsystems_lock: return self._subsystems.get(subsystem_name) def list_subsystems(self) -> List[str]: """列出所有子系统名称""" with self._subsystems_lock: return list(self._subsystems.keys()) def register_subsystem_from_config(self) -> None: """ 从配置注册子系统 Args: config: 可选的子系统配置列表 """ global config if config is None: # 使用默认配置 config = [] for subsys_conf in config: try: cls = subsys_conf['class'] name = subsys_conf['name'] params = subsys_conf.get('params', {}) subsystem = cls(name, **params) self.register_subsystem(subsystem, autostart=subsys_conf.get('autostart', True)) except Exception as e: logger.error(f"从配置注册子系统失败: {e}", exc_info=True) # ======================================================================== # 生命周期管理 # ======================================================================== def start(self) -> bool: """ 启动系统管理器和所有已注册的子系统 Returns: 是否启动成功 """ try: logger.info("系统管理器启动中...") # 启动所有子系统 with self._subsystems_lock: failed = [] for name, subsystem in self._subsystems.items(): try: if not subsystem.start(): failed.append(name) logger.error(f"启动子系统 '{name}' 失败") except Exception as e: failed.append(name) logger.error(f"启动子系统 '{name}' 异常: {e}", exc_info=True) if failed: logger.warning(f"有 {len(failed)} 个子系统启动失败: {failed}") # 启动监控线程 self._running = True self._stop_event.clear() self._stats['started_at'] = datetime.utcnow().isoformat() self._monitor_thread = threading.Thread( target=self._monitor_loop, name="system-monitor", daemon=False ) self._monitor_thread.start() logger.info(f"系统管理器已启动（{len(self._subsystems)} 个子系统）") return True except Exception as e: logger.error(f"启动系统管理器失败: {e}", exc_info=True) self._running = False return False def stop(self, timeout: int = 30) -> bool: """ 停止系统管理器和所有子系统 Args: timeout: 等待所有子系统停止的总超时时间 Returns: 是否成功停止 """ try: logger.info("系统管理器停止中...") # 通知监控线程停止 self._running = False self._stop_event.set() # 等待监控线程 if self._monitor_thread and self._monitor_thread.is_alive(): self._monitor_thread.join(timeout=5) # 停止所有子系统 with self._subsystems_lock: subsystems_list = list(self._subsystems.items()) # 计算每个子系统的超时时间 per_subsystem_timeout = timeout // max(len(subsystems_list), 1) failed = [] for name, subsystem in subsystems_list: try: if not subsystem.stop(timeout=per_subsystem_timeout): failed.append(name) logger.error(f"停止子系统 '{name}' 失败或超时") except Exception as e: failed.append(name) logger.error(f"停止子系统 '{name}' 异常: {e}", exc_info=True) self._stats['stopped_at'] = datetime.utcnow().isoformat() if failed: logger.warning(f"有 {len(failed)} 个子系统停止失败: {failed}") return False logger.info("系统管理器已停止") return True except Exception as e: logger.error(f"停止系统管理器失败: {e}", exc_info=True) return False # ======================================================================== # 监控和维护 # ======================================================================== def _monitor_loop(self): """ 监控循环 持续检查子系统健康状态，检测故障并自动恢复 """ logger.info("监控线程已启动") while self._running: try: with self._subsystems_lock: subsystems_snapshot = list(self._subsystems.items()) # 检查每个子系统的健康状态 for name, subsystem in subsystems_snapshot: try: health = subsystem.health_check() # 检查子系统是否运行异常 if not health['running']: logger.warning( f"检测到子系统 '{name}' 已停止运行\n" f" - 线程活跃: {health['thread_alive']}\n" f" - 统计: {health['stats']}" ) # 自动重启失败的子系统 if self._auto_restart: logger.info(f"正在自动重启子系统 '{name}'...") if subsystem.start(): self._stats['subsystem_restarts'][name] += 1 logger.info( f"子系统 '{name}' 已重启 " f"(共 {self._stats['subsystem_restarts'][name]} 次)" ) else: logger.error(f"重启子系统 '{name}' 失败") # 检查最后一次错误 if health['stats']['last_error']: logger.warning( f"子系统 '{name}' 上次错误: {health['stats']['last_error']}\n" f" 时间: {health['stats']['last_error_time']}\n" f" 失败消息数: {health['stats']['failed_messages']}" ) except Exception as e: logger.error(f"检查子系统 '{name}' 健康状态失败: {e}", exc_info=True) # 等待直到下一次检查或接收停止信号 self._stop_event.wait(timeout=self._monitor_interval) except Exception as e: logger.error(f"监控循环异常: {e}", exc_info=True) self._stop_event.wait(timeout=self._monitor_interval) logger.info("监控线程已停止") def get_system_status(self) -> Dict[str, Any]: """ 获取完整的系统状态 Returns: 系统状态字典 """ with self._subsystems_lock: subsystems_status = {} for name, subsystem in self._subsystems.items(): try: subsystems_status[name] = subsystem.health_check() except Exception as e: subsystems_status[name] = { 'name': name, 'running': False, 'error': str(e) } return { 'manager_running': self._running, 'timestamp': datetime.utcnow().isoformat(), 'total_subsystems': len(self._subsystems), 'subsystems': subsystems_status, 'stats': self._stats.copy(), 'auto_restart': self._auto_restart, 'monitor_interval': self._monitor_interval } def set_monitor_interval(self, interval: int): """设置监控间隔时间（秒）""" self._monitor_interval = interval logger.info(f"监控间隔已设置为 {interval} 秒") def set_auto_restart(self, enabled: bool): """设置是否自动重启失败的子系统""" self._auto_restart = enabled logger.info(f"自动重启已设置为 {enabled}") _system_manager:Optional[SystemManager]=None def get_system_manager(db_manager, message_bus) -> SystemManager: """获取全局系统管理器实例""" global _system_manager with system_manager_lock: if _system_manager is None: _system_manager = SystemManager() return _system_manager