如何写定时检测站点连通性

本文目录导读：

1. 最简单的Shell脚本方案
2. Python脚本方案（功能更强大）
3. Docker部署方案
4. 专业监控方案（仅监控逻辑）
5. 最佳实践建议
6. 选择建议

我来介绍几种写定时检测站点连通性的方法,从简单到复杂：

最简单的Shell脚本方案

Linux/Mac (使用cron)

#!/bin/bash
# check_site.sh
SITE_URL="https://example.com"
LOG_FILE="/var/log/site_check.log"
# 检测网站连通性
if curl -s -o /dev/null -w "%{http_code}" --connect-timeout 5 $SITE_URL | grep -q "200"; then
    echo "$(date): $SITE_URL 正常" >> $LOG_FILE
else
    echo "$(date): $SITE_URL 异常" >> $LOG_FILE
fi

设置cron定时任务：

# 每5分钟运行一次
*/5 * * * * /path/to/check_site.sh

Windows (使用任务计划程序)

@echo off
REM check_site.bat
set SITE_URL=https://example.com
set LOG_FILE=C:\logs\site_check.log
curl -s -o NUL -w "%%{http_code}" --connect-timeout 5 %SITE_URL% | find "200" > NUL
if %errorlevel% equ 0 (
    echo %date% %time%: %SITE_URL% 正常 >> %LOG_FILE%
) else (
    echo %date% %time%: %SITE_URL% 异常 >> %LOG_FILE%
)

Python脚本方案（功能更强大）

基础检测脚本

import requests
import time
import logging
from datetime import datetime
class SiteChecker:
    def __init__(self, url, check_interval=300):
        self.url = url
        self.check_interval = check_interval
        self.setup_logging()
    def setup_logging(self):
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('site_check.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger(__name__)
    def check_site(self):
        """检测站点连通性"""
        try:
            start_time = time.time()
            response = requests.get(
                self.url, 
                timeout=10,
                headers={'User-Agent': 'Mozilla/5.0'}
            )
            response_time = time.time() - start_time
            if response.status_code == 200:
                self.logger.info(f"✅ {self.url} 正常 - 响应时间: {response_time:.2f}s")
                return True
            else:
                self.logger.warning(f"⚠️ {self.url} 返回状态码: {response.status_code}")
                return False
        except requests.exceptions.Timeout:
            self.logger.error(f"❌ {self.url} 超时")
            return False
        except requests.exceptions.ConnectionError:
            self.logger.error(f"❌ {self.url} 连接失败")
            return False
        except Exception as e:
            self.logger.error(f"❌ {self.url} 错误: {str(e)}")
            return False
    def run_continuous(self):
        """连续运行检测"""
        self.logger.info(f"开始监控 {self.url}")
        while True:
            self.check_site()
            time.sleep(self.check_interval)
# 使用示例
if __name__ == "__main__":
    checker = SiteChecker("https://example.com", check_interval=60)
    checker.run_continuous()

进阶版：多站点检测+通知

import requests
import time
import json
from datetime import datetime
import smtplib
from email.mime.text import MIMEText
from typing import List, Dict
class AdvancedSiteChecker:
    def __init__(self, config_file='sites.json'):
        self.sites = self.load_config(config_file)
        self.failed_count = {}
    def load_config(self, config_file):
        """加载站点配置"""
        with open(config_file, 'r') as f:
            return json.load(f)
    def load_sites(self):
        """从配置文件加载站点列表"""
        # sites.json 格式:
        # {
        #     "sites": [
        #         {"url": "https://example1.com", "name": "站点1", "interval": 300},
        #         {"url": "https://example2.com", "name": "站点2", "interval": 600}
        #     ],
        #     "notifications": {
        #         "email": {
        #             "smtp_server": "smtp.gmail.com",
        #             "smtp_port": 587,
        #             "sender": "your@email.com",
        #             "password": "your_password",
        #             "recipients": ["admin@email.com"]
        #         }
        #     }
        # }
        ...
    def check_site(self, site: Dict) -> Dict:
        """检测单个站点"""
        result = {
            'url': site['url'],
            'name': site['name'],
            'timestamp': datetime.now().isoformat(),
            'status': 'unknown',
            'response_time': 0,
            'error': None
        }
        try:
            start = time.time()
            resp = requests.get(
                site['url'],
                timeout=10,
                verify=True
            )
            result['response_time'] = round(time.time() - start, 2)
            if resp.status_code == 200:
                result['status'] = 'up'
            else:
                result['status'] = 'down'
                result['error'] = f"HTTP {resp.status_code}"
        except Exception as e:
            result['status'] = 'down'
            result['error'] = str(e)
        return result
    def send_alert(self, site: Dict, result: Dict):
        """发送告警通知（示例为邮件）"""
        # 实际实现邮件或短信通知
        print(f"⚠️ Alert: {site['name']} is down! Error: {result['error']}")
    def run(self):
        """主运行循环"""
        while True:
            for site in self.sites['sites']:
                result = self.check_site(site)
                self.log_result(result)
                if result['status'] == 'down':
                    # 连续失败3次才发送告警，避免误报
                    if site['url'] not in self.failed_count:
                        self.failed_count[site['url']] = 0
                    self.failed_count[site['url']] += 1
                    if self.failed_count[site['url']] >= 3:
                        self.send_alert(site, result)
                else:
                    self.failed_count[site['url']] = 0
            time.sleep(min([s.get('interval', 300) for s in self.sites['sites']]))
# 简化版运行
if __name__ == "__main__":
    checker = AdvancedSiteChecker()
    checker.run()

Docker部署方案

Dockerfile

FROM python:3.9-slim
WORKDIR /app
# 安装依赖
RUN pip install requests schedule
# 复制脚本
COPY site_checker.py .
COPY sites.json .
# 运行检测
CMD ["python", "site_checker.py"]

docker-compose.yml

version: '3.8'
services:
  site-checker:
    build: .
    container_name: site-monitor
    restart: always
    volumes:
      - ./logs:/app/logs
    environment:
      - CHECK_INTERVAL=300
      - NOTIFICATION_EMAIL=admin@example.com

专业监控方案（仅监控逻辑）

import asyncio
import aiohttp
import time
from dataclasses import dataclass
from typing import Callable
@dataclass
class CheckResult:
    url: str
    status: str
    response_time: float
    timestamp: float
    error: str = None
class AsyncSiteChecker:
    def __init__(self, sites: list, on_check: Callable = None):
        self.sites = sites
        self.on_check = on_check or print
    async def check_single(self, session: aiohttp.ClientSession, url: str) -> CheckResult:
        """异步检测单个站点"""
        start = time.time()
        try:
            async with session.get(url, timeout=10) as response:
                result = CheckResult(
                    url=url,
                    status='up' if response.status == 200 else 'down',
                    response_time=time.time() - start,
                    timestamp=time.time()
                )
                if response.status != 200:
                    result.error = f"HTTP {response.status}"
        except Exception as e:
            result = CheckResult(
                url=url,
                status='down',
                response_time=time.time() - start,
                timestamp=time.time(),
                error=str(e)
            )
        return result
    async def check_all(self):
        """并发检测所有站点"""
        async with aiohttp.ClientSession() as session:
            tasks = [self.check_single(session, url) for url in self.sites]
            results = await asyncio.gather(*tasks)
            for result in results:
                self.on_check(result)
    async def run(self, interval: int = 300):
        """持续运行"""
        while True:
            await self.check_all()
            await asyncio.sleep(interval)
# 使用示例
async def main():
    sites = [
        "https://example.com",
        "https://google.com",
        "https://github.com"
    ]
    def handle_result(result: CheckResult):
        icon = "✅" if result.status == 'up' else "❌"
        print(f"{icon} {result.url} - {result.response_time:.2f}s - {result.error or 'OK'}")
    checker = AsyncSiteChecker(sites, on_check=handle_result)
    await checker.run(interval=60)
if __name__ == "__main__":
    asyncio.run(main())

最佳实践建议

配置文件示例 (sites.json)

{
  "sites": [
    {
      "url": "https://api.example.com/health",
      "name": "API服务",
      "interval": 30,
      "timeout": 5,
      "expected_status": 200,
      "keywords": ["healthy", "ok"]
    }
  ],
  "notifications": {
    "email": {
      "enabled": false,
      "smtp_server": "smtp.gmail.com",
      "recipients": ["admin@example.com"]
    },
    "webhook": {
      "enabled": true,
      "url": "https://hooks.slack.com/...",
      "method": "POST"
    }
  },
  "general": {
    "log_level": "INFO",
    "max_retries": 3,
    "alert_after_failures": 3
  }
}

选择建议

• 简单需求：使用Shell脚本 + cron
• 中等需求：Python脚本，支持多站点和多通知方式
• 复杂需求：使用成熟的监控工具如Prometheus + Grafana
• 微服务架构：使用专门的健康检查框架

推荐从简单的Shell脚本开始,根据需求逐步升级到更复杂的方案，最重要的是定期检查日志并根据需要调整告警策略。