脚本如何实时监测网站连通性

本文目录导读：

1. Python 方案（推荐）
2. Shell 脚本方案
3. Node.js 方案
4. Docker 化部署
5. 高级特性实现
6. 使用建议

Python 方案（推荐）

基础监测脚本

import requests
import time
import logging
from datetime import datetime
# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('website_monitor.log'),
        logging.StreamHandler()
    ]
)
class WebsiteMonitor:
    def __init__(self, url, interval=60, timeout=10):
        self.url = url
        self.interval = interval  # 监测间隔（秒）
        self.timeout = timeout    # 超时时间（秒）
    def check_website(self):
        """检查网站连通性"""
        try:
            start_time = time.time()
            response = requests.get(
                self.url, 
                timeout=self.timeout,
                headers={'User-Agent': 'Mozilla/5.0'}
            )
            response_time = time.time() - start_time
            if response.status_code == 200:
                return True, response_time
            else:
                return False, response.status_code
        except requests.exceptions.Timeout:
            return False, "超时"
        except requests.exceptions.ConnectionError:
            return False, "连接失败"
        except Exception as e:
            return False, str(e)
    def monitor_loop(self):
        """主监测循环"""
        print(f"开始监测: {self.url}")
        print(f"监测间隔: {self.interval}秒")
        while True:
            status, result = self.check_website()
            if status:
                logging.info(f"✅ {self.url} - 正常 (响应时间: {result:.2f}s)")
            else:
                logging.warning(f"❌ {self.url} - 异常: {result}")
            time.sleep(self.interval)
# 使用示例
if __name__ == "__main__":
    monitor = WebsiteMonitor(
        url="https://www.example.com",
        interval=30,  # 每30秒检查一次
        timeout=5     # 5秒超时
    )
    monitor.monitor_loop()

多网站并行监测

import asyncio
import aiohttp
import time
class AsyncMonitor:
    def __init__(self):
        self.websites = []
    def add_website(self, url, interval=30):
        self.websites.append({'url': url, 'interval': interval})
    async def check_single(self, session, website):
        """异步检查单个网站"""
        try:
            start = time.time()
            async with session.get(website['url']) as response:
                response_time = time.time() - start
                status = "正常" if response.status == 200 else f"状态码: {response.status}"
                print(f"{website['url']} - {status} ({response_time:.2f}s)")
                return True
        except Exception as e:
            print(f"{website['url']} - 异常: {e}")
            return False
    async def monitor_loop(self):
        """异步监测循环"""
        async with aiohttp.ClientSession() as session:
            while True:
                tasks = []
                for website in self.websites:
                    task = asyncio.create_task(self.check_single(session, website))
                    tasks.append(task)
                await asyncio.gather(*tasks)
                await asyncio.sleep(10)  # 整体间隔
# 使用示例
async def main():
    monitor = AsyncMonitor()
    monitor.add_website("https://www.google.com")
    monitor.add_website("https://www.github.com")
    monitor.add_website("https://www.baidu.com")
    await monitor.monitor_loop()
# asyncio.run(main())

Shell 脚本方案

基础监测脚本

#!/bin/bash
# 监测配置
URL="https://www.example.com"
INTERVAL=30  # 监测间隔（秒）
TIMEOUT=10   # 超时时间（秒）
LOG_FILE="website_monitor.log"
# 监测函数
check_website() {
    local start_time=$(date +%s%N)
    # 使用 curl 检查
    if curl -s -o /dev/null -w "%{http_code}" \
             --connect-timeout $TIMEOUT \
             --max-time $TIMEOUT \
             "$URL" | grep -q "200"; then
        local end_time=$(date +%s%N)
        local response_time=$(( (end_time - start_time) / 1000000 ))  # 毫秒
        echo "$(date '+%Y-%m-%d %H:%M:%S') ✅ $URL 正常 (${response_time}ms)"
        return 0
    else
        echo "$(date '+%Y-%m-%d %H:%M:%S') ❌ $URL 异常"
        return 1
    fi
}
# 主循环
echo "开始监测: $URL"
echo "监测间隔: ${INTERVAL}秒"
while true; do
    result=$(check_website)
    echo "$result" | tee -a "$LOG_FILE"
    sleep $INTERVAL
done

Node.js 方案

const https = require('https');
const http = require('http');
class WebsiteMonitor {
    constructor(url, interval = 60) {
        this.url = url;
        this.interval = interval * 1000;
        this.protocol = url.startsWith('https') ? https : http;
    }
    checkWebsite() {
        return new Promise((resolve) => {
            const startTime = Date.now();
            const req = this.protocol.get(this.url, (res) => {
                const responseTime = Date.now() - startTime;
                if (res.statusCode === 200) {
                    resolve({ status: true, responseTime });
                } else {
                    resolve({ status: false, code: res.statusCode });
                }
            });
            req.on('error', (err) => {
                resolve({ status: false, error: err.message });
            });
            req.setTimeout(10000, () => {
                req.destroy();
                resolve({ status: false, error: 'Timeout' });
            });
        });
    }
    async start() {
        console.log(`开始监测: ${this.url}`);
        while (true) {
            const result = await this.checkWebsite();
            if (result.status) {
                console.log(`✅ ${this.url} - 正常 (${result.responseTime}ms)`);
            } else {
                console.log(`❌ ${this.url} - 异常: ${result.error || result.code}`);
            }
            await new Promise(resolve => setTimeout(resolve, this.interval));
        }
    }
}
// 使用示例
const monitor = new WebsiteMonitor('https://www.example.com', 30);
monitor.start();

Docker 化部署

Dockerfile

FROM python:3.9-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY monitor.py .
CMD ["python", "monitor.py"]

docker-compose.yml

version: '3'
services:
  monitor:
    build: .
    environment:
      - TARGET_URL=https://www.example.com
      - INTERVAL=30
      - TIMEOUT=5
    restart: always
    volumes:
      - ./logs:/app/logs

高级特性实现

告警功能

import smtplib
from email.mime.text import MIMEText
class AlertManager:
    def __init__(self):
        self.alert_threshold = 3  # 连续失败次数
        self.fail_count = 0
        self.last_alert_time = 0
        self.alert_interval = 300  # 告警间隔（秒）
    def check_alert(self, status):
        if not status:
            self.fail_count += 1
        else:
            self.fail_count = 0
        if self.fail_count >= self.alert_threshold:
            current_time = time.time()
            if current_time - self.last_alert_time > self.alert_interval:
                self.send_alert()
                self.last_alert_time = current_time
    def send_alert(self):
        # 实现邮件、短信、微信等告警
        pass

数据统计

import json
from collections import deque
class Statistics:
    def __init__(self, window_size=100):
        self.results = deque(maxlen=window_size)
        self.total_checks = 0
        self.success_count = 0
    def add_result(self, status, response_time):
        self.results.append({
            'timestamp': time.time(),
            'status': status,
            'response_time': response_time
        })
        self.total_checks += 1
        if status:
            self.success_count += 1
    def get_statistics(self):
        uptime = (self.success_count / self.total_checks * 100) if self.total_checks > 0 else 0
        response_times = [r['response_time'] for r in self.results if r['status']]
        avg_response = sum(response_times) / len(response_times) if response_times else 0
        return {
            'uptime': f"{uptime:.2f}%",
            'total_checks': self.total_checks,
            'avg_response_time': f"{avg_response:.2f}s"
        }

使用建议

1. 生产环境：推荐使用 Python 方案，功能完整
2. 轻量需求：Shell 脚本就够用
3. 分布式监测：使用 Docker 容器化部署
4. 监控平台：集成 Prometheus + Grafana

根据你的具体需求选择合适的方案！