用脚本检测网络节点延迟

wen 实用脚本 2026-07-31 205

本文目录导读：

用脚本检测网络节点延迟

方案1：基础版（本机到多目标延迟检测）
方案2：进阶版（持续监测+统计报告）
方案3：终极版（Web界面+实时图表）
使用说明

我来为您编写一个网络节点延迟检测脚本,包含多个实用的实现方式。

方案1：基础版（本机到多目标延迟检测）

#!/usr/bin/env python3
import subprocess
import re
import time
import json
from concurrent.futures import ThreadPoolExecutor
class NetworkLatencyTester:
    def __init__(self, timeout=5):
        self.timeout = timeout
    def ping_test(self, host):
        """单次ping测试"""
        try:
            cmd = ['ping', '-c', '1', '-W', str(self.timeout), host]
            result = subprocess.run(cmd, capture_output=True, text=True, timeout=self.timeout+2)
            if result.returncode == 0:
                # 提取延迟信息
                match = re.search(r'time[=:]\s*([0-9.]+)', result.stdout)
                if match:
                    return {'host': host, 'latency': float(match.group(1)), 'status': 'success'}
            return {'host': host, 'latency': None, 'status': 'failed'}
        except Exception as e:
            return {'host': host, 'latency': None, 'status': 'error', 'message': str(e)}
    def trace_route(self, host, max_hops=20):
        """路由追踪测试"""
        try:
            cmd = ['traceroute', '-m', str(max_hops), '-n', host]
            result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
            hops = []
            for line in result.stdout.split('\n')[1:]:
                if line.strip():
                    parts = line.split()
                    if len(parts) >= 2 and parts[0].isdigit():
                        hop = int(parts[0])
                        latency = self._extract_latency(line)
                        hops.append({'hop': hop, 'ip': self._extract_ip(line), 'latency': latency})
            return {'host': host, 'hops': hops}
        except Exception as e:
            return {'host': host, 'error': str(e), 'hops': []}
    def _extract_latency(self, line):
        """从traceroute输出提取延迟"""
        matches = re.findall(r'([0-9.]+)\s*ms', line)
        if matches:
            latencies = [float(m) for m in matches[:3]]
            return latencies
        return None
    def _extract_ip(self, line):
        """提取IP地址"""
        match = re.search(r'\d+\.\d+\.\d+\.\d+', line)
        return match.group(0) if match else '*'
    def test_multiple_hosts(self, hosts, max_workers=5):
        """并发测试多个主机"""
        results = {}
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = {executor.submit(self.ping_test, host): host for host in hosts}
            for future in futures:
                host = futures[future]
                try:
                    results[host] = future.result()
                except Exception as e:
                    results[host] = {'host': host, 'error': str(e)}
        return results
# 使用示例
if __name__ == "__main__":
    tester = NetworkLatencyTester()
    # 测试目标
    targets = [
        '8.8.8.8',  # Google
        '1.1.1.1',  # Cloudflare
        '114.114.114.114',  # 114DNS
        'baidu.com',
        'google.com'
    ]
    print("=== 开始网络延迟测试 ===")
    # 测试所有目标
    results = tester.test_multiple_hosts(targets, max_workers=3)
    print("\n=== 延迟测试结果 ===")
    for host, result in results.items():
        if result.get('status') == 'success':
            print(f"✓ {host}: {result['latency']:.2f}ms")
        else:
            print(f"✗ {host}: 连接失败")
    # 执行路由追踪
    print("\n=== 路由追踪测试 ===")
    trace_result = tester.trace_route('8.8.8.8', max_hops=10)
    if trace_result.get('hops'):
        for hop in trace_result['hops']:
            latency_str = ', '.join([f"{l}ms" for l in hop['latency']]) if hop['latency'] else 'N/A'
            print(f"跳数 {hop['hop']:3d}: {hop['ip']:15s} 延迟: {latency_str}")

方案2：进阶版（持续监测+统计报告）

#!/usr/bin/env python3
import socket
import time
import statistics
import sys
from datetime import datetime
import csv
import signal
class ContinuousLatencyMonitor:
    """持续延迟监控器"""
    def __init__(self, host, interval=1, duration=60, threshold_ms=200):
        self.host = host
        self.interval = interval
        self.duration = duration
        self.threshold = threshold_ms
        self.running = True
        self.results = []
    def stop(self, signum, frame):
        """停止监控"""
        print("\n监视已停止")
        self.running = False
    def tcp_connect_test(self, port=80):
        """TCP连接延迟测试"""
        start_time = time.time()
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(2)
            target_ip = socket.gethostbyname(self.host)
            sock.connect((target_ip, port))
            latency = (time.time() - start_time) * 1000
            sock.close()
            return latency
        except Exception:
            return None
    def run_monitor(self, port=80):
        """运行持续监控"""
        print(f"开始监控 {self.host}:{port}")
        print(f"间隔: {self.interval}s, 持续时间: {self.duration}s")
        print("-" * 50)
        signal.signal(signal.SIGINT, self.stop)
        start_time = time.time()
        test_count = 0
        while self.running and (time.time() - start_time) < self.duration:
            latency = self.tcp_connect_test(port)
            test_count += 1
            timestamp = datetime.now().strftime("%H:%M:%S")
            if latency is not None:
                self.results.append(latency)
                status = "⚠️" if latency > self.threshold else "✓"
                print(f"[{timestamp}] {status} 延迟: {latency:.2f}ms")
            else:
                self.results.append(None)
                print(f"[{timestamp}] ✗ 连接失败")
            time.sleep(self.interval)
        self.print_report()
    def print_report(self):
        """输出统计报告"""
        valid_results = [r for r in self.results if r is not None]
        print("\n" + "=" * 50)
        print("监控统计报告")
        print("=" * 50)
        print(f"监控目标: {self.host}")
        print(f"测试次数: {len(self.results)}")
        if valid_results:
            print(f"成功次数: {len(valid_results)}")
            print(f"失败次数: {len(self.results) - len(valid_results)}")
            print(f"成功率: {(len(valid_results)/len(self.results))*100:.2f}%")
            print(f"\n延迟统计:")
            print(f"  最小: {min(valid_results):.2f}ms")
            print(f"  最大: {max(valid_results):.2f}ms")
            print(f"  平均: {statistics.mean(valid_results):.2f}ms")
            if len(valid_results) > 1:
                print(f"  方差: {statistics.stdev(valid_results):.2f}ms")
                print(f"  中位数: {statistics.median(valid_results):.2f}ms")
        else:
            print("没有成功的测试结果")
        # 导出到CSV
        self.export_to_csv()
    def export_to_csv(self, filename="latency_report.csv"):
        """导出结果到CSV"""
        with open(filename, 'w', newline='') as f:
            writer = csv.writer(f)
            writer.writerow(['Timestamp', 'Latency(ms)'])
            start_time = None
            for i, latency in enumerate(self.results):
                timestamp = int(time.time() - (len(self.results) - i) * self.interval)
                dt = datetime.fromtimestamp(timestamp)
                writer.writerow([dt.isoformat(), latency if latency is not None else 'N/A'])
        print(f"\n结果已保存到: {filename}")
# 使用示例
if __name__ == "__main__":
    monitor = ContinuousLatencyMonitor(
        host='baidu.com',  # 监控目标
        interval=2,        # 测试间隔（秒）
        duration=30,       # 持续时间（秒）
        threshold_ms=500   # 延迟阈值
    )
    monitor.run_monitor(port=80)

方案3：终极版（Web界面+实时图表）

<!DOCTYPE html>
<html>
<head>网络延迟监控系统</title>
    <style>
        body { font-family: Arial, sans-serif; margin: 20px; background-color: #f5f5f5; }
        .container { max-width: 1200px; margin: 0 auto; }
        .header { background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); color: white; padding: 20px; border-radius: 10px; }
        .targets { margin: 20px 0; display: flex; flex-wrap: wrap; gap: 10px; }
        .target-card { background: white; padding: 15px; border-radius: 8px; box-shadow: 0 2px 5px rgba(0,0,0,0.1); flex: 1; min-width: 200px; }
        .latency-value { font-size: 24px; font-weight: bold; }
        .status-success { color: #4caf50; }
        .status-failed { color: #f44336; }
        .controls { margin: 20px 0; padding: 20px; background: white; border-radius: 8px; }
        button { background: #667eea; color: white; border: none; padding: 10px 20px; border-radius: 5px; cursor: pointer; }
        input { padding: 8px; margin: 5px; border: 1px solid #ddd; border-radius: 4px; }
        .chart-container { margin: 20px 0; padding: 20px; background: white; border-radius: 8px; }
        #chartCanvas { width: 100%; height: 300px; }
    </style>
</head>
<body>
    <div class="container">
        <div class="header">
            <h1>📡 实时网络延迟监控系统</h1>
            <p>监控多个网络节点状态</p>
        </div>
        <div class="controls">
            <h3>监控配置</h3>
            <input type="text" id="hostInput" placeholder="输入目标地址" value="baidu.com">
            <button onclick="addTarget()">添加目标</button>
            <button onclick="startMonitoring()">开始监控</button>
            <button onclick="stopMonitoring()">停止监控</button>
        </div>
        <div class="targets" id="targets"></div>
        <div class="chart-container">
            <h3>延迟趋势图</h3>
            <canvas id="chartCanvas"></canvas>
        </div>
    </div>
    <script>
        let monitoring = false;
        let intervalId = null;
        let targets = ['baidu.com', 'google.com', '8.8.8.8'];
        let data = {}; // 存储每个目标的延迟数据
        function init() {
            displayTargets();
            for (let target of targets) {
                data[target] = [];
            }
        }
        function displayTargets() {
            const container = document.getElementById('targets');
            container.innerHTML = '';
            for (let target of targets) {
                const card = document.createElement('div');
                card.className = 'target-card';
                card.id = `card-${target}`;
                card.innerHTML = `
                    <h4>${target}</h4>
                    <div class="latency-value" id="value-${target}">等待测试...</div>
                    <div class="status" id="status-${target}">待机</div>
                `;
                container.appendChild(card);
            }
        }
        async function testLatency(target) {
            const start = performance.now();
            try {
                const response = await fetch(`/api/ping?host=${target}&_=${Date.now()}`);
                const latency = performance.now() - start;
                return { success: true, latency };
            } catch (error) {
                return { success: false, latency: null };
            }
        }
        function updateUI(target, result) {
            const valueEl = document.getElementById(`value-${target}`);
            const statusEl = document.getElementById(`status-${target}`);
            if (result.success) {
                valueEl.textContent = `${result.latency.toFixed(2)}ms`;
                valueEl.className = 'latency-value status-success';
                statusEl.className = 'status status-success';
                statusEl.textContent = '正常';
                // 存储数据用于图表
                data[target].push({ time: new Date(), latency: result.latency });
                if (data[target].length > 50) {
                    data[target].shift();
                }
            } else {
                valueEl.textContent = '超时';
                valueEl.className = 'latency-value status-failed';
                statusEl.className = 'status status-failed';
                statusEl.textContent = '连接失败';
            }
        }
        async function monitorTargets() {
            for (let target of targets) {
                const result = await testLatency(target);
                updateUI(target, result);
            }
            drawGraph();
        }
        function drawGraph() {
            const canvas = document.getElementById('chartCanvas');
            const ctx = canvas.getContext('2d');
            ctx.clearRect(0, 0, canvas.width, canvas.height);
            // 绘制背景网格
            ctx.strokeStyle = '#e0e0e0';
            ctx.lineWidth = 1;
            for (let i = 0; i < 10; i++) {
                ctx.beginPath();
                ctx.moveTo(0, (canvas.height / 10) * i);
                ctx.lineTo(canvas.width, (canvas.height / 10) * i);
                ctx.stroke();
            }
            // 绘制每个目标的数据线
            const colors = ['#667eea', '#f44336', '#4caf50', '#ff9800', '#9c27b0'];
            let colorIndex = 0;
            for (let target of targets) {
                if (data[target].length < 2) continue;
                ctx.beginPath();
                ctx.strokeStyle = colors[colorIndex % colors.length];
                ctx.lineWidth = 2;
                const maxLatency = 1000; // 假设最大延迟1000ms
                const step = canvas.width / 50;
                data[target].forEach((point, index) => {
                    const x = index * step;
                    const y = canvas.height - (point.latency / maxLatency) * (canvas.height - 20);
                    if (index === 0) {
                        ctx.moveTo(x, y);
                    } else {
                        ctx.lineTo(x, y);
                    }
                });
                ctx.stroke();
                colorIndex++;
            }
            // 绘制Y轴标签
            ctx.fillStyle = '#333';
            ctx.font = '12px Arial';
            ctx.fillText('1000ms', 5, canvas.height - 15);
            ctx.fillText('500ms', 5, canvas.height / 2);
            ctx.fillText('0ms', 5, 15);
        }
        function addTarget() {
            const host = document.getElementById('hostInput').value.trim();
            if (host && !targets.includes(host)) {
                targets.push(host);
                data[host] = [];
                displayTargets();
            }
        }
        function startMonitoring() {
            if (!monitoring) {
                monitoring = true;
                monitorTargets();
                intervalId = setInterval(monitorTargets, 2000);
            }
        }
        function stopMonitoring() {
            monitoring = false;
            if (intervalId) {
                clearInterval(intervalId);
                intervalId = null;
            }
        }
        // 初始化应用
        init();
    </script>
</body>
</html>

使用说明

安装依赖

# Python方案所需
pip install concurrent.futures  # Python 3.x 自带
# Web方案需要
pip install flask  # 如果使用Flask后端

运行方式

方案1（基础版）：

python3 network_latency_test.py

方案2（高级版）：

python3 continuous_monitor.py

方案3（Web版）： 需要配置后端API：

from flask import Flask, jsonify, request
import subprocess
import re
app = Flask(__name__)
@app.route('/api/ping')
def ping():
    host = request.args.get('host', 'localhost')
    try:
        cmd = ['ping', '-c', '1', '-W', '2', host]
        result = subprocess.run(cmd, capture_output=True, text=True)
        match = re.search(r'time.*?=\s*([0-9.]+)', result.stdout)
        if match:
            return jsonify({'success': True, 'latency': float(match.group(1))})
    except:
        pass
    return jsonify({'success': False})
if __name__ == '__main__':
    app.run(port=5000)

功能特点

多目标并发检测：可同时测试多个目标节点
实时监控：持续追踪网络状态变化
统计分析：计算平均值、方差、成功率等
可视化图表：Web界面显示实时趋势
路由追踪：诊断网络路径问题

这些脚本覆盖了从基础到高级的网络延迟监控需求,您可以根据实际需要选择使用。

本文地址： https://www.bmjup.com/post/34855.html