本文目录导读：
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"
}
使用建议
- 生产环境：推荐使用 Python 方案，功能完整
- 轻量需求：Shell 脚本就够用
- 分布式监测：使用 Docker 容器化部署
- 监控平台：集成 Prometheus + Grafana
根据你的具体需求选择合适的方案！
