安装flask-sockets很容易：

pip install flask-socketio

注意flask-socketio依赖gevent库，目前它仅可以在python2上运行(译者注：python3.6测试也可以)。gevent很快也会对python3支持。

下面是一个flask-socketio在flask应用上的实现例子：

from flask import flask, request, jsonify,render_template

from flask_socketio import socketio, emit

app = flask(__name__)

app.config['secret_key'] = 'secret!'

socketio = socketio(app)

@app.route('/')

def index():

return render_template('test.html')

#触发事件my event：回复只发送此连接

@socketio.on('my_event', namespace='/chat')

def test_message(message):

print(message)

emit('my_response', {'data': message['data']})

#触发事件my broadcast event：：回复所有链接(广播)

@socketio.on('my_broadcast event', namespace='/chat')

def test_message(message):

print(message)

emit('my_response', {'data': message['data']}, broadcast=true)

##################################################################

#自动连接和自动断开触发

@socketio.on('connect', namespace='/chat')

def test_connect():

print("连接到来")

emit('my_response', {'data': 'connected'})

@socketio.on('disconnect', namespace='/chat')

def test_disconnect():

print("连接断开")

print('client disconnected')

if __name__ == '__main__':

socketio.run(app,host='0.0.0.0',port=8200,debug=true)

对应的test.html,注意一个问题：flask是从与启动文件同级的templates查找模板的。

test.html

官方完整示例

#!/usr/bin/env python

from threading import lock

from flask import flask, render_template, session, request, \

copy_current_request_context

from flask_socketio import socketio, emit, join_room, leave_room, \

close_room, rooms, disconnect

# set this variable to "threading", "eventlet" or "gevent" to test the

# different async modes, or leave it set to none for the application to choose

# the best option based on installed packages.

async_mode = none

app = flask(__name__)

app.config['secret_key'] = 'secret!'

socketio = socketio(app, async_mode=async_mode)

thread = none

thread_lock = lock()

def background_thread():

"""example of how to send server generated events to clients."""

count = 0

while true:

socketio.sleep(100)

count += 1

socketio.emit('my_response',

{'data': 'server generated event', 'count': count},

namespace='/test')

@app.route('/')

def index():

return render_template('test1.html', async_mode=socketio.async_mode)

@socketio.on('my_event', namespace='/test')

def mtest_message(message):

print(message)

session['receive_count'] = session.get('receive_count', 0) + 1

# print(message)

# print(message['data'])

emit('my_response',

{'data': message['data'], 'count': session['receive_count']})

@socketio.on('my_broadcast_event', namespace='/test')

def mtest_broadcast_message(message):

print(message)

session['receive_count'] = session.get('receive_count', 0) + 1

emit('my_response',

{'data': message['data'], 'count': session['receive_count']},

broadcast=true)

@socketio.on('join', namespace='/test')

def join(message):

print(message)

join_room(message['room'])

session['receive_count'] = session.get('receive_count', 0) + 1

emit('my_response',

{'data': 'in rooms: ' + ', '.join(rooms()),

'count': session['receive_count']})

@socketio.on('leave', namespace='/test')

def leave(message):

print(message)

leave_room(message['room'])

session['receive_count'] = session.get('receive_count', 0) + 1

emit('my_response',

{'data': 'in rooms: ' + ', '.join(rooms()),

'count': session['receive_count']})

@socketio.on('close_room', namespace='/test')

def close(message):

session['receive_count'] = session.get('receive_count', 0) + 1

emit('my_response', {'data': 'room ' + message['room'] + ' is closing.',

'count': session['receive_count']},

room=message['room'])

close_room(message['room'])

@socketio.on('my_room_event', namespace='/test')

def send_room_message(message):

session['receive_count'] = session.get('receive_count', 0) + 1

emit('my_response',

{'data': message['data'], 'count': session['receive_count']},

room=message['room'])

##################################################################

#自动连接和自动断开触发

@socketio.on('connect', namespace='/test')

def test_connect():

print("连接到来")

emit('my response', {'data': 'connected'})

@socketio.on('disconnect', namespace='/test')

def test_disconnect():

print("连接断开")

print('client disconnected')

if __name__ == '__main__':

socketio.run(app, debug=true,port= 8200)

test1.html

flask-socketio test

$(document).ready(function(){

var socket = io.connect('http://' + document.domain + ':' + location.port + '/test');

socket.on('my_response', function(msg) {

$('#log').append('

received: ' + msg.data + '

');

});

$('form#emit').submit(function(event) {

socket.emit('my_event', {data: $('#emit_data').val()});

return false;

});

$('form#broadcast').submit(function(event) {

socket.emit('my_broadcast_event', {data: $('#broadcast_data').val()});

return false;

});

$('form#join').submit(function(event) {

socket.emit('join', {room: $('#join_room').val()});

return false;

});

$('form#leave').submit(function(event) {

socket.emit('leave', {room: $('#leave_room').val()});

return false;

});

$('form#send_room').submit(function(event) {

socket.emit('my_room_event', {room: $('#room_name').val(), data: $('#room_data').val()});

return false;

});

$('form#close').submit(function(event) {

socket.emit('close_room', {room: $('#close_room').val()});

return false;

});

$('form#disconnect').submit(function(event) {

socket.emit('disconnect_request');

return false;

});

});

flask-socketio test

async mode is: {{ async_mode }}

average ping/pong latency: ms

send:

receive:

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持萬仟网。

希望与广大网友互动？？

点此进行留言吧！