Python 工具： Windows 带宽监控工具
Python 工具： Windows 带宽监控工具
- 环境
- 介绍
- 会使用的库
- 多线程关键代码：
- 系统流量采集：用 psutil 获取网络数据
- 概念：网络流量的“增量”与“总量”
- 代码中的流量采集逻辑
- Flask Web框架：搭建后端服务
- 前端部分
- 交互逻辑（JavaScript）
- 1. fetch()：异步请求后端数据
- 2. Chart.js：绘制实时图表
- 3. 定时更新：setInterval()
- 4. 辅助函数：格式转换
- 完整代码
- 完整 index 代码：
- 完整后端代码
环境
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 的带宽监控工具，这个工具融合了Python后端开发、多线程、系统信息采集、Web框架、前端可视化等多个技术领域。
源码可以看开源地址：
https://gitee.com/daolizhe/python_tool/tree/master/Windows%E5%B8%A6%E5%AE%BD%E7%9B%91%E6%8E%A7%E5%B7%A5%E5%85%B7
会使用的库
| 库名 | 核心作用 | 代码中的应用场景 |
|---|---|---|
| threading | 实现多线程编程，让程序“同时做多个任务” | 创建TrafficMonitor监控线程，与FlaskWeb服务并行运行 |
| time | 提供时间相关功能（延迟、时间戳等） | time.sleep(1)实现每秒采集一次流量数据 |
| datetime | 处理日期和时间，提供更丰富的时间格式 | 记录流量数据对应的时间戳（如datetime.now()） |
| collections.deque | 双端队列，支持固定长度（自动丢弃旧数据） | 存储最近5分钟的流量数据（maxlen=300，每秒1条共300秒） |
| io | 内存中的输入输出流，模拟文件操作 | 存储Matplotlib生成的图表（避免写入本地文件） |
| base64 | 将二进制数据编码为文本格式 | 把图表二进制数据转为base64，方便前端HTML显示 |
| json | 处理JSON数据（Python字典与JSON字符串互转） | 生成JSON格式的流量报告，供前端下载 |
| os | 与操作系统交互（本代码中未实际使用，预留扩展） | 可用于获取系统信息、管理文件路径等 |
多线程关键代码：
class TrafficMonitor(threading.Thread):def __init__(self, interface_name=None):super().__init__() # 调用父类构造函数self.daemon = True # 设置为守护线程self.running = True # 控制线程循环的开关def run(self):# 1. 自动选择网络接口（优先非回环接口，且有发送数据的接口）# 2. 初始化上次流量统计值（用于计算每秒增量）# 3. 循环采集数据：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端调用）traffic_data['incoming'].append(incoming_rate)time.sleep(1) # 每秒采集一次def stop(self):self.running = False # 关闭循环，线程退出
系统流量采集：用 psutil 获取网络数据
概念：网络流量的“增量”与“总量”
- 总量（bytes_recv/bytes_sent）：从系统启动到当前，网络接口接收/发送的总字节数（不会重置）。
- 增量（每秒流量）：当前总量 - 上次总量，即每秒的实际流量（如“100KB/s”）。
代码中的流量采集逻辑
- 选择网络接口：
若未指定接口（如TrafficMonitor()），代码会自动遍历所有接口，排除回环接口（lo，本地测试用），选择有数据发送的接口（bytes_sent > 0）。 - 初始化上次总量：
self.last_incoming = interfaces[self.interface_name].bytes_recv，记录初始总量。 - 计算每秒增量：
每次循环中，用当前总量减去上次总量，得到每秒流量（如incoming_rate），再更新上次总量。 - 存储数据：
将每秒流量和对应的时间戳存入全局变量traffic_data（deque类型，自动保留最近300条）。
Flask Web框架：搭建后端服务
Flask是轻量级Web框架，
这部分有不明白的看下面链接地址
https://blog.csdn.net/qq_36051316/article/details/136024400
前端部分
交互逻辑（JavaScript）
前端的核心是“实时获取后端数据并更新页面”，主要通过以下函数实现：
1. fetch()：异步请求后端数据
fetch(‘/traffic-data’)会向后端/traffic-data路由发送请求，获取JSON格式的流量数据，再用这些数据更新图表。
2. Chart.js：绘制实时图表
Chart.js是轻量级前端绘图库，代码中用它绘制入站/出站流量曲线：
- 初始化图表：指定canvas元素、图表类型（line折线图）、初始数据（空）、坐标轴配置。
- 更新图表：每次fetch到新数据后，修改chart.data.labels（时间戳）和chart.data.datasets（流量数据），再调用chart.update()刷新图表。
3. 定时更新：setInterval()
用setInterval实现周期性更新：
- setInterval(updateChart, 1000)：每秒更新一次图表（与后端采集频率一致）。
- setInterval(updateTotalStats, 5000)：每5秒更新一次总流量（总流量变化较慢，无需频繁更新）。
4. 辅助函数：格式转换
- formatBytes()：将字节数（如102400）转为易读格式（如100 KB），支持B/KB/MB/GB/TB。
- formatBytesPerSec()：在formatBytes()基础上添加/s，如100 KB/s。
完整代码
完整 index 代码：
<!DOCTYPE html>
<html>
<head><title>Windows带宽监控</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带宽监控工具</h1><p>实时监控网络接口流量</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>// 创建图表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: '网络流量监控'}}}});// 格式化字节大小为易读格式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];}// 格式化字节/秒为易读格式function formatBytesPerSec(bytes, decimals = 2) {return formatBytes(bytes, decimals) + '/s';}// 更新图表数据function updateChart() {fetch('/traffic-data').then(response => response.json()).then(data => {// 转换数据为KB/sconst 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);}});}// 更新总流量统计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 => {// 重新加载页面数据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 osapp = Flask(__name__)# 全局变量存储流量数据
traffic_data = {'incoming': deque(maxlen=300), # 存储最近5分钟的入站流量(每秒一个数据点)'outgoing': deque(maxlen=300), # 存储最近5分钟的出站流量'timestamps': deque(maxlen=300) # 存储对应的时间戳
}# 流量统计类
class TrafficMonitor(threading.Thread):def __init__(self, interface_name=None):super().__init__()self.daemon = Trueself.last_incoming = 0self.last_outgoing = 0self.interface_name = interface_nameself.running = Truedef run(self):print("开始监控网络流量...")# 获取网络接口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 = namebreakif not self.interface_name:print("未找到可用的网络接口")returnprint(f"监控接口: {self.interface_name}")# 初始化计数器self.last_incoming = interfaces[self.interface_name].bytes_recvself.last_outgoing = interfaces[self.interface_name].bytes_sent# 开始监控循环while self.running:try:# 获取当前流量统计current_stats = psutil.net_io_counters(pernic=True)[self.interface_name]current_incoming = current_stats.bytes_recvcurrent_outgoing = current_stats.bytes_sent# 计算每秒流量incoming_rate = current_incoming - self.last_incomingoutgoing_rate = current_outgoing - self.last_outgoing# 更新计数器self.last_incoming = current_incomingself.last_outgoing = current_outgoing# 更新全局流量数据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"监控出错: {e}")time.sleep(5)def stop(self):self.running = False# 创建并启动流量监控线程
monitor = TrafficMonitor()# Flask路由
@app.route('/')
def index():return render_template('index.html')@app.route('/traffic-data')
def get_traffic_data():# 返回JSON格式的流量数据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():# 获取总流量统计interfaces = psutil.net_io_counters(pernic=True)interface_name = monitor.interface_nametotal_incoming = interfaces[interface_name].bytes_recv if interface_name in interfaces else 0total_outgoing = interfaces[interface_name].bytes_sent if interface_name in interfaces else 0return jsonify({'total_incoming': total_incoming,'total_outgoing': total_outgoing})@app.route('/traffic-plot')
def get_traffic_plot():# 生成流量图表并返回base64编码的图像if not traffic_data['timestamps']:return "暂无数据"plt.figure(figsize=(10, 6))# 转换数据为KB/sincoming_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('实时网络流量监控')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>暂无数据可生成报告</p>"# 计算统计信息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.0return f"{bytes:.2f} TB"def format_bps(bytes_per_sec):return format_bytes(bytes_per_sec) + "/s"report = f"""<h2>带宽使用报告</h2><p>统计时间段: {traffic_data['timestamps'][0]} 到 {traffic_data['timestamps'][-1]}</p><p>监控时长: {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 "暂无数据", 404# 准备报告数据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)# 创建响应from flask import Responseresponse = Response(report_json,mimetype="application/json",headers={"Content-Disposition": "attachment;filename=bandwidth_report.json"})return response@app.route('/change-view')
def change_view():# 改变数据视图范围range = request.args.get('range', 'minute')# 根据范围调整数据保留数量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分钟# 创建新的deque并复制现有数据def resize_deque(old_deque, new_maxlen):new_deque = deque(maxlen=new_maxlen)for item in old_deque:new_deque.append(item)return new_dequetraffic_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__':# 启动流量监控线程monitor.start()# 启动Flask应用print("启动带宽监控Web界面...")print("请访问 http://localhost:5000")app.run(debug=True, host='0.0.0.0', port=5000, use_reloader=False)