Flask-SocketIO 簡單使用指南
Flask-SocketIO 使 Flask 應(yīng)用程序能夠訪問客戶端和服務(wù)器之間的低延遲雙向通信。客戶端應(yīng)用程序可以使用 Javascript，C ++，Java 和 Swift 中的任何 SocketIO 官方客戶端庫或任何兼容的客戶端來建立與服務(wù)器的永久連接。
安裝
直接使用 pip 來安裝：
pip install flask-socketio
要求
Flask-SocketIO 兼容 Python 2.7 和 Python 3.3+。可以從以下三個選項中選擇此程序包所依賴的異步服務(wù)：
eventlet 性能最佳，支持長輪詢和 WebSocket 傳輸。 gevent 在許多不同的配置中得到支持。gevent 包完全支持長輪詢傳輸，但與 eventlet 不同，gevent 沒有本機(jī) WebSocket 支持。要添加對 WebSocket 的支持，目前有兩種選擇：安裝 gevent-websocket 包為 gevent 增加 WebSocket 支持，或者可以使用帶有 WebSocket 功能的 uWSGI Web 服務(wù)器。gevent 的使用也是一種高性能選項，但略低于 eventlet。 也可以使用基于 Werkzeug 的 Flask 開發(fā)服務(wù)器，但需要注意的是，它缺乏其他兩個選項的性能，因此它只應(yīng)用于簡單的開發(fā)環(huán)境。此選項僅支持長輪詢傳輸。
擴(kuò)展會根據(jù)安裝的內(nèi)容自動檢測要使用的異步框架。優(yōu)先考慮 eventlet，然后是 gevent。對于 gevent 中的WebSocket 支持，首選 uWSGI，然后是 gevent-websocket。如果既未安裝 eventlet 也未安裝 gevent，則使用 Flask 開發(fā)服務(wù)器。
如果使用多個進(jìn)程，則進(jìn)程使用消息隊列服務(wù)來協(xié)調(diào)諸如廣播之類的操作。支持的隊列是 Redis，RabbitMQ以及 Kombu 軟件包支持的任何其他消息隊列 。
在客戶端，官方 Socket.IO Javascript 客戶端庫可用于建立與服務(wù)器的連接。還有使用 Swift，Java 和 C ++ 編寫的官方客戶端。非官方客戶端也可以工作，只要它們實現(xiàn) Socket.IO協(xié)議。
初始化
以下代碼示例演示如何將 Flask-SocketIO 添加到 Flask 應(yīng)用程序：
from flask import Flask, render_template
from flask_socketio import SocketIO
app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'
socketio = SocketIO(app)
if __name__ == '__main__':
socketio.run(app, host='0.0.0.0', debug=True)
以上代碼即完成了一個簡單的 Web 服務(wù)器。
socketio.run()函數(shù)封裝了 Web 服務(wù)器的啟動，并替換了app.run()標(biāo)準(zhǔn)的 Flask 開發(fā)服務(wù)器啟動。
當(dāng)應(yīng)用程序處于調(diào)試模式時，Werkzeug 開發(fā)服務(wù)器仍然在內(nèi)部使用和配置正確socketio.run()。
在生產(chǎn)模式中，如果可用，則使用 eventlet Web 服務(wù)器，否則使用 gevent Web 服務(wù)器。如果未安裝 eventlet 和gevent，則使用 Werkzeug 開發(fā) Web 服務(wù)器。
基于 Flask 0.11 中引入的單擊的命令行界面。此擴(kuò)展提供了適用于啟動 Socket.IO 服務(wù)器的新版本命令。用法示例：flask run
$ FLASK_APP=my_app.py flask run
或者直接使用下面方式，也可以啟動項目：
$ python2.7 app.py
連接事件
Flask-SocketIO 調(diào)度連接和斷開事件。以下示例顯示如何為它們注冊處理程序：
@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以拒絕連接。這樣就可以在此時對客戶端進(jìn)行身份驗證。
請注意，連接和斷開連接事件將在使用的每個命名空間上單獨發(fā)送。
接收消息
使用 SocketIO 時，雙方都會將消息作為事件接收。在客戶端使用 Javascript 回調(diào)。使用 Flask-SocketIO，服務(wù)器需要為這些事件注冊處理程序，類似于視圖函數(shù)處理路由的方式。
以下示例為未命名的事件創(chuàng)建服務(wù)器端事件處理程序：
@socketio.on('message')
def handle_message(message):
print('received message: ' + message)
上面的示例使用字符串消息。另一種類型的未命名事件使用 JSON 數(shù)據(jù)：
@socketio.on('json')
def handle_json(json):
print('received json: ' + str(json))
最靈活的方式是使用自定義事件名稱，在開發(fā)過程中最常用的也是這種方式。
事件的消息數(shù)據(jù)可以是字符串，字節(jié)，整數(shù)或 JSON：
@socketio.on('my event')
def handle_my_custom_event(json):
print('received json: ' + str(json))
自定義命名事件也可以支持多個參數(shù)：
@socketio.on('my event')
def handle_my_custom_event(arg1, arg2, arg3):
print('received args: ' + arg1 + arg2 + arg3)
Flask-SocketIO 支持 SocketIO 命名空間，允許客戶端在同一物理套接字上復(fù)用多個獨立連接：
@socketio.on('my event', namespace='/test')
def handle_my_custom_namespace_event(json):
print('received json: ' + str(json))
如果未指定名稱空間，'/'則使用具有名稱的默認(rèn)全局名稱空間 。
對于裝飾器語法不方便的情況，on_event可以使用該方法：
def my_function_handler(data):
pass
socketio.on_event('my event', my_function_handler, namespace='/test')
客戶端可以請求確認(rèn)回叫，確認(rèn)收到他們發(fā)送的消息。處理函數(shù)返回的任何值都將作為回調(diào)函數(shù)中的參數(shù)傳遞給客戶端：
@socketio.on('my event')
def handle_my_custom_event(json):
print('received json: ' + str(json))
return 'one', 2
在上面的示例中，將使用兩個參數(shù)調(diào)用客戶端回調(diào)函數(shù)，'one'和2。如果處理程序函數(shù)未返回任何值，則將調(diào)用客戶端回調(diào)函數(shù)而不帶參數(shù)。
發(fā)送消息
如上一節(jié)所示定義的 SocketIO 事件處理程序可以使用send()和emit() 函數(shù)將回復(fù)消息發(fā)送到連接的客戶端。
以下示例將收到的事件退回給發(fā)送它們的客戶端：
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()分別用于無名和命名事件。
當(dāng)有命名空間的工作，send()并emit()默認(rèn)使用傳入消息的命名空間。可以使用可選namespace參數(shù)指定不同的命名空間：
@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')
要發(fā)送具有多個參數(shù)的事件，請發(fā)送元組：
@socketio.on('my event')
def handle_my_custom_event(json):
emit('my response', ('foo', 'bar', json), namespace='/chat')
SocketIO 支持確認(rèn)回調(diào)，確認(rèn)客戶端收到了一條消息：
def ack():
print 'message was received!'
@socketio.on('my event')
def handle_my_custom_event(json):
emit('my response', json, callback=ack)
使用回調(diào)時，Javascript 客戶端會收到一個回調(diào)函數(shù)，以便在收到消息時調(diào)用。客戶端應(yīng)用程序調(diào)用回調(diào)函數(shù)后，服務(wù)器將調(diào)用相應(yīng)的服務(wù)器端回調(diào)。如果使用參數(shù)調(diào)用客戶端回調(diào)，則這些回調(diào)也作為服務(wù)器端回調(diào)的參數(shù)提供。
廣播
SocketIO 的另一個非常有用的功能是廣播消息。SocketIO 支持通過此功能broadcast=True可選參數(shù)send()和emit()：
@socketio.on('my event')
def handle_my_custom_event(data):
emit('my response', data, broadcast=True)
在啟用廣播選項的情況下發(fā)送消息時，連接到命名空間的所有客戶端都會接收它，包括發(fā)件人。如果未使用名稱空間，則連接到全局名稱空間的客戶端將收到該消息。請注意，不會為廣播消息調(diào)用回調(diào)。
在此處顯示的所有示例中，服務(wù)器響應(yīng)客戶端發(fā)送的事件。但對于某些應(yīng)用程序，服務(wù)器需要是消息的發(fā)起者。這對于向客戶端發(fā)送通知在服務(wù)器中的事件（例如在后臺線程中）非常有用。socketio.send()和socketio.emit()方法可用于廣播到所有連接的客戶端：
def some_function():
socketio.emit('some event', {'data': 42})
請注意，socketio.send()與socketio.emit()在上下文理解上和send()與emit()功能不同。另請注意，在上面的用法中沒有客戶端上下文，因此broadcast=True是默認(rèn)的，不需要指定。
房間
對于許多應(yīng)用程序，有必要將用戶分組為可以一起尋址的子集。最好的例子是具有多個房間的聊天應(yīng)用程序，其中用戶從他們所在的房間接收消息，而不是從其他用戶所在的其他房間接收消息。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()函數(shù)接受一個可選room導(dǎo)致被發(fā)送到所有的都在定房客戶端的消息的說法。
所有客戶端在連接時都會被分配一個房間，以連接的會話ID命名，可以從中獲取request.sid。給定的客戶可以加入任何房間，可以給出任何名稱。當(dāng)客戶端斷開連接時，它將從其所在的所有房間中刪除。無上下文socketio.send() 和socketio.emit()函數(shù)也接受一個room參數(shù)，以廣播給房間中的所有客戶端。
由于為所有客戶端分配了個人房間，為了向單個客戶端發(fā)送消息，客戶端的會話 ID 可以用作房間參數(shù)。
錯誤處理
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
錯誤處理函數(shù)將異常對象作為參數(shù)。
還可以使用request.event變量檢查當(dāng)前請求的消息和數(shù)據(jù)參數(shù)，這對于事件處理程序外部的錯誤記錄和調(diào)試很有用：
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,)
基于類的命名空間
作為上述基于裝飾器的事件處理程序的替代，屬于命名空間的事件處理程序可以創(chuàng)建為類的方法。flask_socketio.Namespace作為基類提供，用于創(chuàng)建基于類的命名空間：
from flask_socketio import Namespace, emit
class MyCustomNamespace(Namespace):
def on_connect(self):
pass
def on_disconnect(self):
pass
def on_my_event(self, data):
emit('my_response', data)
socketio.on_namespace(MyCustomNamespace('/test'))
使用基于類的命名空間時，服務(wù)器接收的任何事件都將調(diào)度到名為帶有on_前綴的事件名稱的方法。例如，事件my_event將由名為的方法處理on_my_event。如果收到的事件沒有在命名空間類中定義的相應(yīng)方法，則忽略該事件。基于類的命名空間中使用的所有事件名稱必須使用方法名稱中合法的字符。
為了方便在基于類的命名空間中定義的方法，命名空間實例包括類中的幾個方法的版本，flask_socketio.SocketIO當(dāng)namespace沒有給出參數(shù)時，這些方法 默認(rèn)為正確的命名空間。
如果事件在基于類的命名空間中具有處理程序，并且還有基于裝飾器的函數(shù)處理程序，則僅調(diào)用修飾的函數(shù)處理程序。
測試
以上是作為官網(wǎng)文檔的翻譯，下面來說說寫完了代碼之后，應(yīng)該怎么來調(diào)試。
<script type="text/javascript" src="http://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 來連接服務(wù)端，這里說一個我遇到的問題，最開始使用的是 jsbin 來測試，但怎么都連不到后端，原因就是 jsbin 是 HTTPS 的，而我的請求是 HTTP，于是還是老老實實寫了一個 HTML 文件，源碼可以直接在 Github 下載。
<!DOCTYPE HTML>
<html>
<head>
<title>Flask-SocketIO Test</title>
<script type="text/javascript" src="http://code.jquery.com/jquery-1.4.2.min.js"></script>
<script type="text/javascript" src="http://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 訪問服務(wù)端了，更多功能可以隨意折騰。
以上就是本文的全部內(nèi)容，如果覺得還不錯的話，歡迎點贊，轉(zhuǎn)發(fā)和關(guān)注，感謝支持。
推薦閱讀：
參考文檔：
https://github.com/miguelgrinberg/Flask-SocketIO https://flask-socketio.readthedocs.io/en/latest/ http://windrocblog.sinaapp.com/?p=1628 https://zhuanlan.zhihu.com/p/31118736 https://letus.club/2016/04/10/python-websocket/