WebScoket详解：基于Flask、Flask-SocketIO、Flask-Apscheduler、HTML、JavaScript、浏览器
千层冷面 2024-09-04 08:33:01 阅读 81
WebSocket 是一种在单个 TCP 连接上进行全双工通讯的协议。它使得客户端和服务器之间的数据交换变得更加简单，允许服务端主动向客户端推送数据。本文是一个简单的示例，重在体验其原理和作用，细节还需要进一步学习。
一、代码
文件结构
说明
socket：任意创建一个包index.html: flask的模板文件，websocket的客户端代码，基于javascript<code>__init__.py: 包文件，这个例子中为空scoket_io_demo.py：websocket的服务端代码，基于flask
index.html
<!DOCTYPE html>
<html lang="en">code>
<head>
<meta charset="UTF-8">code>
<title>Socket客户端</title>
<script src="//cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.0/socket.io.js"></script>code>
<script>
// var socket = io.connect('http://' + document.domain + ':' + location.port + '/test');
var socket = io.connect('http://localhost:5000/test'); // 连接本地flask服务器
// 监听服务器的connect事件，连接成功浏览器控制台打印`连接到服务器`
socket.on('connect', function() { -- -->
console.log('连接到服务器');
document.getElementById('messages').innerHTML += '<p style="color:blue;">连接到服务器' + '</p>';code>
});
// 监听服务器的disconnect事件，断开连接浏览器控制台打印`断开连接`
socket.on('disconnect', function() { -- -->
console.log('断开连接');
document.getElementById('messages').innerHTML += '<p style="color:blue;">断开连接' + '</p>';code>
});
socket.on('server_send', function(msg) { -- -->
console.log('接收到服务器发送的消息: ' + msg.data);
// 你可以在这里更新DOM元素来显示消息
var color_arr = ['red', 'green', 'blue'];
var color_ = color_arr[Math.floor(Math.random() * 3)];
document.getElementById('messages').innerHTML += `<p style="color:${ color_}">接收到服务器发送的消息: ${ -- -->msg.data}</p>`;code>
});
// 发送消息到服务器的示例
function sendMessage() { -- -->
var message = document.getElementById('messageInput').value;
socket.emit('client_send', { 'data': message});
document.getElementById('messageInput').value = ''; // 清空输入框
}
</script>
</head>
<body>
<h1>Flask-SocketIO 客户端</h1>
<input type="text" id="messageInput" placeholder="输入要发送的信息" />code>
<button onclick="sendMessage()">发送消息</button>code>
<div id="messages"></div>code>
</body>
</html>
socket_io_demo.py
from datetime import datetime
from flask import Flask, render_template
from flask_socketio import SocketIO, emit
from flask_apscheduler import APScheduler
app = Flask(__name__)
app.config['SECRET_KEY'] = 'your_secret_key'
app.config['DEBUG'] = True
socketio = SocketIO(app)
scheduler = APScheduler(app=app)
@app.route('/')
def index():
"""渲染 HTML 客户端页面"""
return render_template('index.html') # 会去找当前目录中templates目录下的index.html文件
@socketio.on('connect', namespace='/test')code>
def test_connect():
"""处理客户端连接事件"""
print('客户端已连接')
# 发送sever_send事件，携带一条欢迎消息给客户端，namespace为/test
socketio.emit('server_send', { -- -->'data': '欢迎使用服务!'}, namespace='/test')code>
@socketio.on('disconnect', namespace='/test')code>
def test_disconnect():
"""处理客户端断开连接事件"""
print('客户端断开连接')
# 监听客户端client_send事件，接收客户端发送的消息，并广播给所有连接的客户端
@socketio.on('client_send', namespace='/test') # 监听客户端发送的消息，namespace为/testcode>
def handle_message(message):
"""处理客户端发送的消息，并将其广播给所有连接的客户端"""
print('接收到客户端的消息: ' + message['data'])
# 将消息直接广播
emit('server_send', { -- -->'data': message['data']}, broadcast=True, namespace='/test')code>
# 模拟实时数据推送（每10秒发送一次消息）
@scheduler.task('interval', id='real_time_update', seconds=10)code>
def setup_real_time_loop():
current_time = f'当前时间 { -- -->datetime.now().strftime("%Y-%m-%d %H:%M:%S")}'
print(f'发送给客户端的消息：{ current_time}')
socketio.emit('server_send', { 'data': f'{ current_time}'}, namespace='/test')code>
if __name__ == '__main__':
scheduler.start()
socketio.run(app, debug=True, host='localhost', port=5000, use_reloader=False, allow_unsafe_werkzeug=True)code>
二、配置
创建目录结构
将index.html代码拷入对应文件，若缺少以下js依赖，pychram会自动提示安装，安装即可src="//cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.0/socket.io.js"code>
将socker_io_demo.py拷入对应文件，若未安装flask相关库，则执行以下命令
# -i 加入清华源加速安装，以下库名对大小写不敏感，但是在import时大小写敏感
pip install Flask Flask-socketio Flask-apscheduler -i https://pypi.tuna.tsinghua.edu.cn/simple
三、运行
运行socket_io_demo.py
点击蓝色链接，在浏览器中自动打开index.html文件
回到pycharm界面，多次点击蓝色链接，打开多个客户端，我这里打开2个
在任意一个界面，输入信息如<code>哈哈哈，点击发送，则会在另一个界面收到哈哈哈，pycharm运行窗口也会打印 接收到到的信息
四、原理
上图
代码细节
客户端：<code>socket.on 和 socket.emit
收信：监听服务器的某个发送信息的事件，socket.on('disconnect', function() {});发信：触发某个带有信息的事件，socket.emit('client_send', {'data': message});
服务器：SocketIO.on 和 SocketIO.emit(无广播功能)或flask_socketio.emit(有广播功能)
收信：监听服务器的某个发送信息的事件
@socketio.on('client_send', namespace='/test') # 监听客户端发送的消息code>
发信：触发某个带有信息的事件，示例中用到了带广播功能的flask_socketio.emit
emit('server_send', { -- -->'data': message['data']}, broadcast=True, namespace='/test')code>
五、其他疑问解答
SocketIO的使用
# 导入
from flask import Flask
from flask_socketio import SocketIO, emit
# 创建flask App
app = Flask(__name__)
# 初始化SocketIO对象，将其绑定到App
socketio = SocketIO(app)
# 使用
@socketio.on('xxx') # 监听客户端的事件
def process_message():
# 直接做广播处理
emit('yyy', { -- -->'data': message['data']}, broadcast=True) # 触发事件，此事件应该在客户端监听
# 启动，使用了socketio来扩展flask，则启动方式由app.run()变为socketio.run()
socketio.run(app)
APScheduler定时任务使用
# 导入
from flask import Flask
from flask_apscheduler import APScheduler
# 创建flask App
app = Flask(__name__)
# 初始化APScheduler对象，将其绑定到App
scheduler = APScheduler(app=app)
# 使用
@scheduler.task('interval', id='send_message', seconds=10) # interval间隔模式启动，每隔10s，还有cron(类似linux定时任务)、date(指定时间执行一次)方式code>
def send_message():
current_time = f'当前时间 { -- -->datetime.now().strftime("%Y-%m-%d %H:%M:%S")}'
print(f'发送给客户端的消息：{ current_time}')
socketio.emit('server_send', { 'data': f'{ current_time}'}) # server_send事件应该在客户端代码中监听并处理
# 启动, 先启动任务，再启动App
scheduler.start()
app.run()
误区解答：无论是服务器主动推送数据的websocket，还是客户端主动请求无状态的HTTP，通常情况下都是客户端主动向服务器发起连接请求，而不是服务器主动找客户端，这是因为客户端知道它需要与服务器进行通信，并且它知道服务器的地址（URL或IP地址）。因此在运行代码时应该先启动服务器(flask)的代码，然后启动客户端HTML界面。
HTTP和WebSocket异同详解传送门: https://blog.csdn.net/qq_52011411/article/details/141569619
声明
本文内容仅代表作者观点，或转载于其他网站，本站不以此文作为商业用途
如有涉及侵权，请联系本站进行删除
转载本站原创文章，请注明来源及作者。