网络延迟高怎么办？用ping命令实时监测网络状态！

网络状态实时监测工具

下面是一个基于Python的图形化网络状态实时监测工具,使用Ping命令检查目标主机的连通性并显示实时结果。

import subprocess
import time
import platform
import tkinter as tk
from tkinter import ttk, messagebox
import matplotlib.pyplot as plt
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg
import threading
import queue
class PingMonitor:
    def __init__(self, root):
        self.root = root
        self.root.title("网络状态实时监测工具")
        self.root.geometry("900x700")
        self.root.resizable(True, True)
        # 创建队列用于线程间通信
        self.data_queue = queue.Queue()
        # 初始化变量
        self.target_host = tk.StringVar(value="www.google.com")
        self.interval = tk.DoubleVar(value=1.0)  # 默认1秒
        self.ping_count = tk.IntVar(value=0)
        self.success_count = tk.IntVar(value=0)
        self.fail_count = tk.IntVar(value=0)
        self.running = False
        self.ping_thread = None
        # 创建UI
        self.create_widgets()
        # 初始化图表
        self.setup_chart()
        # 初始化数据列表
        self.timestamps = []
        self.latencies = []
        self.statuses = []  # 0表示失败，1表示成功
    def create_widgets(self):
        # 控制面板
        control_frame = ttk.LabelFrame(self.root, text="控制面板")
        control_frame.pack(fill=tk.X, padx=10, pady=10)
        # 目标主机输入
        ttk.Label(control_frame, text="目标主机:").grid(row=0, column=0, padx=5, pady=5, sticky=tk.W)
        host_entry = ttk.Entry(control_frame, textvariable=self.target_host, width=30)
        host_entry.grid(row=0, column=1, padx=5, pady=5, sticky=tk.W)
        # 间隔时间
        ttk.Label(control_frame, text="检测间隔(秒):").grid(row=0, column=2, padx=5, pady=5, sticky=tk.W)
        interval_spin = ttk.Spinbox(control_frame, from_=0.1, to=10, increment=0.1, 
                                   textvariable=self.interval, width=5)
        interval_spin.grid(row=0, column=3, padx=5, pady=5, sticky=tk.W)
        # 按钮
        self.start_button = ttk.Button(control_frame, text="开始监测", command=self.start_monitoring)
        self.start_button.grid(row=0, column=4, padx=5, pady=5)
        self.stop_button = ttk.Button(control_frame, text="停止监测", command=self.stop_monitoring, state=tk.DISABLED)
        self.stop_button.grid(row=0, column=5, padx=5, pady=5)
        # 统计信息
        stats_frame = ttk.LabelFrame(self.root, text="统计信息")
        stats_frame.pack(fill=tk.X, padx=10, pady=5)
        ttk.Label(stats_frame, text="已发送请求:").grid(row=0, column=0, padx=10, pady=5, sticky=tk.W)
        ttk.Label(stats_frame, textvariable=self.ping_count).grid(row=0, column=1, padx=10, pady=5, sticky=tk.W)
        ttk.Label(stats_frame, text="成功次数:").grid(row=0, column=2, padx=10, pady=5, sticky=tk.W)
        ttk.Label(stats_frame, textvariable=self.success_count, foreground="green").grid(row=0, column=3, padx=10, pady=5, sticky=tk.W)
        ttk.Label(stats_frame, text="失败次数:").grid(row=0, column=4, padx=10, pady=5, sticky=tk.W)
        ttk.Label(stats_frame, textvariable=self.fail_count, foreground="red").grid(row=0, column=5, padx=10, pady=5, sticky=tk.W)
        # 日志区域
        log_frame = ttk.LabelFrame(self.root, text="实时日志")
        log_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=5)
        self.log_text = tk.Text(log_frame, height=8)
        self.log_text.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)
        self.log_text.config(state=tk.DISABLED)
        # 图表区域
        chart_frame = ttk.LabelFrame(self.root, text="延迟图表")
        chart_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=5)
        self.figure = plt.Figure(figsize=(8, 4), dpi=100)
        self.ax = self.figure.add_subplot(111)
        self.canvas = FigureCanvasTkAgg(self.figure, master=chart_frame)
        self.canvas.get_tk_widget().pack(fill=tk.BOTH, expand=True)
        # 状态栏
        self.status_var = tk.StringVar(value="就绪")
        status_bar = ttk.Label(self.root, textvariable=self.status_var, relief=tk.SUNKEN, anchor=tk.W)
        status_bar.pack(side=tk.BOTTOM, fill=tk.X)
    def setup_chart(self):
        self.ax.clear()
        self.ax.set_title('网络延迟监测')
        self.ax.set_xlabel('时间')
        self.ax.set_ylabel('延迟 (ms)')
        self.ax.grid(True, linestyle='--', alpha=0.7)
        self.canvas.draw()
    def update_chart(self):
        if not self.timestamps or not self.latencies:
            return
        self.ax.clear()
        # 绘制成功点
        success_times = [self.timestamps[i] for i in range(len(self.statuses)) if self.statuses[i] == 1]
        success_latencies = [self.latencies[i] for i in range(len(self.statuses)) if self.statuses[i] == 1]
        if success_times:
            self.ax.plot(success_times, success_latencies, 'go-', label='成功', linewidth=1, markersize=4)
        # 绘制失败点
        fail_times = [self.timestamps[i] for i in range(len(self.statuses)) if self.statuses[i] == 0]
        fail_latencies = [self.latencies[i] for i in range(len(self.statuses)) if self.statuses[i] == 0]
        if fail_times:
            self.ax.plot(fail_times, fail_latencies, 'ro', label='失败', markersize=6)
        # 设置图表属性
        self.ax.set_title(f'目标: {self.target_host.get()} 延迟监测')
        self.ax.set_xlabel('时间')
        self.ax.set_ylabel('延迟 (ms)')
        self.ax.grid(True, linestyle='--', alpha=0.7)
        self.ax.legend()
        # 自动调整Y轴范围
        if success_latencies:
            max_val = max(success_latencies) * 1.2 if max(success_latencies) > 0 else 100
            self.ax.set_ylim(0, max_val)
        self.canvas.draw()
    def log_message(self, message, color="black"):
        self.log_text.config(state=tk.NORMAL)
        self.log_text.insert(tk.END, message + "n", color)
        self.log_text.tag_config("red", foreground="red")
        self.log_text.tag_config("green", foreground="green")
        self.log_text.tag_config("blue", foreground="blue")
        self.log_text.see(tk.END)  # 滚动到最新消息
        self.log_text.config(state=tk.DISABLED)
    def ping_host(self):
        """执行Ping命令并返回结果"""
        host = self.target_host.get()
        # 根据操作系统选择不同的ping参数
        param = "-n" if platform.system().lower() == "windows" else "-c"
        count = "1"  # 只发送一个包
        try:
            # 执行ping命令
            output = subprocess.check_output(
                ["ping", param, count, host],
                stderr=subprocess.STDOUT,
                universal_newlines=True,
                timeout=self.interval.get()
            )
            # 解析ping结果
            if "time=" in output:
                # 提取延迟时间
                time_str = [line for line in output.split('n') if "time=" in line][0]
                latency = float(time_str.split("time=")[1].split(" ")[0])
                status = 1  # 成功
            else:
                latency = 0
                status = 0  # 失败
            return latency, status, output
        except subprocess.CalledProcessError:
            return 0, 0, "请求超时或目标不可达"
        except subprocess.TimeoutExpired:
            return 0, 0, "命令执行超时"
        except Exception as e:
            return 0, 0, f"错误: {str(e)}"
    def monitoring_loop(self):
        """监控循环"""
        while self.running:
            start_time = time.time()
            # 执行ping
            latency, status, output = self.ping_host()
            timestamp = time.strftime("%H:%M:%S")
            # 更新统计数据
            self.ping_count.set(self.ping_count.get() + 1)
            if status == 1:
                self.success_count.set(self.success_count.get() + 1)
                log_color = "green"
                log_message = f"[{timestamp}] 来自 {self.target_host.get()} 的回复: 时间={latency:.2f}ms"
            else:
                self.fail_count.set(self.fail_count.get() + 1)
                log_color = "red"
                log_message = f"[{timestamp}] 请求超时"
            # 将数据放入队列
            self.data_queue.put((timestamp, latency, status, log_message, log_color))
            # 计算剩余等待时间
            elapsed = time.time() - start_time
            wait_time = max(0, self.interval.get() - elapsed)
            time.sleep(wait_time)
    def start_monitoring(self):
        """开始监控"""
        if not self.target_host.get():
            messagebox.showerror("错误", "请输入目标主机地址")
            return
        if self.running:
            return
        # 重置状态
        self.running = True
        self.ping_count.set(0)
        self.success_count.set(0)
        self.fail_count.set(0)
        self.timestamps = []
        self.latencies = []
        self.statuses = []
        # 清空日志
        self.log_text.config(state=tk.NORMAL)
        self.log_text.delete(1.0, tk.END)
        self.log_text.config(state=tk.DISABLED)
        # 更新按钮状态
        self.start_button.config(state=tk.DISABLED)
        self.stop_button.config(state=tk.NORMAL)
        # 更新状态栏
        self.status_var.set(f"正在监测 {self.target_host.get()} ...")
        # 启动监控线程
        self.ping_thread = threading.Thread(target=self.monitoring_loop, daemon=True)
        self.ping_thread.start()
        # 启动UI更新循环
        self.update_ui()
        self.log_message(f"开始监测 {self.target_host.get()}，间隔 {self.interval.get()} 秒", "blue")
    def stop_monitoring(self):
        """停止监控"""
        if self.running:
            self.running = False
            if self.ping_thread and self.ping_thread.is_alive():
                self.ping_thread.join(timeout=1.0)
            self.start_button.config(state=tk.NORMAL)
            self.stop_button.config(state=tk.DISABLED)
            self.status_var.set("监测已停止")
            self.log_message("监测已停止", "blue")
    def update_ui(self):
        """更新UI元素"""
        # 处理队列中的所有消息
        while not self.data_queue.empty():
            timestamp, latency, status, log_message, log_color = self.data_queue.get()
            # 更新日志
            self.log_message(log_message, log_color)
            # 更新数据列表
            self.timestamps.append(timestamp)
            self.latencies.append(latency)
            self.statuses.append(status)
            # 更新图表
            if len(self.timestamps) % 3 == 0 or status == 0:  # 每3个点更新一次图表，或者有失败时更新
                self.update_chart()
        # 如果仍在运行，则继续调度更新
        if self.running:
            self.root.after(100, self.update_ui)
if __name__ == "__main__":
    root = tk.Tk()
    app = PingMonitor(root)
    root.mainloop()

功能说明

这个网络状态实时监测工具提供以下功能：

1. 目标主机监测：可以输入任何域名或IP地址进行监测
2. 实时日志：显示每次Ping请求的结果（成功或失败）
3. 统计信息：
   • 已发送请求总数
   • 成功次数（绿色显示）
   • 失败次数（红色显示）
4. 延迟图表：以折线图形式展示网络延迟变化，绿色点表示成功，红色点表示失败
5. 控制选项：
   • 可调节检测间隔（0.1-10秒）
   • 开始/停止监测按钮

使用说明

1. 在”目标主机”输入框中输入要监测的域名或IP地址（如www.google.com）
2. 设置检测间隔（默认1秒）
3. 点击”开始监测”按钮启动监控
4. 实时日志将显示每次Ping的结果
5. 图表区域将显示网络延迟的变化情况
6. 点击”停止监测”按钮结束监控

运行要求

• Python 3.x
• 需要安装以下库：

  pip install matplotlib tk

这个工具可以帮助您实时监测网络连接状态，快速发现网络中断或延迟增加的问题,适用于网络维护和故障排查场景。