基于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框架、前端可視化等多個技術領域。
源碼可以看開源地址
會使用的庫
| 庫名 | 核心作用 | 代碼中的應用場景 |
|---|---|---|
| threading | 實現(xiàn)多線程編程，讓程序“同時做多個任務” | 創(chuàng)建TrafficMonitor監(jiān)控線程，與FlaskWeb服務并行運行 |
| time | 提供時間相關功能（延遲、時間戳等） | time.sleep(1)實現(xiàn)每秒采集一次流量數(shù)據(jù) |
| datetime | 處理日期和時間，提供更豐富的時間格式 | 記錄流量數(shù)據(jù)對應的時間戳（如datetime.now()） |
| collections.deque | 雙端隊列，支持固定長度（自動丟棄舊數(shù)據(jù)） | 存儲最近5分鐘的流量數(shù)據(jù)（maxlen=300，每秒1條共300秒） |
| io | 內(nèi)存中的輸入輸出流，模擬文件操作 | 存儲Matplotlib生成的圖表（避免寫入本地文件） |
| base64 | 將二進制數(shù)據(jù)編碼為文本格式 | 把圖表二進制數(shù)據(jù)轉為base64，方便前端HTML顯示 |
| json | 處理JSON數(shù)據(jù)（Python字典與JSON字符串互轉） | 生成JSON格式的流量報告，供前端下載 |
| os | 與操作系統(tǒng)交互（本代碼中未實際使用，預留擴展） | 可用于獲取系統(tǒng)信息、管理文件路徑等 |
多線程關鍵代碼
class TrafficMonitor(threading.Thread):
def __init__(self, interface_name=None):
super().__init__() # 調(diào)用父類構造函數(shù)
self.daemon = True # 設置為守護線程
self.running = True # 控制線程循環(huán)的開關
def run(self):
# 1. 自動選擇網(wǎng)絡接口（優(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 # 關閉循環(huán)，線程退出
系統(tǒng)流量采集：用 psutil 獲取網(wǎng)絡數(shù)據(jù)
概念：網(wǎng)絡流量的“增量”與“總量”
總量（bytes_recv/bytes_sent）：從系統(tǒng)啟動到當前，網(wǎng)絡接口接收/發(fā)送的總字節(jié)數(shù)（不會重置）。
增量（每秒流量）：當前總量 - 上次總量，即每秒的實際流量（如“100KB/s”）。
代碼中的流量采集邏輯
選擇網(wǎng)絡接口：若未指定接口（如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ù)：將每秒流量和對應的時間戳存入全局變量traffic_data（deque類型，自動保留最近300條）。
Flask Web框架：搭建后端服務
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ù)：格式轉換
- formatBytes()：將字節(jié)數(shù)（如102400）轉為易讀格式（如100 KB），支持B/KB/MB/GB/TB。
- formatBytesPerSec()：在formatBytes()基礎上添加/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)絡接口流量</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">總入站流量</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)絡流量監(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 => {
// 轉換數(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) # 存儲對應的時間戳
}
# 流量統(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)絡流量...")
# 獲取網(wǎng)絡接口
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)絡接口")
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))
# 轉換數(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)絡流量監(jiān)控')
plt.legend()
plt.grid(True)
plt.gcf().autofmt_xdate()
plt.gca().xaxis.set_major_formatter(mdates.DateFormatter('%H:%M:%S'))
# 將圖表轉換為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'])
# 轉換為更友好的單位
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>總入站流量: {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
}
}
# 轉換為JSON字符串
report_json = json.dumps(report_data, indent=2)
# 創(chuà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并復制現(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應用
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)容，更多關于Python帶寬監(jiān)控的資料請關注腳本之家其它相關文章！
相關文章
這篇文章主要介紹了Python基礎練習之用戶登錄實現(xiàn)代碼分享，還是比較不錯的，這里分享給大家，供需要的朋友參考。2017-11-11
numpy.random.shuffle打亂順序函數(shù)的實現(xiàn)
這篇文章主要介紹了numpy.random.shuffle打亂順序函數(shù)的實現(xiàn)，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2019-09-09
這篇文章主要介紹了Python去除html標簽的幾種方法總結，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2023-01-01
Python利用xlrd?與?xlwt?模塊操作?Excel
這篇文章主要介紹了Python利用xlrd?與?xlwt?模塊操作?Excel，文章圍繞主題展開詳細的內(nèi)容，具有一定的參考價值，需要的小伙伴可以參考一下2022-05-05
這篇文章主要為大家詳細介紹了Python tkinter實現(xiàn)日期選擇器，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2021-02-02
最近在用Python,做一個小腳本,有個操作就是要把內(nèi)容換成數(shù)組對象再進行相關操作,下面這篇文章主要給大家介紹了關于如何在Python中將字符串轉換為數(shù)組的相關資料,需要的朋友可以參考下2022-12-12
淺談Tensorflow 動態(tài)雙向RNN的輸出問題
今天小編就為大家分享一篇淺談Tensorflow 動態(tài)雙向RNN的輸出問題，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-01-01
fabric 是一個python包 是一個基于ssh的部署工具包，這篇文章主要介紹了python三大神器之fabric,需要的朋友可以參考下2019-06-06
這篇文章主要介紹了python之擬合的實現(xiàn)，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2019-07-07
利用Python實時獲取steam特惠游戲數(shù)據(jù)
Steam是由美國電子游戲商Valve于2003年9月12日推出的數(shù)字發(fā)行平臺，被認為是計算機游戲界最大的數(shù)碼發(fā)行平臺之一。本文將利用Python實時獲取steam特惠游戲數(shù)據(jù)，感興趣的可以嘗試一下2022-06-06