pip install flask-socketio
from flask import Flask, render_template
from flask_socketio import SocketIO, emit
app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret_key'
socketio = SocketIO()
socketio.init_app(app, cors_allowed_origins='*') # 解決跨域問題
name_space = '/dcenter'
@app.route('/')
def index():
return render_template('index.html')
@app.route('/push')
def push_once():
event_name = 'dcenter'
broadcasted_data = {
'data': "test message!"}
# 服務端發送給前端事件消息
socketio.emit(event_name, broadcasted_data, broadcast=False, namespace=name_space)
return 'done!'
@socketio.on('connect', namespace=name_space)
def connected_msg():
print('client connected.')
@socketio.on('disconnect', namespace=name_space)
def disconnect_msg():
print('client disconnected.')
# 後端接收事件消息
@socketio.on('my_event', namespace=name_space)
def mtest_message(message):
# 不同的事件
print(message)
emit('my_response',
{
'data': message['data'], 'count': 1})
if __name__ == '__main__':
socketio.run(app, host='0.0.0.0', port=5000, debug=True)
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>SocketIO Demo</title>
<script type="text/javascript" src="//cdn.bootcss.com/jquery/3.1.1/jquery.min.js"></script>
<script type="text/javascript" src="//cdn.socket.io/4.4.1/socket.io.js"></script>
</head>
<body>
<h2>Demo of SocketIO</h2>
<div id="t"></div>
<script> $(document).ready(function () {
namespace = '/dcenter'; var socket = io.connect(location.protocol + '//' + document.domain + ':' + location.port + namespace); // 前端發送消息 socket.emit('my_event',{
'data':'hello world'}); // 前端接收消息 socket.on('dcenter', function (res) {
var t = res.data; if (t) {
$("#t").append(t).append('<br/>'); } }); socket.on('my_response', function (res) {
var t = res.data; if (t) {
$("#t").append(t).append('<br/><br/>'); } }) }); </script>
</body>
</html>
這裡需要注意的是，當前我使用的flask-socketio版本是5.2，需要對應前端socket.io版本是4.x，否則會報錯。
