基于Python實現(xiàn)Windows帶寬監(jiān)控工具

更新時間：2025年09月09日 09:21:36 作者：盜理者

這篇文章主要為大家詳細介紹了如何基于Python實現(xiàn)一個簡單的Windows帶寬監(jiān)控工具,文中的示例代碼講解詳細,感興趣的小伙伴可以跟隨小編一起學習一下

環(huán)境

Python 3.12.2

psutil 版本: 7.0.0

Flask 版本: 3.1.2

matplotlib 版本: 3.10.6

pip 安裝指定版本

pip install psutil==7.0.0 flask==3.1.2 matplotlib==3.10.6

介紹

做一個適用于 windows 的帶寬監(jiān)控工具，這個工具融合了Python后端開發(fā)、多線程、系統(tǒng)信息采集、Web框架、前端可視化等多個技術(shù)領(lǐng)域。

源碼可以看開源地址

會使用的庫

庫名	核心作用	代碼中的應(yīng)用場景
threading	實現(xiàn)多線程編程，讓程序“同時做多個任務(wù)”	創(chuàng)建TrafficMonitor監(jiān)控線程，與FlaskWeb服務(wù)并行運行
time	提供時間相關(guān)功能（延遲、時間戳等）	time.sleep(1)實現(xiàn)每秒采集一次流量數(shù)據(jù)
datetime	處理日期和時間，提供更豐富的時間格式	記錄流量數(shù)據(jù)對應(yīng)的時間戳（如datetime.now()）
collections.deque	雙端隊列，支持固定長度（自動丟棄舊數(shù)據(jù)）	存儲最近5分鐘的流量數(shù)據(jù)（maxlen=300，每秒1條共300秒）
io	內(nèi)存中的輸入輸出流，模擬文件操作	存儲Matplotlib生成的圖表（避免寫入本地文件）
base64	將二進制數(shù)據(jù)編碼為文本格式	把圖表二進制數(shù)據(jù)轉(zhuǎn)為base64，方便前端HTML顯示
json	處理JSON數(shù)據(jù)（Python字典與JSON字符串互轉(zhuǎn)）	生成JSON格式的流量報告，供前端下載
os	與操作系統(tǒng)交互（本代碼中未實際使用，預(yù)留擴展）	可用于獲取系統(tǒng)信息、管理文件路徑等

多線程關(guān)鍵代碼

class TrafficMonitor(threading.Thread):
    def __init__(self, interface_name=None):
        super().__init__()  # 調(diào)用父類構(gòu)造函數(shù)
        self.daemon = True  # 設(shè)置為守護線程
        self.running = True  # 控制線程循環(huán)的開關(guān)

    def run(self):
        # 1. 自動選擇網(wǎng)絡(luò)接口（優(yōu)先非回環(huán)接口，且有發(fā)送數(shù)據(jù)的接口）
        # 2. 初始化上次流量統(tǒng)計值（用于計算每秒增量）
        # 3. 循環(huán)采集數(shù)據(jù)：
        while self.running:
            current_stats = psutil.net_io_counters(pernic=True)[self.interface_name]
            incoming_rate = current_stats.bytes_recv - self.last_incoming  # 每秒入站流量
            outgoing_rate = current_stats.bytes_sent - self.last_outgoing  # 每秒出站流量
            # 更新全局變量（供Web端調(diào)用）
            traffic_data['incoming'].append(incoming_rate)
            time.sleep(1)  # 每秒采集一次

    def stop(self):
        self.running = False  # 關(guān)閉循環(huán)，線程退出

系統(tǒng)流量采集：用 psutil 獲取網(wǎng)絡(luò)數(shù)據(jù)

概念：網(wǎng)絡(luò)流量的“增量”與“總量”

總量（bytes_recv/bytes_sent）：從系統(tǒng)啟動到當前，網(wǎng)絡(luò)接口接收/發(fā)送的總字節(jié)數(shù)（不會重置）。

增量（每秒流量）：當前總量 - 上次總量，即每秒的實際流量（如“100KB/s”）。

代碼中的流量采集邏輯

選擇網(wǎng)絡(luò)接口：若未指定接口（如TrafficMonitor()），代碼會自動遍歷所有接口，排除回環(huán)接口（lo，本地測試用），選擇有數(shù)據(jù)發(fā)送的接口（bytes_sent > 0）。

初始化上次總量：self.last_incoming = interfaces[self.interface_name].bytes_recv，記錄初始總量。

計算每秒增量：每次循環(huán)中，用當前總量減去上次總量，得到每秒流量（如incoming_rate），再更新上次總量。

存儲數(shù)據(jù)：將每秒流量和對應(yīng)的時間戳存入全局變量traffic_data（deque類型，自動保留最近300條）。

Flask Web框架：搭建后端服務(wù)

Flask是輕量級Web框架，這部分有不明白的可以參考：python flask編寫一個簡易的web端程序(附demo)

前端部分

交互邏輯（JavaScript）

前端的核心是“實時獲取后端數(shù)據(jù)并更新頁面”，主要通過以下函數(shù)實現(xiàn)：

1. fetch()：異步請求后端數(shù)據(jù)

fetch(‘/traffic-data’)會向后端/traffic-data路由發(fā)送請求，獲取JSON格式的流量數(shù)據(jù)，再用這些數(shù)據(jù)更新圖表。

2. Chart.js：繪制實時圖表

Chart.js是輕量級前端繪圖庫，代碼中用它繪制入站/出站流量曲線：

初始化圖表：指定canvas元素、圖表類型（line折線圖）、初始數(shù)據(jù)（空）、坐標軸配置。
更新圖表：每次fetch到新數(shù)據(jù)后，修改chart.data.labels（時間戳）和chart.data.datasets（流量數(shù)據(jù)），再調(diào)用chart.update()刷新圖表。

3. 定時更新：setInterval()

用setInterval實現(xiàn)周期性更新：

setInterval(updateChart, 1000)：每秒更新一次圖表（與后端采集頻率一致）。
setInterval(updateTotalStats, 5000)：每5秒更新一次總流量（總流量變化較慢，無需頻繁更新）。

4. 輔助函數(shù)：格式轉(zhuǎn)換

formatBytes()：將字節(jié)數(shù)（如102400）轉(zhuǎn)為易讀格式（如100 KB），支持B/KB/MB/GB/TB。
formatBytesPerSec()：在formatBytes()基礎(chǔ)上添加/s，如100 KB/s。

完整代碼

完整 index 代碼

<!DOCTYPE html>
<html>
<head>
    <title>Windows帶寬監(jiān)控</title>
    <script src="https://cdn.jsdelivr.net/npm/chart.js"></script>
    <style>
        body { font-family: Arial, sans-serif; margin: 20px; background-color: #f5f5f5; }
        .container { max-width: 1200px; margin: 0 auto; background: white; padding: 20px; border-radius: 8px; box-shadow: 0 2px 4px rgba(0,0,0,0.1); }
        .header { text-align: center; margin-bottom: 20px; }
        .stats-container { display: flex; justify-content: space-around; margin-bottom: 20px; }
        .stat-card { background: #f0f0f0; padding: 15px; border-radius: 5px; text-align: center; min-width: 200px; }
        .stat-value { font-size: 24px; font-weight: bold; }
        .chart-container { position: relative; height: 400px; margin-bottom: 20px; }
        .controls { margin-bottom: 20px; text-align: center; }
        button { background: #4CAF50; color: white; border: none; padding: 10px 15px; border-radius: 4px; cursor: pointer; }
        button:hover { background: #45a049; }
    </style>
</head>
<body>
<div class="container">
    <div class="header">
        <h1>Windows帶寬監(jiān)控工具</h1>
        <p>實時監(jiān)控網(wǎng)絡(luò)接口流量</p>
    </div>

    <div class="stats-container">
        <div class="stat-card">
            <div class="stat-title">當前入站流量</div>
            <div class="stat-value" id="current-in">0 B/s</div>
        </div>
        <div class="stat-card">
            <div class="stat-title">當前出站流量</div>
            <div class="stat-value" id="current-out">0 B/s</div>
        </div>
        <div class="stat-card">
            <div class="stat-title">總?cè)胝玖髁?lt;/div>
            <div class="stat-value" id="total-in">0 MB</div>
        </div>
        <div class="stat-card">
            <div class="stat-title">總出站流量</div>
            <div class="stat-value" id="total-out">0 MB</div>
        </div>
    </div>

    <div class="controls">
        <button onclick="downloadReport()">下載報表</button>
        <button onclick="changeView('minute')">最近1分鐘</button>
        <button onclick="changeView('hour')">最近1小時</button>
        <button onclick="changeView('day')">最近24小時</button>
    </div>

    <div class="chart-container">
        <canvas id="trafficChart"></canvas>
    </div>

    <div id="report"></div>
</div>

<script>
    // 創(chuàng)建圖表
    const ctx = document.getElementById('trafficChart').getContext('2d');
    const chart = new Chart(ctx, {
        type: 'line',
        data: {
            labels: [],
            datasets: [
                {
                    label: '入站流量 (KB/s)',
                    data: [],
                    borderColor: 'rgba(75, 192, 192, 1)',
                    backgroundColor: 'rgba(75, 192, 192, 0.2)',
                    fill: true,
                    tension: 0.4
                },
                {
                    label: '出站流量 (KB/s)',
                    data: [],
                    borderColor: 'rgba(255, 99, 132, 1)',
                    backgroundColor: 'rgba(255, 99, 132, 0.2)',
                    fill: true,
                    tension: 0.4
                }
            ]
        },
        options: {
            responsive: true,
            maintainAspectRatio: false,
            scales: {
                y: {
                    beginAtZero: true,
                    title: {
                        display: true,
                        text: '流量 (KB/秒)'
                    }
                },
                x: {
                    title: {
                        display: true,
                        text: '時間'
                    }
                }
            },
            plugins: {
                legend: {
                    position: 'top',
                },
                title: {
                    display: true,
                    text: '網(wǎng)絡(luò)流量監(jiān)控'
                }
            }
        }
    });

    // 格式化字節(jié)大小為易讀格式
    function formatBytes(bytes, decimals = 2) {
        if (bytes === 0) return '0 B';

        const k = 1024;
        const dm = decimals < 0 ? 0 : decimals;
        const sizes = ['B', 'KB', 'MB', 'GB', 'TB'];

        const i = Math.floor(Math.log(bytes) / Math.log(k));

        return parseFloat((bytes / Math.pow(k, i)).toFixed(dm)) + ' ' + sizes[i];
    }

    // 格式化字節(jié)/秒為易讀格式
    function formatBytesPerSec(bytes, decimals = 2) {
        return formatBytes(bytes, decimals) + '/s';
    }

    // 更新圖表數(shù)據(jù)
    function updateChart() {
        fetch('/traffic-data')
            .then(response => response.json())
            .then(data => {
                // 轉(zhuǎn)換數(shù)據(jù)為KB/s
                const incomingKB = data.incoming.map(value => (value / 1024).toFixed(2));
                const outgoingKB = data.outgoing.map(value => (value / 1024).toFixed(2));

                chart.data.labels = data.timestamps;
                chart.data.datasets[0].data = incomingKB;
                chart.data.datasets[1].data = outgoingKB;
                chart.update();

                // 更新當前流量顯示
                if (incomingKB.length > 0) {
                    const currentIn = incomingKB[incomingKB.length - 1];
                    document.getElementById('current-in').textContent = formatBytesPerSec(currentIn * 1024);
                }

                if (outgoingKB.length > 0) {
                    const currentOut = outgoingKB[outgoingKB.length - 1];
                    document.getElementById('current-out').textContent = formatBytesPerSec(currentOut * 1024);
                }
            });
    }

    // 更新總流量統(tǒng)計
    function updateTotalStats() {
        fetch('/total-traffic')
            .then(response => response.json())
            .then(data => {
                document.getElementById('total-in').textContent = formatBytes(data.total_incoming);
                document.getElementById('total-out').textContent = formatBytes(data.total_outgoing);
            });
    }

    // 更新報告
    function updateReport() {
        fetch('/report')
            .then(response => response.text())
            .then(data => {
                document.getElementById('report').innerHTML = data;
            });
    }

    // 下載報表
    function downloadReport() {
        fetch('/download-report')
            .then(response => response.blob())
            .then(blob => {
                const url = window.URL.createObjectURL(blob);
                const a = document.createElement('a');
                a.style.display = 'none';
                a.href = url;
                a.download = 'bandwidth_report.json';
                document.body.appendChild(a);
                a.click();
                window.URL.revokeObjectURL(url);
            });
    }

    // 切換視圖
    function changeView(range) {
        fetch('/change-view?range=' + range)
            .then(response => response.json())
            .then(data => {
                // 重新加載頁面數(shù)據(jù)
                updateChart();
                updateReport();
            });
    }

    // 初始加載
    updateChart();
    updateTotalStats();
    updateReport();

    // 定時更新
    setInterval(updateChart, 1000);
    setInterval(updateTotalStats, 5000);
    setInterval(updateReport, 10000);
</script>
</body>
</html>

完整后端代碼

import threading
import time
import psutil
from datetime import datetime
from collections import deque
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
from flask import Flask, render_template, jsonify, request
import io
import base64
import json
import os

app = Flask(__name__)

# 全局變量存儲流量數(shù)據(jù)
traffic_data = {
    'incoming': deque(maxlen=300),  # 存儲最近5分鐘的入站流量(每秒一個數(shù)據(jù)點)
    'outgoing': deque(maxlen=300),  # 存儲最近5分鐘的出站流量
    'timestamps': deque(maxlen=300)  # 存儲對應(yīng)的時間戳
}


# 流量統(tǒng)計類
class TrafficMonitor(threading.Thread):
    def __init__(self, interface_name=None):
        super().__init__()
        self.daemon = True
        self.last_incoming = 0
        self.last_outgoing = 0
        self.interface_name = interface_name
        self.running = True

    def run(self):
        print("開始監(jiān)控網(wǎng)絡(luò)流量...")

        # 獲取網(wǎng)絡(luò)接口
        interfaces = psutil.net_io_counters(pernic=True)

        # 如果沒有指定接口，使用第一個活動接口
        if not self.interface_name:
            for name in interfaces:
                if name != 'lo' and interfaces[name].bytes_sent > 0:
                    self.interface_name = name
                    break

        if not self.interface_name:
            print("未找到可用的網(wǎng)絡(luò)接口")
            return

        print(f"監(jiān)控接口: {self.interface_name}")

        # 初始化計數(shù)器
        self.last_incoming = interfaces[self.interface_name].bytes_recv
        self.last_outgoing = interfaces[self.interface_name].bytes_sent

        # 開始監(jiān)控循環(huán)
        while self.running:
            try:
                # 獲取當前流量統(tǒng)計
                current_stats = psutil.net_io_counters(pernic=True)[self.interface_name]
                current_incoming = current_stats.bytes_recv
                current_outgoing = current_stats.bytes_sent

                # 計算每秒流量
                incoming_rate = current_incoming - self.last_incoming
                outgoing_rate = current_outgoing - self.last_outgoing

                # 更新計數(shù)器
                self.last_incoming = current_incoming
                self.last_outgoing = current_outgoing

                # 更新全局流量數(shù)據(jù)
                now = datetime.now()
                traffic_data['timestamps'].append(now)
                traffic_data['incoming'].append(incoming_rate)
                traffic_data['outgoing'].append(outgoing_rate)

                # 每秒更新一次
                time.sleep(1)

            except Exception as e:
                print(f"監(jiān)控出錯: {e}")
                time.sleep(5)

    def stop(self):
        self.running = False


# 創(chuàng)建并啟動流量監(jiān)控線程
monitor = TrafficMonitor()


# Flask路由
@app.route('/')
def index():
    return render_template('index.html')


@app.route('/traffic-data')
def get_traffic_data():
    # 返回JSON格式的流量數(shù)據(jù)
    data = {
        'timestamps': [ts.strftime('%H:%M:%S') for ts in traffic_data['timestamps']],
        'incoming': list(traffic_data['incoming']),
        'outgoing': list(traffic_data['outgoing'])
    }
    return jsonify(data)


@app.route('/total-traffic')
def get_total_traffic():
    # 獲取總流量統(tǒng)計
    interfaces = psutil.net_io_counters(pernic=True)
    interface_name = monitor.interface_name

    total_incoming = interfaces[interface_name].bytes_recv if interface_name in interfaces else 0
    total_outgoing = interfaces[interface_name].bytes_sent if interface_name in interfaces else 0

    return jsonify({
        'total_incoming': total_incoming,
        'total_outgoing': total_outgoing
    })


@app.route('/traffic-plot')
def get_traffic_plot():
    # 生成流量圖表并返回base64編碼的圖像
    if not traffic_data['timestamps']:
        return "暫無數(shù)據(jù)"

    plt.figure(figsize=(10, 6))

    # 轉(zhuǎn)換數(shù)據(jù)為KB/s
    incoming_kb = [x / 1024 for x in traffic_data['incoming']]
    outgoing_kb = [x / 1024 for x in traffic_data['outgoing']]

    plt.plot(traffic_data['timestamps'], incoming_kb, label='入站流量 (KB/s)')
    plt.plot(traffic_data['timestamps'], outgoing_kb, label='出站流量 (KB/s)')

    # 格式化圖表
    plt.xlabel('時間')
    plt.ylabel('流量 (KB/秒)')
    plt.title('實時網(wǎng)絡(luò)流量監(jiān)控')
    plt.legend()
    plt.grid(True)
    plt.gcf().autofmt_xdate()
    plt.gca().xaxis.set_major_formatter(mdates.DateFormatter('%H:%M:%S'))

    # 將圖表轉(zhuǎn)換為base64編碼
    img = io.BytesIO()
    plt.savefig(img, format='png')
    img.seek(0)
    plot_url = base64.b64encode(img.getvalue()).decode()
    plt.close()

    return f'<img src="data:image/png;base64,{plot_url}">'


@app.route('/report')
def generate_report():
    # 生成帶寬使用報告
    if not traffic_data['timestamps']:
        return "<p>暫無數(shù)據(jù)可生成報告</p>"

    # 計算統(tǒng)計信息
    total_in = sum(traffic_data['incoming'])
    total_out = sum(traffic_data['outgoing'])
    avg_in = total_in / len(traffic_data['incoming'])
    avg_out = total_out / len(traffic_data['outgoing'])
    max_in = max(traffic_data['incoming'])
    max_out = max(traffic_data['outgoing'])

    # 轉(zhuǎn)換為更友好的單位
    def format_bytes(bytes):
        for unit in ['B', 'KB', 'MB', 'GB']:
            if bytes < 1024.0:
                return f"{bytes:.2f} {unit}"
            bytes /= 1024.0
        return f"{bytes:.2f} TB"

    def format_bps(bytes_per_sec):
        return format_bytes(bytes_per_sec) + "/s"

    report = f"""
    <h2>帶寬使用報告</h2>
    <p>統(tǒng)計時間段: {traffic_data['timestamps'][0]} 到 {traffic_data['timestamps'][-1]}</p>
    <p>監(jiān)控時長: {len(traffic_data['timestamps'])} 秒</p>
    <p>總?cè)胝玖髁? {format_bytes(total_in)}</p>
    <p>總出站流量: {format_bytes(total_out)}</p>
    <p>平均入站速率: {format_bps(avg_in)}</p>
    <p>平均出站速率: {format_bps(avg_out)}</p>
    <p>最大入站速率: {format_bps(max_in)}</p>
    <p>最大出站速率: {format_bps(max_out)}</p>
    """

    return report


@app.route('/download-report')
def download_report():
    # 生成并下載JSON格式的詳細報告
    if not traffic_data['timestamps']:
        return "暫無數(shù)據(jù)", 404

    # 準備報告數(shù)據(jù)
    report_data = {
        "generated_at": datetime.now().isoformat(),
        "time_period": {
            "start": traffic_data['timestamps'][0].isoformat() if traffic_data['timestamps'] else None,
            "end": traffic_data['timestamps'][-1].isoformat() if traffic_data['timestamps'] else None,
            "duration_seconds": len(traffic_data['timestamps'])
        },
        "traffic_data": {
            "timestamps": [ts.isoformat() for ts in traffic_data['timestamps']],
            "incoming_bytes_per_sec": list(traffic_data['incoming']),
            "outgoing_bytes_per_sec": list(traffic_data['outgoing'])
        },
        "statistics": {
            "total_incoming_bytes": sum(traffic_data['incoming']),
            "total_outgoing_bytes": sum(traffic_data['outgoing']),
            "avg_incoming_bytes_per_sec": sum(traffic_data['incoming']) / len(traffic_data['incoming']),
            "avg_outgoing_bytes_per_sec": sum(traffic_data['outgoing']) / len(traffic_data['outgoing']),
            "max_incoming_bytes_per_sec": max(traffic_data['incoming']) if traffic_data['incoming'] else 0,
            "max_outgoing_bytes_per_sec": max(traffic_data['outgoing']) if traffic_data['outgoing'] else 0
        }
    }

    # 轉(zhuǎn)換為JSON字符串
    report_json = json.dumps(report_data, indent=2)

    # 創(chuàng)建響應(yīng)
    from flask import Response
    response = Response(
        report_json,
        mimetype="application/json",
        headers={"Content-Disposition": "attachment;filename=bandwidth_report.json"}
    )

    return response


@app.route('/change-view')
def change_view():
    # 改變數(shù)據(jù)視圖范圍
    range = request.args.get('range', 'minute')

    # 根據(jù)范圍調(diào)整數(shù)據(jù)保留數(shù)量
    if range == 'minute':
        new_maxlen = 60  # 1分鐘
    elif range == 'hour':
        new_maxlen = 3600  # 1小時
    elif range == 'day':
        new_maxlen = 86400  # 24小時
    else:
        new_maxlen = 300  # 默認5分鐘

    # 創(chuàng)建新的deque并復(fù)制現(xiàn)有數(shù)據(jù)
    def resize_deque(old_deque, new_maxlen):
        new_deque = deque(maxlen=new_maxlen)
        for item in old_deque:
            new_deque.append(item)
        return new_deque

    traffic_data['incoming'] = resize_deque(traffic_data['incoming'], new_maxlen)
    traffic_data['outgoing'] = resize_deque(traffic_data['outgoing'], new_maxlen)
    traffic_data['timestamps'] = resize_deque(traffic_data['timestamps'], new_maxlen)

    return jsonify({"status": "success", "new_maxlen": new_maxlen})


if __name__ == '__main__':
    # 啟動流量監(jiān)控線程
    monitor.start()

    # 啟動Flask應(yīng)用
    print("啟動帶寬監(jiān)控Web界面...")
    print("請訪問 http://localhost:5000")
    app.run(debug=True, host='0.0.0.0', port=5000, use_reloader=False)

以上就是基于Python實現(xiàn)Windows帶寬監(jiān)控工具的詳細內(nèi)容，更多關(guān)于Python帶寬監(jiān)控的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: