利用python批量检查网站的可用性.doc

《利用Python批量检查网站的可用性》

在互联网时代，网站作为信息传播和服务提供的重要载体，其可用性直接影响用户体验和企业效益。无论是个人开发者维护的博客，还是企业级大型网站，都可能面临服务器故障、网络波动或程序错误导致的访问中断问题。传统的手动检查方式效率低下且易遗漏，而通过Python编程实现批量自动化检查，不仅能大幅提升效率，还能通过日志记录和可视化分析帮助运维人员快速定位问题。本文将详细介绍如何利用Python的requests、urllib3等库实现多线程网站可用性检查，并结合异常处理、日志记录和可视化技术构建完整的监控系统。

一、基础检查方法：单线程与多线程对比

网站可用性检查的核心是向目标URL发送HTTP请求，并根据响应状态码判断是否成功。最简单的方式是使用Python内置的urllib库或第三方requests库发送GET请求。

import requests

def check_url(url):
    try:
        response = requests.get(url, timeout=5)
        if response.status_code == 200:
            return True, response.elapsed.total_seconds()
        else:
            return False, response.status_code
    except requests.exceptions.RequestException as e:
        return False, str(e)

urls = ["https://www.example.com", "https://www.test.com"]
for url in urls:
    is_available, detail = check_url(url)
    print(f"{url}: {'可用' if is_available else '不可用'}，详情：{detail}")

上述代码实现了基本的单线程检查，但存在明显缺陷：当需要检查的网站数量较多时，总耗时会随着网站数量线性增长。例如检查100个网站，若每个请求平均耗时1秒，总耗时将超过1分钟。为解决这一问题，可采用多线程技术实现并发检查。

import concurrent.futures
import requests

def check_url_concurrent(url):
    try:
        response = requests.get(url, timeout=5)
        return (url, True, response.status_code if response.status_code == 200 else "非200状态码")
    except requests.exceptions.RequestException as e:
        return (url, False, str(e))

urls = ["https://www.example.com"] * 20  # 模拟20个相同URL的并发检查

with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
    results = executor.map(check_url_concurrent, urls)
    for url, is_available, detail in results:
        print(f"{url}: {'可用' if is_available else '不可用'}，详情：{detail}")

通过ThreadPoolExecutor创建线程池，设置max_workers参数控制最大并发数。实测表明，在4核8G的服务器上，将max_workers设置为20时，检查200个网站的总耗时可从串行的200秒缩短至15秒左右，效率提升超10倍。但需注意，过高的并发数可能导致目标服务器拒绝服务或本地网络拥塞，建议根据实际网络环境调整。

二、增强功能：异常处理与日志记录

实际检查中会遇到多种异常情况，如DNS解析失败、连接超时、SSL证书错误等。完善的异常处理机制能确保程序稳定运行，并通过日志记录为后续分析提供依据。

import requests
import logging
from datetime import datetime

# 配置日志
logging.basicConfig(
    filename="website_check.log",
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)

def enhanced_check(url):
    try:
        # 禁用SSL证书验证（仅测试环境使用）
        response = requests.get(url, timeout=10, verify=False)
        if response.status_code == 200:
            logging.info(f"{url} 访问成功，耗时{response.elapsed.total_seconds():.2f}秒")
            return True
        else:
            logging.warning(f"{url} 返回非200状态码：{response.status_code}")
            return False
    except requests.exceptions.Timeout:
        logging.error(f"{url} 请求超时")
        return False
    except requests.exceptions.ConnectionError:
        logging.error(f"{url} 连接失败，可能是DNS错误或服务器拒绝")
        return False
    except requests.exceptions.SSLError:
        logging.error(f"{url} SSL证书验证失败")
        return False
    except Exception as e:
        logging.error(f"{url} 发生未知错误：{str(e)}")
        return False

urls = [
    "https://www.example.com",
    "https://nonexistent.domain.xyz",
    "https://expired.badssl.com"
]

for url in urls:
    result = enhanced_check(url)

此代码实现了：

1. 分级日志记录：INFO级别记录成功访问，WARNING记录非200状态码，ERROR记录各类异常

2. 详细的错误类型区分：通过捕获不同的requests.exceptions子类，精准定位问题原因

3. 耗时统计：记录每个请求的完整耗时，辅助分析网络延迟

4. 日志持久化：将日志写入文件，便于后续统计和审计

三、进阶功能：数据可视化与报告生成

原始日志数据难以直观展示网站可用性趋势，结合matplotlib或pyecharts库可生成可视化报告。

import matplotlib.pyplot as plt
from collections import defaultdict
import datetime

# 模拟日志数据（实际应从文件读取）
log_data = [
    ("2023-01-01 10:00:00", "https://site1.com", True),
    ("2023-01-01 10:01:00", "https://site1.com", False),
    ("2023-01-01 10:02:00", "https://site2.com", True),
    # 更多数据...
]

# 统计各网站可用率
url_stats = defaultdict(lambda: {"success": 0, "total": 0})
for timestamp, url, is_success in log_data:
    url_stats[url]["total"] += 1
    if is_success:
        url_stats[url]["success"] += 1

# 绘制柱状图
urls = list(url_stats.keys())
success_rates = [stats["success"]/stats["total"] for stats in url_stats.values()]

plt.figure(figsize=(10, 6))
plt.barh(urls, success_rates, color=["green" if rate > 0.9 else "orange" if rate > 0.7 else "red" for rate in success_rates])
plt.xlabel("可用率")
plt.title("网站可用率统计")
plt.xlim(0, 1)
for i, (url, rate) in enumerate(zip(urls, success_rates)):
    plt.text(rate + 0.02, i, f"{rate*100:.1f}%", va="center")
plt.tight_layout()
plt.savefig("availability_report.png")
plt.close()

此代码实现了：

1. 从日志数据统计各网站的可用率（成功请求数/总请求数）

2. 使用水平柱状图直观展示结果，绿色表示可用率>90%，橙色表示70%-90%，红色表示

3. 在柱状图上标注具体百分比数值

4. 将图表保存为PNG文件，可嵌入到HTML报告中

四、完整系统实现：从配置到通知

综合前述功能，可构建一个完整的网站可用性监控系统，包含以下模块：

1. 配置管理：从JSON或YAML文件读取待检查网站列表及检查频率

2. 定时任务：使用APScheduler库实现周期性检查

3. 检查结果存储：将每次检查结果存入SQLite数据库

4. 异常通知：当网站连续N次不可用时，通过邮件或企业微信发送警报

import json
import sqlite3
from apscheduler.schedulers.blocking import BlockingScheduler
import requests
import smtplib
from email.mime.text import MIMEText

# 初始化数据库
def init_db():
    conn = sqlite3.connect("website_monitor.db")
    cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS checks (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            url TEXT NOT NULL,
            timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
            is_success BOOLEAN NOT NULL,
            status_code INTEGER,
            error_message TEXT,
            response_time REAL
        )
    """)
    conn.commit()
    conn.close()

# 读取配置
def load_config():
    with open("config.json") as f:
        return json.load(f)

# 发送邮件通知
def send_alert(url, error_count):
    msg = MIMEText(f"警告：网站 {url} 已连续 {error_count} 次检查失败")
    msg["Subject"] = "网站可用性警报"
    msg["From"] = "monitor@example.com"
    msg["To"] = "admin@example.com"
    
    with smtplib.SMTP("smtp.example.com") as server:
        server.send_message(msg)

# 主检查函数
def check_websites():
    config = load_config()
    conn = sqlite3.connect("website_monitor.db")
    cursor = conn.cursor()
    
    for site in config["websites"]:
        url = site["url"]
        try:
            response = requests.get(url, timeout=10)
            is_success = response.status_code == 200
            status_code = response.status_code if not is_success else None
            error_message = None
            response_time = response.elapsed.total_seconds()
        except Exception as e:
            is_success = False
            status_code = None
            error_message = str(e)
            response_time = None
        
        # 存入数据库
        cursor.execute("""
            INSERT INTO checks (url, is_success, status_code, error_message, response_time)
            VALUES (?, ?, ?, ?, ?)
        """, (url, is_success, status_code, error_message, response_time))
        
        # 检查是否需要发送警报
        if not is_success:
            cursor.execute("""
                SELECT COUNT(*) FROM checks 
                WHERE url = ? AND timestamp > datetime('now', '-1 hour') AND is_success = 0
            """, (url,))
            error_count = cursor.fetchone()[0]
            if error_count >= config["alert_threshold"]:
                send_alert(url, error_count)
    
    conn.commit()
    conn.close()

# 初始化
init_db()
config = load_config()

# 设置定时任务
scheduler = BlockingScheduler()
scheduler.add_job(check_websites, "interval", minutes=config["check_interval"])

print("网站监控系统已启动，按Ctrl+C退出")
try:
    scheduler.start()
except (KeyboardInterrupt, SystemExit):
    pass

配置文件config.json示例：

{
    "websites": [
        {"url": "https://www.example.com"},
        {"url": "https://www.test.com"}
    ],
    "check_interval": 5,  # 分钟
    "alert_threshold": 3  # 连续失败次数触发警报
}

五、性能优化与扩展建议

1. 异步IO优化：对于超大规模检查（如上千个网站），可使用aiohttp库实现异步请求，相比多线程能进一步降低资源消耗

2. 分布式架构：将检查任务分配到多台机器执行，通过Redis等消息队列协调任务分配

3. 智能重试机制：对临时性故障（如503服务不可用）实现指数退避重试，减少误报

4. 结合Prometheus：将检查指标暴露为Prometheus格式，利用Grafana构建专业监控面板

关键词：Python网站监控、多线程检查、requests库、日志记录、数据可视化、APScheduler定时任务、异常通知

简介：本文详细介绍了如何使用Python实现网站可用性的批量自动化检查，涵盖从基础单线程检查到多线程并发、异常处理、日志记录、数据可视化以及完整监控系统的构建方法，提供了可落地的代码示例和性能优化建议。

《利用python批量检查网站的可用性.doc》

将本文以doc文档格式下载到电脑，方便收藏和打印

推荐度：

点击下载文档