flask-socketio實現(xiàn)前后端實時通信的功能的示例

更新時間：2023年04月21日 09:28:54 作者：今天上上簽

本文主要介紹了flask-socketio實現(xiàn)前后端實時通信的功能的示例，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧

一、需求

準備實現(xiàn)一個前后端實時發(fā)送消息的功能，即后端發(fā)送一條消息，前端就可以接收到該消息并顯示在頁面上。

二、解決方法

利用Flask-SocketIO實現(xiàn)，F(xiàn)lask-SocketIO使Flask應用程序能夠訪問客戶端和服務器之間的低延遲雙向通信。能夠很好的解決我們的問題

三、具體實現(xiàn)

步驟一:安裝

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':
        # 模擬執(zhí)行你的業(yè)務邏輯一
        time.sleep(5)
        # 利用全局變量表明業(yè)務邏輯一已完成
        flag1 = 1
        print("步驟1完成")
        # 模擬執(zhí)行你的業(yè)務邏輯二
        time.sleep(2)
        # 利用全局變量表明業(yè)務邏輯二已完成
        flag2 = 1
        print("步驟2完成")
        # 模擬執(zhí)行你的業(yè)務邏輯三
        time.sleep(3)
        # 利用全局變量表明業(yè)務邏輯三已完成
        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連接，則開啟一個線程，專門給前端發(fā)送消息
            thread = socketio.start_background_task(target=background_thread)


@socketio.on('disconnect', namespace='/test_conn')
def disconnect_msg():
    """
    此函數在socket斷開時被調用
    """
    print('client disconnected!')


def background_thread():
    """
    該線程專門用來給前端發(fā)送消息
    :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環(huán)境
    new Vue({
        el: '#test',
        data() {return{} },
        methods: {
            onSubmit() {
            	// 向后端發(fā)送post請求
                axios.post('http://127.0.0.1:5000/');
                // 建立socket連接，此時會觸發(fā)后端的@socketio.on('connect', namespace='/test_conn')路由，執(zhí)行test_connect函數
                let socket = io.connect('http://127.0.0.1:5000/test_conn');
                socket.on('server_response', function (res) {
                	//接收到后端發(fā)送過來的消息
                    var t = res.data;
                    console.log(t)
                    if (t) {
                    	// 顯示在頁面上
                        $("#t").append(t).append('<br/>');
                    }
                });
            }
        }
    })
</script>
</html>

結果展示

啟動flask項目，并訪問：

點擊提交后，前端發(fā)送post請求給后端，并建立socket連接：

同時執(zhí)行index()函數：

同時前端頁面接收到后端發(fā)送的消息：

至此，前后端交互功能完成！

四、注意事項

1、我們不能直接在socket連接函數中寫一個while循環(huán)去發(fā)送信息，例如這樣：

@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函數每次都會執(zhí)行，理論上客戶端應該可以定時收到服務端的隨機數。但是結果是客戶端根本接收不到，連soketio.on函數都沒有觸發(fā)運行。
原因應該是當服務端陷入死循環(huán)，會影響與客戶端之間的websocket連接，總之寫while true需謹慎。
所以，在flask_socketio的示例程序中，我們 用后臺線程 進行while循環(huán)以解決這個問題。

2、websocket連接的一個典型應用是實現(xiàn)一個網絡聊天室，所以就會有群發(fā)和私聊的功能，我們此處演示的示例，是一個群發(fā)功能，也就是我們并沒有指定socket的room空間，所以當我們從后端發(fā)送給前端消息時，所有打開此頁面的人，都可以看到我們發(fā)送的消息，也就是一個群發(fā)功能。

到此這篇關于flask-socketio實現(xiàn)前后端實時通信的功能的示例的文章就介紹到這了,更多相關flask-socketio實時通信內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: