脚本如何监控CPU使用率告警

本文目录导读：

1. Shell脚本（Linux）
2. Python脚本（跨平台）
3. 配置文件示例 (cpu_alert_config.json)
4. Windows PowerShell脚本
5. Docker监控方案
6. 扩展功能
7. 安装依赖
8. 使用建议

我来介绍几种监控CPU使用率并触发告警的脚本方案：

Shell脚本（Linux）

#!/bin/bash
# CPU告警阈值（百分比）
CPU_THRESHOLD=80
# 检查间隔（秒）
CHECK_INTERVAL=10
# 告警日志文件
ALERT_LOG="/var/log/cpu_alert.log"
while true; do
    # 获取CPU使用率（取1分钟平均）
    CPU_USAGE=$(top -bn1 | grep "Cpu(s)" | awk '{print $2 + $4}' | cut -d. -f1)
    if [ "$CPU_USAGE" -gt "$CPU_THRESHOLD" ]; then
        # 记录告警
        echo "$(date): CPU使用率告警: ${CPU_USAGE}% > ${CPU_THRESHOLD}%" >> $ALERT_LOG
        # 发送邮件告警（需要配置mail服务）
        echo "CPU使用率已达 ${CPU_USAGE}%，请及时处理" | mail -s "CPU告警通知" admin@example.com
        # 发送系统通知
        wall "警告: CPU使用率 ${CPU_USAGE}% 超过阈值 ${CPU_THRESHOLD}%"
    fi
    sleep $CHECK_INTERVAL
done

Python脚本（跨平台）

#!/usr/bin/env python3
import psutil
import time
import logging
import smtplib
from email.mime.text import MIMEText
import json
class CPUAlertMonitor:
    def __init__(self, config_file='cpu_alert_config.json'):
        # 默认配置
        self.config = {
            'cpu_threshold': 80,      # CPU阈值 %
            'check_interval': 10,      # 检查间隔 秒
            'duration_threshold': 60,  # 持续告警时间 秒
            'alert_methods': ['log', 'email'],
            'email': {
                'smtp_server': 'smtp.example.com',
                'smtp_port': 587,
                'sender': 'monitor@example.com',
                'password': 'your_password',
                'recipients': ['admin@example.com']
            }
        }
        # 加载配置文件
        try:
            with open(config_file, 'r') as f:
                self.config.update(json.load(f))
        except FileNotFoundError:
            pass
        # 设置日志
        logging.basicConfig(
            filename='cpu_alert.log',
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )
        self.high_cpu_start_time = None
    def get_cpu_usage(self):
        """获取CPU使用率"""
        return psutil.cpu_percent(interval=1)
    def send_email_alert(self, cpu_usage):
        """发送邮件告警"""
        try:
            msg = MIMEText(f"CPU使用率告警\n当前CPU使用率: {cpu_usage}%\n阈值: {self.config['cpu_threshold']}%\n时间: {time.strftime('%Y-%m-%d %H:%M:%S')}")
            msg['Subject'] = 'CPU告警通知'
            msg['From'] = self.config['email']['sender']
            msg['To'] = ','.join(self.config['email']['recipients'])
            server = smtplib.SMTP(self.config['email']['smtp_server'], 
                                 self.config['email']['smtp_port'])
            server.starttls()
            server.login(self.config['email']['sender'], 
                        self.config['email']['password'])
            server.send_message(msg)
            server.quit()
            logging.info("邮件告警已发送")
        except Exception as e:
            logging.error(f"邮件发送失败: {e}")
    def check_and_alert(self):
        """检查CPU使用率并触发告警"""
        cpu_usage = self.get_cpu_usage()
        if cpu_usage > self.config['cpu_threshold']:
            if self.high_cpu_start_time is None:
                self.high_cpu_start_time = time.time()
            duration = time.time() - self.high_cpu_start_time
            if duration >= self.config['duration_threshold']:
                alert_msg = f"CPU使用率告警: {cpu_usage}% (持续{duration:.0f}秒)"
                logging.warning(alert_msg)
                if 'email' in self.config['alert_methods']:
                    self.send_email_alert(cpu_usage)
                if 'log' in self.config['alert_methods']:
                    print(alert_msg)
        else:
            self.high_cpu_start_time = None
    def run(self):
        """运行监控"""
        print(f"CPU监控启动，阈值: {self.config['cpu_threshold']}%")
        while True:
            self.check_and_alert()
            time.sleep(self.config['check_interval'])
if __name__ == "__main__":
    monitor = CPUAlertMonitor()
    try:
        monitor.run()
    except KeyboardInterrupt:
        print("\n监控已停止")

配置文件示例 (cpu_alert_config.json)

{
    "cpu_threshold": 80,
    "check_interval": 10,
    "duration_threshold": 60,
    "alert_methods": ["log", "email"],
    "email": {
        "smtp_server": "smtp.gmail.com",
        "smtp_port": 587,
        "sender": "monitor@gmail.com",
        "password": "your_app_password",
        "recipients": ["admin@company.com"]
    }
}

Windows PowerShell脚本

# CPU监控告警脚本
$threshold = 80
$checkInterval = 10
$alertLog = "C:\Logs\cpu_alert.log"
while ($true) {
    # 获取CPU使用率
    $cpuUsage = (Get-Counter "\Processor(_Total)\% Processor Time").CounterSamples.CookedValue
    if ($cpuUsage -gt $threshold) {
        $time = Get-Date -Format "yyyy-MM-dd HH:mm:ss"
        $alertMsg = "CPU告警: ${cpuUsage}% > ${threshold}% - ${time}"
        # 写入日志
        Add-Content -Path $alertLog -Value $alertMsg
        # 弹出告警窗口
        if ([System.Environment]::UserInteractive) {
            [System.Windows.Forms.MessageBox]::Show($alertMsg, "CPU告警", 
                [System.Windows.Forms.MessageBoxButtons]::OK, 
                [System.Windows.Forms.MessageBoxIcon]::Warning)
        }
        # 写入事件日志
        Write-EventLog -LogName Application -Source "CPU Monitor" `
            -EntryType Warning -EventId 1001 -Message $alertMsg
    }
    Start-Sleep -Seconds $checkInterval
}

Docker监控方案

# docker-compose.yml
version: '3.8'
services:
  cpu-monitor:
    image: python:3.9-slim
    volumes:
      - ./cpu_monitor.py:/app/cpu_monitor.py
      - ./config.json:/app/config.json
      - /var/run/docker.sock:/var/run/docker.sock
    command: python /app/cpu_monitor.py
    restart: unless-stopped
    environment:
      - TZ=Asia/Shanghai

扩展功能

1 多阈值告警

def get_alert_level(cpu_usage):
    """根据CPU使用率返回告警级别"""
    if cpu_usage > 95:
        return "CRITICAL"
    elif cpu_usage > 85:
        return "WARNING"
    elif cpu_usage > 75:
        return "INFO"
    return None

2 集成告警通道

# 企业微信通知
def send_wechat_alert(message):
    webhook_url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY"
    requests.post(webhook_url, json={"msgtype": "text", "text": {"content": message}})
# Slack通知
def send_slack_alert(message):
    webhook_url = "https://hooks.slack.com/services/YOUR/WEBHOOK/URL"
    requests.post(webhook_url, json={"text": message})

安装依赖

# Python依赖
pip install psutil
# Linux邮件服务
apt-get install mailutils  # Debian/Ubuntu
yum install mailx          # CentOS/RHEL

使用建议

1. 设置合理阈值：通常80%作为警告，95%作为严重告警
2. 加入持续检测：避免短时峰值造成误报
3. 记录告警历史：便于后续分析
4. 告警升级机制：长时间高负载自动升级告警级别
5. 集成运维平台：对接Zabbix、Prometheus等监控系统

这些脚本可以满足不同场景的CPU告警需求,你可以根据实际情况选择合适的方案。