本文目录导读：
进程监控（基础版）
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()
使用建议
-
选择合适的监控方式：
- 简单应用：进程监控即可
- Web服务：HTTP监控更准确
- 关键服务：组合多种监控方式
-
配置合理的检查间隔：
- 重要服务：10-30秒
- 普通服务：1-5分钟
- 避免过于频繁导致资源浪费
-
设置完善的告警机制：
- 邮件、短信、钉钉/企业微信通知
- 分级告警（警告/严重/紧急）
- 设置告警抑制，避免重复通知
-
日志管理：
- 记录所有检查结果
- 设置日志轮转，避免占用过多磁盘
- 定期分析日志，发现潜在问题
这些脚本可以根据你的具体需求进行修改和扩展,需要我帮你针对特定场景进行优化吗？
