import multiprocessing import subprocess import time import threading import os import signal import json from datetime import datetime from typing import Any, Dict, List, Optional import sys import hashlib import platform PROGRAM_DIR = "/opt/lnxall_app/bin" CONFIG_PATH = "/app/config/device_config.json" BIN_FILE = ["bmser_trans", "bmser_alarm", "bmser_upgrade"] class ProcessManager: def __init__(self): self.processes: Dict[str, subprocess.Popen] = {} self.process_info: Dict[str, dict] = {} self.monitoring = False self.last_hash = "" self.condition_met = False self.lock = threading.Lock() self.config = { "monitor_interval": 5, # 监控间隔(秒) "max_restart_attempts": 1000, # 最大重启尝试次数 "health_check_timeout": 10, # 健康检查超时时间 "programs": { "bmser_trans": { "path": "/opt/lnxall_app/bin/bmser_trans", "args": [], "condition_check_interval": 5, "restart_delay": 2 }, "bmser_alarm": { "path": "/opt/lnxall_app/bin/bmser_alarm", "args": [], "condition_check_interval": 5, "restart_delay": 2 }, "bmser_upgrade": { "path": "/opt/lnxall_app/bin/bmser_upgrade", "args": [], "condition_check_interval": 5, "restart_delay": 2 } } } def get_protocol(self): try: with open(CONFIG_PATH, 'r', encoding='utf-8') as f: config: Dict[str, Any] = json.load(f) self.condition_met = False for item in config.get("dev_class", []): for dev in item.get("devs", []): channel = dev.get("channel", "") if "bmser_lib" in channel: self.condition_met = True return True return False except Exception as e: print(f"Unexpected error in get_protocol: {e}", file=sys.stderr) return False def get_file_hash(self): try: with open(CONFIG_PATH, 'rb') as f: return hashlib.md5(f.read()).hexdigest() except Exception: return None def check_config_change(self): current_hash = self.get_file_hash() if current_hash is None or current_hash == self.last_hash: return False self.last_hash = current_hash return True def log(self, message: str): timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") print(f"[{timestamp}] {message}") def check_condition(self) -> bool: try: if self.check_config_change(): return self.get_protocol() return False except Exception as e: self.log(f"条件检查失败: {e}") return False def start_program(self, program_name: str) -> bool: with self.lock: if program_name in self.processes and self.is_process_running(program_name): self.log(f"程序 {program_name} 已在运行") return True program_config = self.config["programs"].get(program_name) if not program_config: self.log(f"未找到程序 {program_name} 的配置") return False try: cmd = [program_config["path"]] + program_config["args"] process = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True ) self.processes[program_name] = process self.process_info[program_name] = { "start_time": datetime.now(), "restart_count": 0, "last_restart": None } self.log(f"程序 {program_name} 已启动 (PID: {process.pid})") return True except Exception as e: self.log(f"启动程序 {program_name} 失败: {e}") return False def stop_program(self, program_name: str) -> bool: """停止指定程序""" with self.lock: if program_name not in self.processes: return True process = self.processes[program_name] try: # 先尝试优雅终止 process.terminate() try: process.wait(timeout=5) except subprocess.TimeoutExpired: # 强制终止 process.kill() process.wait() del self.processes[program_name] self.log(f"程序 {program_name} 已停止") return True except Exception as e: self.log(f"停止程序 {program_name} 失败: {e}") return False def is_process_running(self, program_name: str) -> bool: """检查程序是否在运行""" if program_name not in self.processes: return False process = self.processes[program_name] return process.poll() is None def is_process_zombie(self, program_name: str) -> bool: """检查并处理僵尸进程""" zombie_count = 0 try: # Unix/Linux系统使用ps命令查找僵尸进程 result = subprocess.run( "ps -A -o pid,state | grep 'Z'", capture_output=True, text=True, shell=True ) if result.returncode == 0: lines = result.stdout.strip().split('\n') for line in lines: if line.strip(): parts = line.split() if len(parts) >= 2 and parts[1] == 'Z': pid = int(parts[0]) try: # 尝试回收僵尸进程 os.waitpid(pid, 0) zombie_count += 1 print(f"已回收僵尸进程 PID: {pid}") except: pass except Exception as e: print(f"检查僵尸进程时发生错误: {e}") def restart_program(self, program_name: str) -> bool: """重启指定程序""" self.log(f"正在重启程序 {program_name}") # 先停止 self.stop_program(program_name) # 添加重启延迟 program_config = self.config["programs"].get(program_name, {}) restart_delay = program_config.get("restart_delay", 2) time.sleep(restart_delay) # 再启动 return self.start_program(program_name) def monitor_processes(self): """监控所有进程状态""" while self.monitoring: time.sleep(self.config["monitor_interval"]) try: self.check_condition() condition_met = self.condition_met for program_name in BIN_FILE: if not condition_met: self.stop_program(program_name) continue if not self.is_process_running(program_name): self.log(f"程序 {program_name} 已退出，尝试重启") self.restart_program(program_name) elif self.is_process_zombie(program_name): self.log(f"程序 {program_name} 进入僵尸状态，强制重启") self.restart_program(program_name) if program_name in self.process_info: info = self.process_info[program_name] max_attempts = self.config.get("max_restart_attempts", 3) if info["restart_count"] >= max_attempts: self.log(f"程序 {program_name} 已达到最大重启次数，停止重启") continue except Exception as e: self.log(f"监控过程中发生错误: {e}") def start_monitoring(self): """开始监控""" if self.monitoring: self.log("监控已在运行") return self.monitoring = True monitor_thread = threading.Thread(target=self.monitor_processes, daemon=True) monitor_thread.start() self.log("进程监控已启动") def stop_monitoring(self): """停止监控""" self.monitoring = False self.log("进程监控已停止") # 停止所有程序 for program_name in list(self.processes.keys()): self.stop_program(program_name) def main(): manager = ProcessManager() manager.monitoring = True try: manager.log("进程监控启动") manager.monitor_processes() except Exception as e: manager.log(f"监控过程中发生错误: {e}") manager.stop_monitoring() if __name__ == "__main__": main()