nat_pagle 2022-02-15 12:47 采纳率: 66.7%
浏览 133
已结题

flask中利用websocket制作一个实时更新的前端

测试的后端弄一个可以供测试的就行（比如在console依次输出1，2，3），前端要一个按钮，点击后可以跟进更新，不需要刷新，显示1，2，3。我能完成测试就采纳了。

  • 写回答

7条回答 默认 最新

  • 王辉辉的猫 2022-02-15 23:35
    关注

    server.py

    
import json, time, random
from flask import Flask, render_template
from flask import request
from flask import make_response
from gevent import pywsgi
from geventwebsocket.handler import WebSocketHandler
from threading import Thread

# 实例化产生一个Flask对象
app = Flask(__name__)
WS_LIST = []
count = 0


def get_host_ip():
    # 获取本机ip
    import socket
    hostname = socket.gethostname()
    ip = socket.gethostbyname(hostname)
    if '127.0.0.1' in ip:
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        try:
            s.connect(('8.8.8.8', 80))
            ip = s.getsockname()[0]
        finally:
            s.close()
    return ip


self_ip = get_host_ip()


# flask的路由是基于装饰器的
@app.route('/', methods=['GET', 'POST'])
def hello_world():
    return render_template("index.html",self_ip=self_ip)


@app.route("/main")
def func():
    global count
    ws = request.environ.get('wsgi.websocket')
    if not ws:
        print('【客户端退出1】', str(ws))
        return make_response("error", 404)
    WS_LIST.append(ws)
    print('len(WS_LIST)', len(WS_LIST))  # 客户端数量

    while True:
        # 等待用户发送消息，并接受
        message = ws.receive()

        # 关闭：message=None
        if not message:
            try:
                print('【客户端退出2】', str(ws))
                WS_LIST.remove(ws)
                del ws
                print('len(WS_LIST)', len(WS_LIST))
            except Exception as e:
                print(e)
            return make_response("error", 404)

        # msg = json.loads(message)
        # print(msg)

        for item in WS_LIST:
            res = json.dumps({'data': count})
            item.send(res)


if __name__ == '__main__':
    def foo():
        global count
        while True:
            time.sleep(1)
            count += 1
            print(count)


    t1 = Thread(target=foo)  # 创建线程对象
    t1.start()  # 启动线程

    server = pywsgi.WSGIServer(('0.0.0.0', 6000), app, handler_class=WebSocketHandler)
    print(f'server start http://{self_ip}:6000')
    server.serve_forever()

    <!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>text</title>
</head>
<body>
<input type="button" value="按钮" onclick="test()"/>
<div id="one"></div>
<script>
    function test() {
        let self_ip = '{{self_ip|safe }}'
        let url = "ws://" + self_ip + ":6000/main"
        console.log(url)
        let ws = new WebSocket(url);
        ws.onmessage = function (event) {
            document.getElementById('one').innerHTML = event.data;
            setTimeout(function () {
                ws.send(JSON.stringify({'hello': 'world'}));
            }, 1000)
        };
        ws.onopen = function (evt) {
            console.log("Websocket open success", url);
            ws.send(JSON.stringify({'hello': 'world'}));
        };
        ws.onclose = function (evt) {
            console.log("Connection closed.", evt);
            ws.close();
        };
    }

</script>
</body>
</html>

    打开多少个客户端都没问题
    Flask==1.1.1
    Flask-Sockets ==0.2.1
    gevent==21.8.0
    gevent-websocket==0.10.1

    评论 编辑记录

报告相同问题？

问题事件

  • 系统已结题 2月23日
  • 创建了问题 2月15日

悬赏问题

  • ¥15 在虚拟机中安装flash code
  • ¥15 单片机stm32f10x编写光敏电阻调节3.3伏大功率灯亮度（光强越大灯越暗，白天正常光强灯不亮，使用ADC，PWM等模块）望各位找一下错误或者提供一个可实现功能的代码
  • ¥20 verilog状态机方法流水灯
  • ¥15 pandas代码实现不了意图
  • ¥15 GD32H7 从存储器到外设SPI传输数据无法重复启用DMA
  • ¥25 LT码在高斯信道下的误码率仿真
  • ¥45 渲染完成之后将物体的材质贴图改变，自动化进行这个操作
  • ¥15 yolov5目标检测并显示目标出现的时间或视频帧
  • ¥15 电视版的优酷可以设置电影连续播放吗？
  • ¥50 复现论文；matlab代码编写