实时网络速度监控面板：Python Flask + SSE 技术详解

技术架构与工具选择

Python Flask 作为轻量级 Web 框架，适合快速构建后端服务。SSE（Server-Sent Events）是一种服务器推送技术，允许服务器主动向客户端发送实时数据。结合 speedtest-cli 库可实现网络速度测试功能。

环境准备

安装必要依赖库：

pip install flask speedtest-cli

后端实现（Flask + SSE）

创建 Flask 应用并配置 SSE 路由：

from flask import Flask, Response, render_template
import speedtest
import json
import time

app = Flask(__name__)

def speed_test():
    st = speedtest.Speedtest()
    st.get_best_server()
    download = st.download() / 10**6  # 转换为 Mbps
    upload = st.upload() / 10**6
    return download, upload

@app.route('/stream')
def stream():
    def event_stream():
        while True:
            download, upload = speed_test()
            data = {
                'download': round(download, 2),
                'upload': round(upload, 2),
                'timestamp': int(time.time() * 1000)
            }
            yield f"data: {json.dumps(data)}\n\n"
            time.sleep(60)  # 每分钟测试一次
    return Response(event_stream(), mimetype="text/event-stream")

@app.route('/')
def index():
    return render_template('index.html')

前端实现（HTML + JavaScript）

创建 templates/index.html 文件：

<!DOCTYPE html>
<html>
<head>
    <title>实时网络速度监控</title>
    <script src="https://cdn.jsdelivr.net/npm/chart.js"></script>
</head>
<body>
    <div style="width: 80%; margin: auto;">
        <h1>实时网络速度监控</h1>
        <canvas id="speedChart"></canvas>
    </div>

    <script>
        const ctx = document.getElementById('speedChart').getContext('2d');
        const chart = new Chart(ctx, {
            type: 'line',
            data: {
                datasets: [
                    { label: '下载速度 (Mbps)', borderColor: 'rgb(75, 192, 192)', data: [] },
                    { label: '上传速度 (Mbps)', borderColor: 'rgb(255, 99, 132)', }
                ]
            },
            options: {
                scales: { x: { type: 'realtime', realtime: { delay: 2000 } } }
            }
        });

        const eventSource = new EventSource('/stream');
        eventSource.onmessage = (e) => {
            const data = JSON.parse(e.data);
            const timestamp = new Date(data.timestamp);
            
            chart.data.datasets[0].data.push({
                x: timestamp,
                y: data.download
            });
            
            chart.data.datasets[1].data.push({
                x: timestamp,
                y: data.upload
            });
            
            chart.update('quiet');
        };
    </script>
</body>
</html>

关键功能解析

SSE 实现原理：通过 text/event-stream 内容类型建立长连接，服务器持续发送 前缀的格式化消息。前端通过EventSource` API 监听并处理数据。

动态图表更新：使用 Chart.js 的实时刻度插件（需额外引入 chartjs-adapter-date-fns），通过 chart.update() 方法动态刷新数据。

部署与优化

生产环境建议：

  • 使用 Gunicorn 或 uWSGI 部署 Flask 应用
  • 添加 Nginx 反向代理
  • 实现 SSE 连接失败后的自动重连机制
  • 增加前端数据缓存限制防止内存溢出
// 前端优化示例
eventSource.onerror = () => {
    setTimeout(() => location.reload(), 5000);
};

扩展功能方向

  1. 历史数据存储：集成 SQLite 或 PostgreSQL 记录测试结果
  2. 多节点监控：部署多个测试节点并聚合数据
  3. 报警机制：当速度低于阈值时触发邮件/短信通知
  4. 移动端适配：使用响应式设计优化移动端展示

完整项目结构

project/
├── app.py
├── templates/
│   └── index.html
└── requirements.txt

通过上述实现，一个具备实时数据推送、动态可视化的网络速度监控系统即可完成。测试时运行 flask run 命令访问本地端口即可查看实时图表。

# 技术问答 # 人工智能 # AI智能体
Logo
HarmonyOS开发者社区

讨论HarmonyOS开发技术，专注于API与组件、DevEco Studio、测试、元服务和应用上架分发等。

更多推荐

cover

Kafka 深度剖析：架构演进、核心概念与设计精髓

avatar HarmonyOS开发者社区
cover

鸿蒙学习实战之路-Web 页面适配最佳实践

avatar HarmonyOS开发者社区
cover

【鸿蒙2025领航者闯关】鸿蒙2025：从技术突破到生态共建，开发者的成长与蜕变

avatar HarmonyOS开发者社区