脚本如何监控软件运行状态

本文目录导读：

1. 进程监控（基础版）
2. 服务监控（进阶版）
3. HTTP服务监控
4. 综合监控脚本（带日志和告警）
5. 系统资源监控（扩展）
6. 使用建议

进程监控（基础版）

Windows (Batch/PowerShell)

@echo off
:check
tasklist | find /i "notepad.exe" >nul
if errorlevel 1 (
    echo Notepad is not running! Starting...
    start notepad.exe
) else (
    echo Notepad is running normally
)
timeout /t 5 /nobreak
goto check

Linux (Shell)

#!/bin/bash
PROCESS_NAME="nginx"
while true; do
    if pgrep -x "$PROCESS_NAME" > /dev/null; then
        echo "$(date): $PROCESS_NAME is running"
    else
        echo "$(date): $PROCESS_NAME is not running! Restarting..."
        systemctl start $PROCESS_NAME
    fi
    sleep 5
done

服务监控（进阶版）

Python 通用脚本

import psutil
import time
import subprocess
import logging
# 配置日志
logging.basicConfig(level=logging.INFO, 
                    format='%(asctime)s - %(levelname)s - %(message)s')
def check_process_status(process_name):
    """检查进程是否存在"""
    for proc in psutil.process_iter(['name', 'pid', 'status']):
        if proc.info['name'] == process_name:
            return True
    return False
def monitor_process(process_name, restart_command):
    """监控进程并自动重启"""
    while True:
        if check_process_status(process_name):
            logging.info(f"{process_name} is running normally")
        else:
            logging.warning(f"{process_name} is down! Attempting restart...")
            try:
                subprocess.run(restart_command, shell=True, check=True)
                logging.info(f"{process_name} restarted successfully")
            except Exception as e:
                logging.error(f"Failed to restart {process_name}: {e}")
        time.sleep(10)
# 使用示例
if __name__ == "__main__":
    monitor_process("chrome.exe", "start chrome.exe")

HTTP服务监控

import requests
import time
import smtplib
from email.mime.text import MIMEText
def check_http_service(url, timeout=10):
    """检查HTTP服务是否正常"""
    try:
        response = requests.get(url, timeout=timeout)
        if response.status_code == 200:
            return True, f"Service is UP (Status: {response.status_code})"
        else:
            return False, f"Service is responding but status: {response.status_code}"
    except requests.exceptions.RequestException as e:
        return False, f"Service is DOWN: {str(e)}"
def send_alert(subject, body):
    """发送告警通知（示例使用邮件）"""
    msg = MIMEText(body)
    msg['Subject'] = subject
    msg['From'] = '[email protected]'
    msg['To'] = '[email protected]'
    # 配置你的邮件服务器
    # with smtplib.SMTP('smtp.gmail.com', 587) as server:
    #     server.login('user', 'password')
    #     server.send_message(msg)
if __name__ == "__main__":
    while True:
        status, message = check_http_service("https://your-service.com")
        print(f"{time.ctime()}: {message}")
        if not status:
            send_alert("Service Alert", message)
        time.sleep(30)

综合监控脚本（带日志和告警）

#!/usr/bin/env python3
import os
import sys
import time
import json
import logging
import subprocess
from datetime import datetime
from typing import Dict, Any
class ServiceMonitor:
    def __init__(self, config_file='monitor_config.json'):
        self.config = self.load_config(config_file)
        self.setup_logging()
    def load_config(self, config_file):
        """加载配置文件"""
        default_config = {
            "services": [
                {
                    "name": "nginx",
                    "type": "process",
                    "check_command": "pgrep nginx",
                    "restart_command": "systemctl restart nginx",
                    "interval": 10
                },
                {
                    "name": "web-app",
                    "type": "http",
                    "url": "http://localhost:8080/health",
                    "expected_status": 200,
                    "interval": 30
                }
            ],
            "alert": {
                "method": "log",
                "email": "[email protected]"
            }
        }
        if os.path.exists(config_file):
            with open(config_file, 'r') as f:
                return json.load(f)
        return default_config
    def setup_logging(self):
        """设置日志"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('monitor.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger('ServiceMonitor')
    def check_process(self, service: Dict[str, Any]) -> bool:
        """检查进程状态"""
        try:
            result = subprocess.run(
                service['check_command'], 
                shell=True, 
                capture_output=True,
                timeout=5
            )
            return result.returncode == 0
        except Exception as e:
            self.logger.error(f"Process check failed: {e}")
            return False
    def check_http(self, service: Dict[str, Any]) -> bool:
        """检查HTTP服务"""
        import requests
        try:
            response = requests.get(
                service['url'], 
                timeout=service.get('timeout', 10)
            )
            return response.status_code == service.get('expected_status', 200)
        except Exception as e:
            self.logger.error(f"HTTP check failed: {e}")
            return False
    def restart_service(self, service: Dict[str, Any]) -> bool:
        """重启服务"""
        try:
            result = subprocess.run(
                service['restart_command'],
                shell=True,
                capture_output=True,
                timeout=30
            )
            if result.returncode == 0:
                self.logger.info(f"Service {service['name']} restarted successfully")
                return True
            else:
                self.logger.error(f"Service restart failed: {result.stderr.decode()}")
                return False
        except Exception as e:
            self.logger.error(f"Service restart error: {e}")
            return False
    def send_alert(self, service_name: str, status: str):
        """发送告警通知"""
        alert_method = self.config['alert']['method']
        message = f"Alert: Service {service_name} is {status}"
        if alert_method == 'log':
            self.logger.warning(message)
        elif alert_method == 'email':
            # 邮件发送代码
            pass
        elif alert_method == 'webhook':
            # Webhook发送代码
            pass
    def monitor_service(self, service: Dict[str, Any]):
        """监控单个服务"""
        if service['type'] == 'process':
            status = self.check_process(service)
        elif service['type'] == 'http':
            status = self.check_http(service)
        else:
            self.logger.error(f"Unknown service type: {service['type']}")
            return
        if not status:
            self.logger.warning(f"Service {service['name']} is DOWN")
            self.send_alert(service['name'], "down")
            # 自动重启
            if 'restart_command' in service:
                if self.restart_service(service):
                    self.send_alert(service['name'], "restarted")
                else:
                    self.send_alert(service['name'], "restart_failed")
        else:
            self.logger.info(f"Service {service['name']} is UP")
    def run(self):
        """运行主监控循环"""
        self.logger.info("Service Monitor started")
        while True:
            for service in self.config['services']:
                try:
                    self.monitor_service(service)
                    time.sleep(service.get('interval', 10))
                except Exception as e:
                    self.logger.error(f"Monitor error for {service['name']}: {e}")
            time.sleep(1)
if __name__ == "__main__":
    monitor = ServiceMonitor()
    monitor.run()

系统资源监控（扩展）

import psutil
import time
def monitor_system():
    """监控系统资源"""
    while True:
        # CPU使用率
        cpu_percent = psutil.cpu_percent(interval=1)
        # 内存使用率
        memory = psutil.virtual_memory()
        memory_percent = memory.percent
        # 磁盘使用率
        disk = psutil.disk_usage('/')
        disk_percent = disk.percent
        # 网络IO
        net_io = psutil.net_io_counters()
        print(f"CPU: {cpu_percent}% | Memory: {memory_percent}% | Disk: {disk_percent}%")
        # 阈值告警
        if cpu_percent > 90:
            print("⚠️ CPU usage critical!")
        if memory_percent > 90:
            print("⚠️ Memory usage critical!")
        if disk_percent > 90:
            print("⚠️ Disk usage critical!")
        time.sleep(5)
if __name__ == "__main__":
    monitor_system()

使用建议

1. 选择合适的监控方式：

   • 简单应用：进程监控即可
   • Web服务：HTTP监控更准确
   • 关键服务：组合多种监控方式

2. 配置合理的检查间隔：

   • 重要服务：10-30秒
   • 普通服务：1-5分钟
   • 避免过于频繁导致资源浪费

3. 设置完善的告警机制：

   • 邮件、短信、钉钉/企业微信通知
   • 分级告警（警告/严重/紧急）
   • 设置告警抑制，避免重复通知

4. 日志管理：

   • 记录所有检查结果
   • 设置日志轮转，避免占用过多磁盘
   • 定期分析日志，发现潜在问题

这些脚本可以根据你的具体需求进行修改和扩展,需要我帮你针对特定场景进行优化吗？