Flask-SocketIO服务端安装及使用代码示例

2023-07-10 17:21:03 71

安装Flask-Sockets很容易：

pipinstallflask-socketio

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

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

fromflaskimportFlask,request,jsonify,render_template
fromflask_socketioimportSocketIO,emit


app=Flask(__name__)
app.config['SECRET_KEY']='secret!'
socketio=SocketIO(app)


@app.route('/')
defindex():
returnrender_template('test.html')

#触发事件myevent：回复只发送此连接
@socketio.on('my_event',namespace='/chat')
deftest_message(message):
print(message)
emit('my_response',{'data':message['data']})


#触发事件mybroadcastevent：：回复所有链接（广播）
@socketio.on('my_broadcastevent',namespace='/chat')
deftest_message(message):
print(message)
emit('my_response',{'data':message['data']},broadcast=True)

##################################################################
#自动连接和自动断开触发
@socketio.on('connect',namespace='/chat')
deftest_connect():
print("连接到来")
emit('my_response',{'data':'Connected'})

@socketio.on('disconnect',namespace='/chat')
deftest_disconnect():
print("连接断开")
print('Clientdisconnected')

if__name__=='__main__':
socketio.run(app,host='0.0.0.0',port=8200,debug=True)

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

test.html

官方完整示例

#!/usr/bin/envpython
fromthreadingimportLock
fromflaskimportFlask,render_template,session,request,\
copy_current_request_context
fromflask_socketioimportSocketIO,emit,join_room,leave_room,\
close_room,rooms,disconnect

#Setthisvariableto"threading","eventlet"or"gevent"totestthe
#differentasyncmodes,orleaveitsettoNonefortheapplicationtochoose
#thebestoptionbasedoninstalledpackages.
async_mode=None

app=Flask(__name__)
app.config['SECRET_KEY']='secret!'
socketio=SocketIO(app,async_mode=async_mode)
thread=None
thread_lock=Lock()

defbackground_thread():
"""Exampleofhowtosendservergeneratedeventstoclients."""
count=0
whileTrue:
socketio.sleep(100)
count+=1
socketio.emit('my_response',
{'data':'Servergeneratedevent','count':count},
namespace='/test')

@app.route('/')
defindex():
returnrender_template('test1.html',async_mode=socketio.async_mode)

@socketio.on('my_event',namespace='/test')
defmtest_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')
defmtest_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')
defjoin(message):
print(message)
join_room(message['room'])
session['receive_count']=session.get('receive_count',0)+1
emit('my_response',
{'data':'Inrooms:'+','.join(rooms()),
'count':session['receive_count']})

@socketio.on('leave',namespace='/test')
defleave(message):
print(message)
leave_room(message['room'])
session['receive_count']=session.get('receive_count',0)+1
emit('my_response',
{'data':'Inrooms:'+','.join(rooms()),
'count':session['receive_count']})

@socketio.on('close_room',namespace='/test')
defclose(message):
session['receive_count']=session.get('receive_count',0)+1
emit('my_response',{'data':'Room'+message['room']+'isclosing.',
'count':session['receive_count']},
room=message['room'])
close_room(message['room'])

@socketio.on('my_room_event',namespace='/test')
defsend_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')
deftest_connect():
print("连接到来")
emit('myresponse',{'data':'Connected'})

@socketio.on('disconnect',namespace='/test')
deftest_disconnect():
print("连接断开")
print('Clientdisconnected')


if__name__=='__main__':
socketio.run(app,debug=True,port=8200)

test1.html




Flask-SocketIOTest



$(document).ready(function(){
varsocket=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()});
returnfalse;
});
$('form#broadcast').submit(function(event){
socket.emit('my_broadcast_event',{data:$('#broadcast_data').val()});
returnfalse;
});
$('form#join').submit(function(event){
socket.emit('join',{room:$('#join_room').val()});
returnfalse;
});
$('form#leave').submit(function(event){
socket.emit('leave',{room:$('#leave_room').val()});
returnfalse;
});
$('form#send_room').submit(function(event){
socket.emit('my_room_event',{room:$('#room_name').val(),data:$('#room_data').val()});
returnfalse;
});
$('form#close').submit(function(event){
socket.emit('close_room',{room:$('#close_room').val()});
returnfalse;
});
$('form#disconnect').submit(function(event){
socket.emit('disconnect_request');
returnfalse;
});
});



Flask-SocketIOTest
Asyncmodeis:{{async_mode}}
Averageping/ponglatency:ms
Send:




























Receive:

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

Flask-SocketIO服务端安装及使用代码示例

Flask-SocketIOTest

Send:

Receive:

热门推荐

随机推荐