如何写一个脚本自动轮询并告警

wen 实用脚本 2026-07-26 1

从零到实战的完整指南

目录导读

背景与意义：为什么需要自动轮询告警？
核心概念：轮询 vs 事件驱动，你该选哪种？
技术选型：脚本语言、工具与框架推荐
实战步骤：手写一个可运行的轮询告警脚本
进阶技巧：错误处理、去重、通知渠道集成
常见问答：轮询频率怎么设？脚本挂了怎么办？
SEO优化：关键词、结构化数据与内容策略

背景与意义：为什么需要自动轮询告警？

在运维、监控、数据采集等场景中，“自动轮询并告警”是最基础也最实用的自动化需求，假设你管理一个电商网站，需要每5分钟检查一次服务器是否在线；或者你是数据分析师，需要定时抓取外部API数据，并在数据异常时发送邮件——这些任务都可以通过一个简单的轮询脚本实现。

如何写一个脚本自动轮询并告警

核心价值：

减少人工盯屏：从“人找故障”变为“故障找人”
提升响应速度：多数轮询脚本可在秒级完成检测并触发告警
成本低廉：相比商业监控系统（如Nagios、Zabbix），自建脚本零成本，且可高度定制

但要注意,轮询不等于实时，如果需要毫秒级响应，应选择Webhook或事件驱动架构，本文将聚焦于“定时轮询+条件告警”这一通用模式。

核心概念：轮询 vs 事件驱动，你该选哪种？

1 轮询（Polling）

定义：脚本按固定时间间隔主动检查目标状态
优点：实现简单，兼容性强（任何系统都能被“轮”）
缺点：延迟取决于间隔时长；可能造成资源浪费（即使目标未变化也持续检查）

2 事件驱动（Event-Driven）

定义：目标主动推送状态变化（如Webhook回调）
优点：实时性强，资源利用率高
缺点：需要目标端支持推送机制，实现成本较高

选择建议：

如果你监控的是HTTP服务、数据库、文件变化等标准协议，轮询完全够用
如果你需要监控消息队列、物联网设备等支持回调的系统，优先用事件驱动

技术选型：脚本语言、工具与框架推荐

场景	推荐方案	理由
简单HTTP检测	Bash + curl	零依赖，适合Linux服务器
复杂逻辑处理	Python + requests + schedule	库丰富，可读性好
企业级监控	Prometheus + Alertmanager	自带轮询与告警生态
无代码方案	UptimeRobot / Pingdom	无需写脚本，但灵活性差

本文实战选择：Python 3（兼容性好，示例清晰） + requests（HTTP轮询）+ schedule（定时调度）+ smtplib（邮件告警）

实战步骤：手写一个可运行的轮询告警脚本

1 需求描述

每60秒轮询一次 https://api.example.com/health
如果返回状态码非200,或响应时间>5秒，则发送告警邮件
同一故障连续告警时,避免重复发送（去重逻辑）

2 代码实现

import requests
import schedule
import time
import smtplib
from email.mime.text import MIMEText
# 配置项
TARGET_URL = "https://api.example.com/health"
CHECK_INTERVAL = 60  # 秒
ALERT_EMAIL = "admin@example.com"
SMTP_SERVER = "smtp.example.com"
SMTP_PORT = 587
SMTP_USER = "alert@example.com"
SMTP_PASS = "your-password"
# 状态变量（用于去重）
last_alert_time = 0
last_alert_reason = ""
def send_email(subject, body):
    """发送告警邮件"""
    msg = MIMEText(body)
    msg["Subject"] = subject
    msg["From"] = SMTP_USER
    msg["To"] = ALERT_EMAIL
    with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
        server.starttls()
        server.login(SMTP_USER, SMTP_PASS)
        server.send_message(msg)
def check_health():
    """执行一次轮询检测"""
    global last_alert_time, last_alert_reason
    try:
        start = time.time()
        resp = requests.get(TARGET_URL, timeout=5)
        elapsed = time.time() - start
        if resp.status_code != 200:
            reason = f"状态码异常: {resp.status_code}"
        elif elapsed > 5:
            reason = f"响应超时: {elapsed:.2f}s"
        else:
            # 正常，清除告警状态
            last_alert_time = 0
            last_alert_reason = ""
            print(f"[OK] {time.ctime()} - 状态正常")
            return
        # 去重：相同原因且距离上次告警不足300秒，不重复发送
        curr_time = time.time()
        if reason == last_alert_reason and (curr_time - last_alert_time) < 300:
            print(f"[跳过] 重复告警: {reason}")
            return
        # 发送告警
        send_email(
            subject=f"[告警] {TARGET_URL} 异常",
            body=f"时间: {time.ctime()}\n原因: {reason}\nURL: {TARGET_URL}"
        )
        print(f"[告警] {time.ctime()} - {reason}")
        # 更新去重记录
        last_alert_time = curr_time
        last_alert_reason = reason
    except requests.exceptions.RequestException as e:
        # 网络错误直接告警
        reason = f"网络错误: {str(e)}"
        curr_time = time.time()
        if reason == last_alert_reason and (curr_time - last_alert_time) < 300:
            print(f"[跳过] 重复告警: {reason}")
            return
        send_email(subject="[告警] 网络连接失败", body=f"时间: {time.ctime()}\n错误: {str(e)}")
        last_alert_time = curr_time
        last_alert_reason = reason
# 主循环
if __name__ == "__main__":
    schedule.every(CHECK_INTERVAL).seconds.do(check_health)
    print(f"轮询已启动，间隔 {CHECK_INTERVAL} 秒")
    while True:
        schedule.run_pending()
        time.sleep(1)

3 运行与优化

部署：使用 nohup python3 monitor.py & 后台运行，或配置为systemd服务
日志：建议将脚本输出重定向到文件 >> monitor.log 2>&1
安全：SMTP密码不应硬编码，建议使用环境变量或密钥管理服务

进阶技巧：错误处理、去重、通知渠道集成

1 去重机制

上述代码已实现“相同原因在5分钟内只告警一次”，避免告警风暴，更高级的方案：使用Redis记录告警状态，实现分布式去重。

2 多通知渠道

邮件：smtplib（示例已实现）
短信：Twilio API
Slack/企业微信：Webhook机器人
钉钉：自定义机器人加签

3 健康自检

脚本本身也可能挂掉,建议使用supervisor或crontab每分钟检查脚本进程是否存在，若不存在则自动重启。

# crontab 示例（每分钟检查）
* * * * * pgrep -f monitor.py || python3 /path/to/monitor.py

4 轮询频率优化

静态目标：间隔可设为5-15分钟（如服务器存活）
动态目标：间隔可设为1-5分钟（如交易数据）
遵循“不要超过目标自身状态变化速度的2倍”原则

常见问答

Q1：轮询脚本用什么语言写最好？
A：Python最适合非运维场景（库全、语法简单）；Bash适合Linux服务器快速验证；Go适合高并发轮询（如同时监控1000+端点），初学者建议从Python入手。

Q2：轮询间隔设置多少秒合适？
A：取决于业务容忍的延迟，监控支付网关，支付失败最好在30秒内发现，则轮询间隔设为15秒，但注意，间隔越短，服务器压力越大。一般服务建议60秒，高可用场景建议30秒。

Q3：如果脚本运行几天后内存泄漏怎么办？
A：在循环体内避免全局变量累积（示例代码已注意）；使用 requests.Session 复用连接；设置系统定时重启，例如每天凌晨4点由crontab杀死并重启脚本。

Q4：轮询和长连接探测有什么区别？
A：轮询是“主动定时查询”，长连接（如WebSocket）是“保持通道，被动接收”，轮询适合标准协议，长连接适合需要实时推送的场景。不建议用长连接实现轮询，因为会大幅增加连接数。

Q5：脚本需要监控大量目标时，如何优化？
A：使用异步库（如 aiohttp）实现并发轮询；将目标列表存储在数据库或配置文件中；增加超时控制，避免单个目标阻塞整个轮询周期。

SEO优化：关键词、结构化数据与内容策略

1 自然关键词布局

核心词：自动轮询脚本、告警脚本、Python监控脚本、健康检查脚本
长尾词：如何写自动轮询并发送邮件、轮询告警去重机制、Python schedule定时任务示例首段、问答中自然融入，避免堆砌

2 结构化数据建议

使用 FAQ Schema 标记问答部分（Q1-Q5），帮助Google直接展示答案
使用 HowTo Schema 标记实战步骤（4.1-4.3），增加代码片段曝光
在文章底部添加 schema.org/Article 标记

3 内容策略要点

代码可复制：示例代码去掉行号，简化注释，让读者直接运行
对比结构：轮询vs事件驱动、Python vs Bash等，满足用户决策需求
：不要只写“轮询告警脚本”，应写“如何用Python写一个自动轮询健康检查并邮件告警的脚本”

附：扩展阅读与工具推荐

Prometheus + Alertmanager：企业级轮询+告警方案，自带时间序列数据库
Healthchecks.io：反向轮询服务（脚本向它发送心跳，心跳停止则告警）
Runscope：API自动轮询监控，支持告警到Slack

实践建议：先运行本文示例脚本监控一个本地的测试URL，确认邮件能正常发送后，再扩展监控线上服务，从简单到复杂，逐步迭代。

本文已综合Stack Overflow、Real Python、Google Cloud监控文档等来源进行去重与重组，确保内容原创且符合SEO最佳实践。