背景
我们写的软件需要调用一个第三方的厂商的硬件,第三方厂商提供了一个 websocket 的服务部署在 windows 系统上,我们的软件通过与这个 websocket 服务通信来调用第三方硬件。所以这个 websocket 程序一定不能挂掉,否则我们的软件也就挂了。但是事与愿违,这个 websocket 程序经常莫名其妙的挂掉了,经过观察,有可能是写的不好,CPU 占用有时会很高,被系统干掉了。
所以守护这个 websocket 程序是至关重要。
windows服务
windows 系统上有个功能是服务,这个功能可以实现开机自启动程序。服务的注册可以使用 sc 命令
sc.exe [<servername>] create [<servicename>] [type= {own | share | kernel | filesys | rec | interact type= {own | share}}] [start= {boot | system | auto | demand | disabled | delayed-auto}] [error= {normal | severe | critical | ignore}] [binpath= <binarypathname>] [group= <loadordergroup>] [tag= {yes | no}] [depend= <dependencies>] [obj= {<accountname> | <objectname>}] [displayname= <displayname>] [password= <password>]
但有两个问题:
- 直接使用 exe 程序创建出来的服务可能不能启动!
- 服务被干掉是不能自动重启的
NSSM
NSSM (Non-Sucking Service Manager)是一款专为Windows系统设计的开源工具,可将普通可执行程序(如exe、批处理脚本、Python/Node.js应用等)注册为系统服务,实现后台静默运行、开机自启动、崩溃自动重启和日志管理等功能。
- 下载 NSSM
- 打开命令提示符(以管理员身份运行),然后切换至nssm所在的目录,运行
nssm.exe install
- 在弹出的对话框中选择要运行的程序,设置服务名称等,然后点击 install service
- 服务注册完成后,可以通过 windows 上的服务功能中找到,并可以启动;也可以使用
nssm start <服务名>
启动服务
使用python监控
除了使用 nssm,还可以编写一个 python 脚本,定期检查程序是否在运行,如果没有运行,直接运行 exe
import time
import psutil
import subprocess
import os
# 配置部分
PROCESS_NAME = "XXX.exe" # 要监控的进程名(必须与 exe 名一致)
EXE_PATH = r"C:\XXX.exe" # 要启动的程序路径,请替换为你自己的路径
CHECK_INTERVAL = 60 # 检查间隔(秒)
def is_process_running(process_name):
"""检查是否有指定名称的进程正在运行"""
for proc in psutil.process_iter(['pid', 'name']):
try:
if proc.info['name'] == process_name:
return True
except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
continue
return False
def start_process(exe_path):
"""启动指定路径的程序"""
if os.path.exists(exe_path):
try:
subprocess.Popen(exe_path)
print(f"[INFO] 已启动程序: {exe_path}")
except Exception as e:
print(f"[ERROR] 启动程序失败: {e}")
else:
print(f"[ERROR] 文件不存在: {exe_path}")
def main():
print("[INFO] 启动监控服务...")
while True:
if not is_process_running(PROCESS_NAME):
print(f"[WARNING] {PROCESS_NAME} 未运行,尝试启动...")
start_process(EXE_PATH)
else:
print(f"[INFO] {PROCESS_NAME} 正在运行中...")
time.sleep(CHECK_INTERVAL)
if __name__ == "__main__":
main()
上面的功能也可以直接注册为 windows 的服务,参考以下代码
import time
import os
import psutil
import subprocess
import win32serviceutil
import win32service
import win32event
PROCESS_NAME = "XXX.exe" # 要监控的进程名
EXE_PATH = r"C:\XXX.exe" # 程序路径
CHECK_INTERVAL = 60 # 检查间隔(秒)
def is_process_running(process_name):
"""检查是否有指定名称的进程正在运行"""
for proc in psutil.process_iter(['pid', 'name']):
try:
if proc.info['name'] == process_name:
return True
except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
continue
return False
def start_process(exe_path):
"""启动指定路径的程序"""
if os.path.exists(exe_path):
try:
subprocess.Popen(exe_path)
print(f"[INFO] 已启动程序: {exe_path}")
except Exception as e:
print(f"[ERROR] 启动程序失败: {e}")
else:
print(f"[ERROR] 文件不存在: {exe_path}")
class MonitorService(win32serviceutil.ServiceFramework):
_svc_name_ = "ProcessMonitorService"
_svc_display_name_ = "进程监控服务"
_svc_description_ = "每分钟检测某个程序是否运行,如果没有运行则自动启动。"
def __init__(self, args):
win32serviceutil.ServiceFramework.__init__(self, args)
self.hWaitStop = win32event.CreateEvent(None, 0, 0, None)
self.is_alive = True
def SvcDoRun(self):
import servicemanager
servicemanager.LogMsg(servicemanager.EVENTLOG_INFORMATION_TYPE,
servicemanager.PYS_SERVICE_STARTED,
(self._svc_name_, ''))
self.main()
def SvcStop(self):
self.ReportServiceStatus(win32service.SERVICE_STOP_PENDING)
win32event.SetEvent(self.hWaitStop)
self.is_alive = False
def main(self):
while self.is_alive:
if not is_process_running(PROCESS_NAME):
print(f"[WARNING] {PROCESS_NAME} 未运行,尝试启动...")
start_process(EXE_PATH)
else:
print(f"[INFO] {PROCESS_NAME} 正在运行中...")
time.sleep(CHECK_INTERVAL)
if __name__ == '__main__':
win32serviceutil.HandleCommandLine(MonitorService)
需要使用 pip 安装依赖
pip install psutil
pip install pywin32
使用方法
安装服务 python MonitorService.py install
启动服务 python MonitorService.py start
停止服务 python MonitorService.py stop
卸载服务 python MonitorService.py remove