文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

flask-socketio实现前后端实时通信的功能的示例

2023-05-17 05:07

关注

一、需求

准备实现一个前后端实时发送消息的功能,即后端发送一条消息,前端就可以接收到该消息并显示在页面上。

二、解决方法

利用Flask-SocketIO实现,Flask-SocketIO使Flask应用程序能够访问客户端和服务器之间的低延迟双向通信。能够很好的解决我们的问题

三、具体实现

步骤一:安装

pip install eventlet
pip install flask-socketio

注意:flask-socketio对版本要求比较严格,如果报错,请更换如下包的版本:

Flask 1.1.4
Flask-SocketIO 4.3.1
python-engineio 3.13.2
python-socketio 4.6.0

步骤二:服务器代码

import time
from flask import Flask, request, render_template
from flask_socketio import SocketIO
from threading import Lock

thread = None
thread_lock = Lock()
app = Flask(__name__)
socketio = SocketIO()
socketio.init_app(app, cors_allowed_origins='*') # 用socketio初始化flask的app

flag1 = 0
flag2 = 0
flag3 = 0


@app.route('/', methods=['POST', 'GET'])
def index():
    global flag1
    global flag2
    global flag3
    if request.method == 'POST':
        # 模拟执行你的业务逻辑一
        time.sleep(5)
        # 利用全局变量表明业务逻辑一已完成
        flag1 = 1
        print("步骤1完成")
        # 模拟执行你的业务逻辑二
        time.sleep(2)
        # 利用全局变量表明业务逻辑二已完成
        flag2 = 1
        print("步骤2完成")
        # 模拟执行你的业务逻辑三
        time.sleep(3)
        # 利用全局变量表明业务逻辑三已完成
        flag3 = 1
        print("步骤3完成")

    return render_template('t1.html')


@socketio.on('connect', namespace='/test_conn')
def test_connect():
    """
    此函数在建立socket连接时被调用
    """
    print("socket 建立连接")
    global thread
    with thread_lock:
        print(thread)
        if thread is None:
            # 如果socket连接,则开启一个线程,专门给前端发送消息
            thread = socketio.start_background_task(target=background_thread)


@socketio.on('disconnect', namespace='/test_conn')
def disconnect_msg():
    """
    此函数在socket断开时被调用
    """
    print('client disconnected!')


def background_thread():
    """
    该线程专门用来给前端发送消息
    :return:
    """
    global flag1
    global flag2
    global flag3
    while True:
        if flag1 == 1:
            print("进入判断一")
            socketio.emit('server_response',
                          {'data': '步骤一已完成'}, namespace='/test_conn')
            flag1 = 2
        socketio.sleep(2)
        if flag2 == 1 and flag1 == 2:
            print("进入判断二")
            socketio.emit('server_response',
                          {'data': '步骤二已完成'}, namespace='/test_conn')
            flag2 = 2
        socketio.sleep(2)
        if flag3 == 1 and flag2 == 2 and flag1 == 2:
            print("进入判断三")
            socketio.emit('server_response',
                          {'data': '步骤三已完成'}, namespace='/test_conn')
            flag3 = 2
        socketio.sleep(2)


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

步骤三:前端代码

前端代码是用vue+element ui编写。(代码中所涉及的js文件可以在百度上自行下载,有很多)

<!DOCTYPE html>
<html>
<head>
    <meta charset="UTF-8" />
    <meta name="viewport" content="width=device-width,initial-scale=1.0" />
    <link rel="stylesheet" href="static/css/index.css" rel="external nofollow" > <!-- 引入element的css文件-->
    <script src="static/js/vue.js"></script> <!-- 引入vue文件-->
    <script src="static/js/index.js"></script> <!-- 引入element的js文件-->
    <script src="static/js/axios.min.js"></script> <!--引入axios文件--> 
    <script type="text/javascript" src="static/js/jquery-1.4.2.min.js"></script> <!--引入jquery文件--> 
    <script type="text/javascript" src="static/js/socket.io.min.js"></script> <!--引入socket文件--> 
</head>
<body>
    <div id="test">
        <el-button type="primary" size="small" @click="onSubmit()">提交</el-button>
        <div id = "t"></div>
    </div>
</body>
<script>
	// 新建vue环境
    new Vue({
        el: '#test',
        data() {return{} },
        methods: {
            onSubmit() {
            	// 向后端发送post请求
                axios.post('http://127.0.0.1:5000/');
                // 建立socket连接,此时会触发后端的@socketio.on('connect', namespace='/test_conn')路由,执行test_connect函数
                let socket = io.connect('http://127.0.0.1:5000/test_conn');
                socket.on('server_response', function (res) {
                	//接收到后端发送过来的消息
                    var t = res.data;
                    console.log(t)
                    if (t) {
                    	// 显示在页面上
                        $("#t").append(t).append('<br/>');
                    }
                });
            }
        }
    })
</script>
</html>

结果展示

启动flask项目,并访问:

点击提交后,前端发送post请求给后端,并建立socket连接:

同时执行index()函数:

同时前端页面接收到后端发送的消息:

至此,前后端交互功能完成!

四、注意事项

1、我们不能直接在socket连接函数中写一个while循环去发送信息,例如这样:

@socketio.on('connect', namespace='/test_conn')
def test_connect():
    while True:
        socketio.sleep(5)
        t = random.randint(1, 100)
        socketio.emit('server_response',
                      {'data': t},namespace='/test_conn')

虽然这样写,emit函数每次都会执行,理论上客户端应该可以定时收到服务端的随机数。但是结果是客户端根本接收不到,连soketio.on函数都没有触发运行。
原因应该是当服务端陷入死循环,会影响与客户端之间的websocket连接,总之写while true需谨慎。
所以,在flask_socketio的示例程序中,我们 用后台线程 进行while循环以解决这个问题。

2、websocket连接的一个典型应用是实现一个网络聊天室,所以就会有群发和私聊的功能,我们此处演示的示例,是一个群发功能,也就是我们并没有指定socket的room空间,所以当我们从后端发送给前端消息时,所有打开此页面的人,都可以看到我们发送的消息,也就是一个群发功能。

到此这篇关于flask-socketio实现前后端实时通信的功能的示例的文章就介绍到这了,更多相关flask-socketio实时通信内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网!

阅读原文内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯