Flask-SocketIO 使 Flask 應用程序能夠訪問客戶端和服務器之間的低延遲雙向通信。客戶端應用程序可以使用 Javascript，C ++，Java 和 Swift 中的任何 SocketIO 官方客戶端庫或任何兼容的客戶端來建立與服務器的永久連接。

安裝

直接使用 pip 來安裝：

pip install flask-socketio 復制代碼

要求

Flask-SocketIO 兼容 Python 2.7 和 Python 3.3+。可以從以下三個選項中選擇此程序包所依賴的異步服務：

• eventlet 性能最佳，支持長輪詢和 WebSocket 傳輸。
• gevent 在許多不同的配置中得到支持。gevent 包完全支持長輪詢傳輸，但與 eventlet 不同，gevent 沒有本機 WebSocket 支持。要添加對 WebSocket 的支持，目前有兩種選擇：安裝 gevent-websocket 包為 gevent 增加 WebSocket 支持，或者可以使用帶有 WebSocket 功能的 uWSGI Web 服務器。gevent 的使用也是一種高性能選項，但略低于 eventlet。
• 也可以使用基于 Werkzeug 的 Flask 開發服務器，但需要注意的是，它缺乏其他兩個選項的性能，因此它只應用于簡單的開發環境。此選項僅支持長輪詢傳輸。

擴展會根據安裝的內容自動檢測要使用的異步框架。優先考慮 eventlet，然后是 gevent。對于 gevent 中的WebSocket 支持，首選 uWSGI，然后是 gevent-websocket。如果既未安裝 eventlet 也未安裝 gevent，則使用 Flask 開發服務器。

如果使用多個進程，則進程使用消息隊列服務來協調諸如廣播之類的操作。支持的隊列是 Redis，RabbitMQ以及 Kombu 軟件包支持的任何其他消息隊列 。

在客戶端，官方 Socket.IO Javascript 客戶端庫可用于建立與服務器的連接。還有使用 Swift，Java 和 C ++ 編寫的官方客戶端。非官方客戶端也可以工作，只要它們實現 Socket.IO協議。

初始化

以下代碼示例演示如何將 Flask-SocketIO 添加到 Flask 應用程序：

from flask import Flask, render_template from flask_socketio import SocketIOapp = Flask(__name__) app.config['SECRET_KEY'] = 'secret!' socketio = SocketIO(app)if __name__ == '__main__':socketio.run(app, host='0.0.0.0', debug=True) 復制代碼

以上代碼即完成了一個簡單的 Web 服務器。

socketio.run()函數封裝了 Web 服務器的啟動，并替換了app.run()標準的 Flask 開發服務器啟動。

當應用程序處于調試模式時，Werkzeug 開發服務器仍然在內部使用和配置正確socketio.run()。

在生產模式中，如果可用，則使用 eventlet Web 服務器，否則使用 gevent Web 服務器。如果未安裝 eventlet 和gevent，則使用 Werkzeug 開發 Web 服務器。

基于 Flask 0.11 中引入的單擊的命令行界面。此擴展提供了適用于啟動 Socket.IO 服務器的新版本命令。用法示例：flask run

$ FLASK_APP=my_app.py flask run 復制代碼

或者直接使用下面方式，也可以啟動項目：

$ python2.7 app.py 復制代碼

連接事件

Flask-SocketIO 調度連接和斷開事件。以下示例顯示如何為它們注冊處理程序：

@socketio.on('connect', namespace='/test') def test_connect():emit('my response', {'data': 'Connected'})@socketio.on('disconnect', namespace='/test') def test_disconnect():print('Client disconnected') 復制代碼

連接事件處理程序可以選擇返回False以拒絕連接。這樣就可以在此時對客戶端進行身份驗證。

請注意，連接和斷開連接事件將在使用的每個命名空間上單獨發送。

接收消息

使用 SocketIO 時，雙方都會將消息作為事件接收。在客戶端使用 Javascript 回調。使用 Flask-SocketIO，服務器需要為這些事件注冊處理程序，類似于視圖函數處理路由的方式。

以下示例為未命名的事件創建服務器端事件處理程序：

@socketio.on('message') def handle_message(message):print('received message: ' + message) 復制代碼

上面的示例使用字符串消息。另一種類型的未命名事件使用 JSON 數據：

@socketio.on('json') def handle_json(json):print('received json: ' + str(json)) 復制代碼

最靈活的方式是使用自定義事件名稱，在開發過程中最常用的也是這種方式。

事件的消息數據可以是字符串，字節，整數或 JSON：

@socketio.on('my event') def handle_my_custom_event(json):print('received json: ' + str(json)) 復制代碼

自定義命名事件也可以支持多個參數：

@socketio.on('my event') def handle_my_custom_event(arg1, arg2, arg3):print('received args: ' + arg1 + arg2 + arg3) 復制代碼

Flask-SocketIO 支持 SocketIO 命名空間，允許客戶端在同一物理套接字上復用多個獨立連接：

@socketio.on('my event', namespace='/test') def handle_my_custom_namespace_event(json):print('received json: ' + str(json)) 復制代碼

如果未指定名稱空間，'/'則使用具有名稱的默認全局名稱空間 。

對于裝飾器語法不方便的情況，on_event可以使用該方法：

def my_function_handler(data):passsocketio.on_event('my event', my_function_handler, namespace='/test') 復制代碼

客戶端可以請求確認回叫，確認收到他們發送的消息。處理函數返回的任何值都將作為回調函數中的參數傳遞給客戶端：

@socketio.on('my event') def handle_my_custom_event(json):print('received json: ' + str(json))return 'one', 2 復制代碼

在上面的示例中，將使用兩個參數調用客戶端回調函數，'one'和2。如果處理程序函數未返回任何值，則將調用客戶端回調函數而不帶參數。

發送消息

如上一節所示定義的 SocketIO 事件處理程序可以使用send()和emit() 函數將回復消息發送到連接的客戶端。

以下示例將收到的事件退回給發送它們的客戶端：

from flask_socketio import send, emit@socketio.on('message') def handle_message(message):send(message)@socketio.on('json') def handle_json(json):send(json, json=True)@socketio.on('my event') def handle_my_custom_event(json):emit('my response', json) 復制代碼

注意如何send()和emit()分別用于無名和命名事件。

當有命名空間的工作，send()并emit()默認使用傳入消息的命名空間。可以使用可選namespace參數指定不同的命名空間：

@socketio.on('message') def handle_message(message):send(message, namespace='/chat')@socketio.on('my event') def handle_my_custom_event(json):emit('my response', json, namespace='/chat') 復制代碼

要發送具有多個參數的事件，請發送元組：

@socketio.on('my event') def handle_my_custom_event(json):emit('my response', ('foo', 'bar', json), namespace='/chat') 復制代碼

SocketIO 支持確認回調，確認客戶端收到了一條消息：

def ack():print 'message was received!'@socketio.on('my event') def handle_my_custom_event(json):emit('my response', json, callback=ack) 復制代碼

使用回調時，Javascript 客戶端會收到一個回調函數，以便在收到消息時調用。客戶端應用程序調用回調函數后，服務器將調用相應的服務器端回調。如果使用參數調用客戶端回調，則這些回調也作為服務器端回調的參數提供。

廣播

SocketIO 的另一個非常有用的功能是廣播消息。SocketIO 支持通過此功能broadcast=True可選參數send()和emit()：

@socketio.on('my event') def handle_my_custom_event(data):emit('my response', data, broadcast=True) 復制代碼

在啟用廣播選項的情況下發送消息時，連接到命名空間的所有客戶端都會接收它，包括發件人。如果未使用名稱空間，則連接到全局名稱空間的客戶端將收到該消息。請注意，不會為廣播消息調用回調。

在此處顯示的所有示例中，服務器響應客戶端發送的事件。但對于某些應用程序，服務器需要是消息的發起者。這對于向客戶端發送通知在服務器中的事件（例如在后臺線程中）非常有用。socketio.send()和socketio.emit()方法可用于廣播到所有連接的客戶端：

def some_function():socketio.emit('some event', {'data': 42}) 復制代碼

請注意，socketio.send()與socketio.emit()在上下文理解上和send()與emit()功能不同。另請注意，在上面的用法中沒有客戶端上下文，因此broadcast=True是默認的，不需要指定。

房間

對于許多應用程序，有必要將用戶分組為可以一起尋址的子集。最好的例子是具有多個房間的聊天應用程序，其中用戶從他們所在的房間接收消息，而不是從其他用戶所在的其他房間接收消息。SocketIO 支持通過房間的概念join_room()和leave_room()功能：

from flask_socketio import join_room, leave_room@socketio.on('join') def on_join(data):username = data['username']room = data['room']join_room(room)send(username + ' has entered the room.', room=room)@socketio.on('leave') def on_leave(data):username = data['username']room = data['room']leave_room(room)send(username + ' has left the room.', room=room) 復制代碼

在send()和emit()函數接受一個可選room導致被發送到所有的都在定房客戶端的消息的說法。

所有客戶端在連接時都會被分配一個房間，以連接的會話ID命名，可以從中獲取request.sid。給定的客戶可以加入任何房間，可以給出任何名稱。當客戶端斷開連接時，它將從其所在的所有房間中刪除。無上下文socketio.send() 和socketio.emit()函數也接受一個room參數，以廣播給房間中的所有客戶端。

由于為所有客戶端分配了個人房間，為了向單個客戶端發送消息，客戶端的會話 ID 可以用作房間參數。

錯誤處理

Flask-SocketIO還可以處理異常：

@socketio.on_error() # Handles the default namespace def error_handler(e):pass@socketio.on_error('/chat') # handles the '/chat' namespace def error_handler_chat(e):pass@socketio.on_error_default # handles all namespaces without an explicit error handler def default_error_handler(e):pass 復制代碼

錯誤處理函數將異常對象作為參數。

還可以使用request.event變量檢查當前請求的消息和數據參數，這對于事件處理程序外部的錯誤記錄和調試很有用：

from flask import request@socketio.on("my error event") def on_my_event(data):raise RuntimeError()@socketio.on_error_default def default_error_handler(e):print(request.event["message"]) # "my error event"print(request.event["args"]) # (data,) 復制代碼

基于類的命名空間

作為上述基于裝飾器的事件處理程序的替代，屬于命名空間的事件處理程序可以創建為類的方法。flask_socketio.Namespace作為基類提供，用于創建基于類的命名空間：

from flask_socketio import Namespace, emitclass MyCustomNamespace(Namespace):def on_connect(self):passdef on_disconnect(self):passdef on_my_event(self, data):emit('my_response', data)socketio.on_namespace(MyCustomNamespace('/test')) 復制代碼

使用基于類的命名空間時，服務器接收的任何事件都將調度到名為帶有on_前綴的事件名稱的方法。例如，事件my_event將由名為的方法處理on_my_event。如果收到的事件沒有在命名空間類中定義的相應方法，則忽略該事件。基于類的命名空間中使用的所有事件名稱必須使用方法名稱中合法的字符。

為了方便在基于類的命名空間中定義的方法，命名空間實例包括類中的幾個方法的版本，flask_socketio.SocketIO當namespace沒有給出參數時，這些方法 默認為正確的命名空間。

如果事件在基于類的命名空間中具有處理程序，并且還有基于裝飾器的函數處理程序，則僅調用修飾的函數處理程序。

測試

以上是作為官網文檔的翻譯，下面來說說寫完了代碼之后，應該怎么來調試。

<script type="text/javascript" src="//cdnjs.cloudflare.com/ajax/libs/socket.io/1.3.6/socket.io.min.js"></script> <script type="text/javascript" charset="utf-8">var socket = io.connect('http://' + document.domain + ':' + location.port);socket.on('connect', function() {socket.emit('my event', {data: 'I\'m connected!'});}); </script> 復制代碼

使用 JavaScript 來連接服務端，這里說一個我遇到的問題，最開始使用的是 jsbin 來測試，但怎么都連不到后端，原因就是 jsbin 是 HTTPS 的，而我的請求是 HTTP，于是還是老老實實寫了一個 HTML 文件，源碼可以直接在 Github 下載。

<!DOCTYPE HTML> <html> <head><title>Flask-SocketIO Test</title><script type="text/javascript" src="//code.jquery.com/jquery-1.4.2.min.js"></script><script type="text/javascript" src="//cdnjs.cloudflare.com/ajax/libs/socket.io/1.3.5/socket.io.min.js"></script><script type="text/javascript" charset="utf-8">$(document).ready(function() {namespace = '/test';var socket = io.connect(location.protocol + '//' + document.domain + ':' + location.port + namespace);socket.on('connect', function() {socket.emit('my_event', {data: 'I\'m connected!'});});socket.on('my_response', function(msg) {$('#log').append('<br>' + $('<div/>').text('Received #' + msg.count + ': ' + msg.data).html());});$('form#emit').submit(function(event) {socket.emit('my_event', {data: $('#emit_data').val()});return false;});});</script> </head> <body><h1>Flask-SocketIO Test</h1><p>Async mode is: <b>{{ async_mode }}</b></p><h2>Send:</h2><form id="emit" method="POST" action='#'><input type="text" name="emit_data" id="emit_data" placeholder="Message"><input type="submit" value="Echo"></form><h2>Receive:</h2><div id="log"></div> </body> </html> 復制代碼

有了這個頁面之后，就可以直接在瀏覽器中輸入 http://127.0.0.1:5000 訪問服務端了，更多功能可以隨意折騰。

相關文檔：

github.com/miguelgrinb…

flask-socketio.readthedocs.io/en/latest/

windrocblog.sinaapp.com/?p=1628

zhuanlan.zhihu.com/p/31118736

letus.club/2016/04/10/…

創作挑戰賽新人創作獎勵來咯，堅持創作打卡瓜分現金大獎

總結

以上是生活随笔為你收集整理的Flask-SocketIO 简单使用指南的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。