Flask-SocketIO如何使用

今天小編給大家分享一下Flask-SocketIO如何使用的相關(guān)知識(shí)點(diǎn)，內(nèi)容詳細(xì)，邏輯清晰，相信大部分人都還太了解這方面的知識(shí)，所以分享這篇文章給大家參考一下，希望大家閱讀完這篇文章后有所收獲，下面我們一起來了解一下吧。

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

安裝

直接使用 pip 來安裝：

pip install flask-socketio

要求

Flask-SocketIO 兼容 Python 2.7 和 Python 3.3+。可以從以下三個(gè)選項(xiàng)中選擇此程序包所依賴的異步服務(wù)：

• eventlet 性能最佳，支持長(zhǎng)輪詢和 WebSocket 傳輸。

• gevent 在許多不同的配置中得到支持。gevent 包完全支持長(zhǎng)輪詢傳輸，但與 eventlet 不同，gevent 沒有本機(jī) WebSocket 支持。要添加對(duì) WebSocket 的支持，目前有兩種選擇：安裝 gevent-websocket  包為 gevent 增加 WebSocket 支持，或者可以使用帶有 WebSocket 功能的 uWSGI  Web 服務(wù)器。gevent 的使用也是一種高性能選項(xiàng)，但略低于 eventlet。

• 也可以使用基于 Werkzeug 的 Flask 開發(fā)服務(wù)器，但需要注意的是，它缺乏其他兩個(gè)選項(xiàng)的性能，因此它只應(yīng)用于簡(jiǎn)單的開發(fā)環(huán)境。此選項(xiàng)僅支持長(zhǎng)輪詢傳輸。

擴(kuò)展會(huì)根據(jù)安裝的內(nèi)容自動(dòng)檢測(cè)要使用的異步框架。優(yōu)先考慮 eventlet，然后是 gevent。對(duì)于 gevent 中的WebSocket 支持，首選 uWSGI，然后是 gevent-websocket。如果既未安裝 eventlet 也未安裝 gevent，則使用 Flask 開發(fā)服務(wù)器。

如果使用多個(gè)進(jìn)程，則進(jìn)程使用消息隊(duì)列服務(wù)來協(xié)調(diào)諸如廣播之類的操作。支持的隊(duì)列是 Redis，RabbitMQ以及 Kombu 軟件包支持的任何其他消息隊(duì)列 。

在客戶端，官方 Socket.IO Javascript 客戶端庫(kù)可用于建立與服務(wù)器的連接。還有使用 Swift，Java 和 C ++ 編寫的官方客戶端。非官方客戶端也可以工作，只要它們實(shí)現(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)

以上代碼即完成了一個(gè)簡(jiǎn)單的 Web 服務(wù)器。

socketio.run()函數(shù)封裝了 Web 服務(wù)器的啟動(dòng)，并替換了app.run()標(biāo)準(zhǔn)的 Flask 開發(fā)服務(wù)器啟動(dòng)。

當(dāng)應(yīng)用程序處于調(diào)試模式時(shí)，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ò)展提供了適用于啟動(dòng) Socket.IO 服務(wù)器的新版本命令。用法示例：flask run。

$ FLASK_APP=my_app.py flask run

或者直接使用下面方式，也可以啟動(dòng)項(xiàng)目：

$ python2.7 app.py

連接事件

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

@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以拒絕連接。這樣就可以在此時(shí)對(duì)客戶端進(jìn)行身份驗(yàn)證。

請(qǐng)注意，連接和斷開連接事件將在使用的每個(gè)命名空間上單獨(dú)發(fā)送。

接收消息

使用 SocketIO 時(shí)，雙方都會(huì)將消息作為事件接收。在客戶端使用 Javascript 回調(diào)。使用 Flask-SocketIO，服務(wù)器需要為這些事件注冊(cè)處理程序，類似于視圖函數(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))

自定義命名事件也可以支持多個(gè)參數(shù)：

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

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

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

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

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

def my_function_handler(data):
pass

socketio.on_event('my event', my_function_handler, namespace='/test')

客戶端可以請(qǐng)求確認(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

在上面的示例中，將使用兩個(gè)參數(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ā)送具有多個(gè)參數(shù)的事件，請(qǐng)發(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)時(shí)，Javascript 客戶端會(huì)收到一個(gè)回調(diào)函數(shù)，以便在收到消息時(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 的另一個(gè)非常有用的功能是廣播消息。SocketIO 支持通過此功能broadcast=True可選參數(shù)send()和emit()：

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

在啟用廣播選項(xiàng)的情況下發(fā)送消息時(shí)，連接到命名空間的所有客戶端都會(huì)接收它，包括發(fā)件人。如果未使用名稱空間，則連接到全局名稱空間的客戶端將收到該消息。請(qǐng)注意，不會(huì)為廣播消息調(diào)用回調(diào)。

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

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

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

房間

對(duì)于許多應(yīng)用程序，有必要將用戶分組為可以一起尋址的子集。最好的例子是具有多個(gè)房間的聊天應(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ù)接受一個(gè)可選room導(dǎo)致被發(fā)送到所有的都在定房客戶端的消息的說法。

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

由于為所有客戶端分配了個(gè)人房間，為了向單個(gè)客戶端發(fā)送消息，客戶端的會(huì)話 ID 可以用作房間參數(shù)。

錯(cuò)誤處理

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

錯(cuò)誤處理函數(shù)將異常對(duì)象作為參數(shù)。

還可以使用request.event變量檢查當(dāng)前請(qǐng)求的消息和數(shù)據(jù)參數(shù)，這對(duì)于事件處理程序外部的錯(cuò)誤記錄和調(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'))

使用基于類的命名空間時(shí)，服務(wù)器接收的任何事件都將調(diào)度到名為帶有on_前綴的事件名稱的方法。例如，事件my_event將由名為的方法處理on_my_event。如果收到的事件沒有在命名空間類中定義的相應(yīng)方法，則忽略該事件?；陬惖拿臻g中使用的所有事件名稱必須使用方法名稱中合法的字符。

為了方便在基于類的命名空間中定義的方法，命名空間實(shí)例包括類中的幾個(gè)方法的版本，flask_socketio.SocketIO當(dāng)namespace沒有給出參數(shù)時(shí)，這些方法 默認(rèn)為正確的命名空間。

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

測(cè)試

以上是作為官網(wǎng)文檔的翻譯，下面來說說寫完了代碼之后，應(yīng)該怎么來調(diào)試。

<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 來連接服務(wù)端，這里說一個(gè)我遇到的問題，最開始使用的是 jsbin 來測(cè)試，但怎么都連不到后端，原因就是 jsbin 是 HTTPS 的，而我的請(qǐng)求是 HTTP，于是還是老老實(shí)實(shí)寫了一個(gè) HTML 文件，源碼可以直接在 Github 下載。

<!DOCTYPE HTML>
<html>
<head>
<title>Flask-SocketIO Testtitle>
<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('
' + $('').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>
<h2>Flask-SocketIO Testh2>
<p>Async mode is: <b>{{ async_mode }}b>p>
<h3>Send:h3>
<form id="emit" method="POST" action='#'>
<input type="text" name="emit_data" id="emit_data" placeholder="Message">
<input type="submit" value="Echo">
form>
<h3>Receive:h3>
<div id="log">div>
body>
html>

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

以上就是“Flask-SocketIO如何使用”這篇文章的所有內(nèi)容，感謝各位的閱讀！相信大家閱讀完這篇文章都有很大的收獲，小編每天都會(huì)為大家更新不同的知識(shí)，如果還想學(xué)習(xí)更多的知識(shí)，請(qǐng)關(guān)注億速云行業(yè)資訊頻道。