如何用脚本监控网络连通性

wen 实用脚本 2026-08-03 2

用脚本打造7×24小时智能连通性监控体系（附实战代码）

如何用脚本监控网络连通性

目录导读

为什么需要脚本监控？ —— 从“被动救火”到“主动预警”的转型
监控脚本的核心设计逻辑 —— 三层探测架构（ICMP/TCP/HTTP）
Python实战：一个可落地的监控脚本全解析 （含异常处理与日志轮转）
告警机制与数据可视化 —— 从邮件到Webhook的进阶之路
高频问题问答（FAQ） —— 解决你部署时的5个“坑”

为什么需要脚本监控？

根据思科《2023全球网络趋势报告》，61%的企业每月至少经历1次关键业务网络中断，而人工巡检平均需要27分钟才能发现故障，传统ping命令只能检测“通不通”，无法感知“通得如何”——丢包率、延迟抖动、DNS解析异常、HTTP状态码错误，这些才是影响用户体验的隐形杀手。

脚本监控的核心价值在于：用可重复的自动化逻辑替代人工轮询，将故障发现时间压缩到秒级，同时通过历史数据趋势分析预测链路劣化，实践中，一套优秀的连通性监控脚本应该做到：低资源占用、高频率探测、智能阈值判断、多渠道告警。

监控脚本的核心设计逻辑

任何连通性监控都逃不开“三明治模型”：

底层：传输层探测（ICMP ping / TCP端口连通）
中层：应用层探测（HTTP GET / DNS解析 / SSL证书有效期）
顶层：业务逻辑探测（模拟登录 / 关键API响应时间）

推荐组合策略：

ICMP Ping（每30秒）：适合首层快速故障定位，但需注意部分云服务商屏蔽ICMP（阿里云、AWS默认禁用）。
TCP Connect（每1分钟）：对80/443/22等端口发起握手，比ping更真实反映防火墙策略。
HTTP状态码+响应时间（每2分钟）：用curl -o /dev/null -s -w "%{http_code} %{time_total}" 检测Web服务真实健康度。

脚本必须内置三重容错：超时重试（3次）、退避算法（失败后间隔翻倍）、抖动抑制（连续3次异常才告警，防止误报）。

Python实战：一个可落地的监控脚本全解析

以下脚本已去除冗余逻辑,保持可读性，使用python3标准库subprocess + socket + smtplib，零第三方依赖，适用任何Linux/Windows环境。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import subprocess, json, time, smtplib
from email.mime.text import MIMEText
from datetime import datetime
import logging, os
# 配置区（采用JSON外部化，便于运维修改）
CONFIG = {
    "targets": [
        {"name": "公司官网", "host": "example.com", "port": 443, "mode": "http"},
        {"name": "核心ERP", "host": "10.10.1.8", "port": 8080, "mode": "tcp"},
        {"name": "云机房网关", "host": "10.20.30.1", "mode": "icmp"}
    ],
    "threshold": {"fail_times": 3, "interval": 30},  # 连续失败3次触发告警
    "mail": {"smtp": "smtp.qq.com", "port": 465, "user": "monitor@xx.com", "pwd": "authcode", "to": ["ops@xx.com"]},
    "log_file": "/var/log/netmon.log"
}
logging.basicConfig(filename=CONFIG["log_file"], level=logging.INFO,
                    format='%(asctime)s %(levelname)s %(message)s')
fail_cache = {}
def icmp_check(host):
    """执行系统ping，返回(丢包率%, 平均延迟ms)"""
    cmd = ["ping", "-n", "3", "-w", "2000", host] if os.name == "nt" else ["ping", "-c", "3", "-W", "2", host]
    res = subprocess.run(cmd, capture_output=True, text=True)
    try:
        # 解析不同系统输出格式（此处简化）
        latency = float(res.stdout.split("time=")[-1].split("ms")[0])
        return 0, latency
    except:
        return 100, -1
def tcp_check(host, port, timeout=3):
    import socket
    s = socket.socket(); s.settimeout(timeout)
    try:
        start = time.time(); s.connect((host, port)); delay = (time.time() - start)*1000
        return 0, round(delay, 1)
    except Exception as e:
        return 100, -1
    finally:
        s.close()
def http_check(host, port):
    import urllib.request
    url = f"https://{host}:{port}/" if port == 443 else f"http://{host}:{port}/"
    try:
        req = urllib.request.Request(url, headers={"User-Agent": "NetMon"})
        start = time.time(); resp = urllib.request.urlopen(req, timeout=5); delay = (time.time()-start)*1000
        return (0 if resp.status == 200 else 50), round(delay, 1)  # 非200码视为丢包率50%
    except:
        return 100, -1
def send_alert(hostname, detail):
    msg = MIMEText(f"[{datetime.now()}] 目标 {hostname} 连续失败，详情: {detail}", 'plain', 'utf-8')
    msg["Subject"] = f"【网络告警】{hostname} 不可达"
    msg["From"] = CONFIG["mail"]["user"]; msg["To"] = ",".join(CONFIG["mail"]["to"])
    try:
        s = smtplib.SMTP_SSL(CONFIG["mail"]["smtp"], CONFIG["mail"]["port"]); s.login(CONFIG["mail"]["user"], CONFIG["mail"]["pwd"]); s.send_message(msg); s.quit()
        logging.warning(f"告警已发送至 {CONFIG['mail']['to']}")
    except Exception as e:
        logging.error(f"邮件发送失败: {e}")
def main():
    while True:
        for t in CONFIG["targets"]:
            name, host = t["name"], t["host"]
            if t.get("mode") == "icmp": loss, lat = icmp_check(host)
            elif t.get("mode") == "tcp": loss, lat = tcp_check(host, t["port"])
            elif t.get("mode") == "http": loss, lat = http_check(host, t["port"])
            # 记录失败次数
            cache = fail_cache.setdefault(name, 0)
            if loss > 0:
                fail_cache[name] = cache + 1
                logging.warning(f"{name} 探测失败，丢包率{loss}%，延迟{lat}ms，连续失败{cache+1}次")
                if fail_cache[name] >= CONFIG["threshold"]["fail_times"]:
                    send_alert(name, f"丢包率={loss}%, 延迟={lat}ms")
                    fail_cache[name] = 0  # 触发后重置
            else:
                fail_cache[name] = 0
                logging.info(f"{name} 正常，延迟 {lat}ms")
        time.sleep(CONFIG["threshold"]["interval"])
if __name__ == "__main__":
    main()

关键设计亮点：

使用subprocess.run调用系统ping命令，避免加载第三方库，脚本体积小于5KB。
失败计数器采用内存缓存,避免告警风暴，但重启后丢失——生产环境建议改用SQLite或Redis。
邮件发送失败不影响主循环,日志完整记录每次探测结果。

告警机制与数据可视化

邮件只是起点,现代运维更推荐分级告警：

一级（短信/电话）：Ping 100%丢包持续3分钟，触发微服务调用链断裂风险。
二级（企业微信/钉钉Webhook）：HTTP响应时间>2秒，发送JSON到群机器人。
三级（邮箱）：SSL证书剩余<30天，仅记录通知。

可视化方面,可让脚本将每次结果追加到InfluxDB，再用Grafana展示延迟折线图+可用性热力图，示例Influx写入代码（仅示意关键行）：

from influxdb_client import InfluxDBClient, Point
client = InfluxDBClient(url="http://localhost:8086", token="mytoken", org="ops")
write_api = client.write_api(write_options=SYNCHRONOUS)
p = Point("netmon").tag("host", name).field("latency", lat).field("loss", loss)
write_api.write(bucket="network", record=p)

高频问题问答（FAQ）

Q1：脚本一直被防火墙拦截，如何确保自己的监控不被误伤？ A：在目标服务器上添加白名单IP，且监控机尽量绑定内网固定IP，另外所有出站探测请显式声明User-Agent: NetworkMonitor/1.0，便于对方过滤。

Q2：监控脚本本身挂了怎么办？ A：必须配合crontab守护，每5分钟检查进程是否存在；或者在Docker/K8s中运行并设置自动重启策略，推荐终极方案：用systemd管理，Restart=always。

Q3：需要监控海量IP（比如5000台），性能瓶颈在哪？ A：用多线程（concurrent.futures.ThreadPoolExecutor）并发探测，2000线程同时发起ICMP请求完全可行，瓶颈在网卡和DNS解析器，建议使用异步库asyncio或Scapy发送原始包。

Q4：如何避免业务高峰时误报？ A：引入“动态阈值”——利用time series预测模型，根据过去7天同时段的延迟基线设置上下浮动20%的告警线，这需要接入历史数据库。

Q5：脚本监控和商业软件（如Zabbix）如何取舍？ A：脚本优势是轻量（无需Agent）、灵活（改一行代码即适配新场景）；商业软件强在自带告警规则库和可视化面板，建议混合：脚本负责边缘节点特殊探测，Zabbix负责核心机房全量监控。

监控的本质不是“技术炫技”，而是将不确定性转化为可预测性，上述脚本已足够支撑百台规模网络，但真正的网络韧性还需配合重拨机制（如FRR联动）、冗余链路检测（BGP会话状态）等等，放下鼠标，让脚本替你值守——这才是运维工程师应有的姿态。

（全文完）