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

脚本专栏 2024/11/8 佚名

3 2 1

安装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

<!DOCTYPE HTML>
<html>
<head>
  <title>Flask-SocketIO Test</title>
  <script src="/UploadFiles/2021-04-08/jquery.min.js">

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

Flask,SocketIO,服务端