Flask 应用中使用 SocketIO 实现实时数据流

使用 Flask 和 SocketIO 实现实时数据流

本示例演示如何在 Flask 应用中使用 SocketIO 实现实时数据流，通过 iperf3 工具收集网络数据并通过 WebSocket 推送至前端页面。

代码示例

import subprocess
import time
import psutil

from flask import Flask, render_template, request, jsonify
from flask_socketio import SocketIO

app = Flask(__name__, template_folder='D:\python\webping')
app.config['SECRET_KEY'] = 'secret_key'
socketio = SocketIO(app, cors_allowed_origins='*')

processes = {}


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


@app.route('/start', methods=['POST'])
def start():
    port = request.json['port']
    if port in processes:
        return jsonify({'message': f'Command already running on port {port}.'})


    process = subprocess.Popen(['iperf3', '-s', '-p', str(port), '-i', '1'], stdout=subprocess.PIPE, universal_newlines=True)
    processes[port] = process

    def stream():
        i = 0
        while process and process.poll() is None:
            output = process.stdout.readline().strip()
            if len(output) > 0:
                i += 1
                print(i, output)

                socketio.emit('stream_output', output)
                # time.sleep(0.1)

        processes.pop(port)


    socketio.start_background_task(stream)
    return jsonify({'message': f'Command started on port {port}.'})


@app.route('/stop', methods=['POST'])
def stop():
    port = request.json['port']
    if port in processes:
        process = processes[port]
        process.terminate()
        processes.pop(port)
        return jsonify({'message': f'Command stopped on port {port}.'})
    else:
        return jsonify({'message': f'No command running on port {port}.'})


if __name__ == '__main__':
    socketio.run(app, host='0.0.0.0', port=5000, debug=False, allow_unsafe_werkzeug=True)

错误分析

错误 "GET /socket.io/?EIO=3&transport=polling&t=Of5PSrp HTTP/1.1" 400 - 通常是由于 SocketIO 连接问题导致的。

解决方法

检查前端代码：确保前端代码中的 Socket 连接地址正确。在 index.html 文件中，检查以下代码段：

var socket = io.connect('http://' + document.domain + ':' + location.port);

确保 document.domain 和 location.port 的值正确，与 Flask 应用运行的地址和端口匹配。

检查 Flask 应用的 SocketIO 初始化：在您的代码中，确保 SocketIO 初始化的方式正确。尝试将以下代码段替换为您的 socketio 初始化代码：

socketio = SocketIO(app, cors_allowed_origins='*')

这将允许来自任何来源的跨域请求。如果您只想允许特定的来源，请将 cors_allowed_origins 参数替换为允许的来源列表。

检查网络连接：确保您的网络连接正常，可以尝试刷新页面或重新启动网络设备。

补充说明

如果以上方法仍然无法解决问题，请提供更多关于错误的详细信息，以便更好地帮助您解决问题。