Flask和Vue框架实现WebSocket消息通信
主要的安装包 Flask、Flask-SocketIO，注意Python版本要求3.6+更新基础环境我的“requirements.txt”的包安装命令。
·
1 安装环境
1.1 安装Flask环境
主要的安装包 Flask、Flask-SocketIO，注意Python版本要求3.6+
# Flask-SocketIO参考地址
https://flask-socketio.readthedocs.io/en/latest/
https://github.com/miguelgrinberg/flask-socketio
# python-socketio地址
https://python-socketio.readthedocs.io/en/stable/
https://github.com/miguelgrinberg/python-socketio
更新基础环境
# 更新pip
python -m pip install --upgrade pip
# 更新setuptools
pip install --upgrade setuptools
# 安装Flask
pip install flask
pip install flask_cors
# 安装关于SocketIO的包
# 安装python-socketio时，会自动安装python-engineio依赖
pip install python-socketio
pip install flask-socketio
# eventlet具有WSGI支持的异步框架，主要功能是通过协程实现并发
pip install eventlet
我的“requirements.txt”的包
bidict==0.22.1
blinker==1.7.0
click==8.1.7
colorama==0.4.6
dnspython==2.4.2
eventlet==0.33.3
Flask==3.0.0
Flask-Cors==4.0.0
Flask-SocketIO==5.3.6
greenlet==3.0.1
h11==0.14.0
importlib-metadata==7.0.0
itsdangerous==2.1.2
Jinja2==3.1.2
MarkupSafe==2.1.3
python-engineio==4.8.0
python-socketio==5.10.0
simple-websocket==1.0.0
six==1.16.0
Werkzeug==3.0.1
wsproto==1.2.0
zipp==3.17.0
安装命令
pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple
1.2 安装Vue环境
（1）使用vite创建项目
# 创建项目
npm create vite@latest
# 选择框架和语言，我选择的是Vue和TypeScript
√ Project name: ... websocket
√ Select a framework: » Vue
√ Select a variant: » TypeScript
Done. Now run:
cd websocket
npm install
npm run dev
# 进入目录
cd websocket
# 安装相关包
npm install
# 安装包结束后提示如下：
added 44 packages, and audited 45 packages in 14s
（2）安装依赖包
# 安装一个包即可
npm install socket.io-client -S
socket.io-client的参考地址
https://socket.io/docs/v4/client-api/
查看npm包版本和例子的网络地址
https://www.npmjs.com/
备注：npm中命令参数的意思
--global（-g）：表示全局全局安装，包安装在了node目录下的node_modules/npm中（可以在任意项目中使用该工具）；不使用-g安装，包安装在了工程目录下的node_modules中下。
--save（-S）：表示写入package.json文件中的dependencies，这里面的包是发布到生产环境中的，例如：vue、axios等。
--save-dev（-D）：表示写入package.json文件中的devDependencies，这里面的包是仅在开发环境中辅助开发，在生产环境中不需要，例如：vite、css-loader等。
不使用参数时：npm使用的是--save（-S）命令
package.json文件
{
"name": "websocket",
"private": true,
"version": "0.0.0",
"type": "module",
"scripts": {
"dev": "vite",
"build": "vue-tsc && vite build",
"preview": "vite preview"
},
"dependencies": {
"socket.io-client": "^4.7.2",
"vue": "^3.3.8"
},
"devDependencies": {
"@types/node": "^20.10.2",
"@vitejs/plugin-vue": "^4.5.0",
"typescript": "^5.2.2",
"vite": "^5.0.0",
"vue-tsc": "^1.8.22"
}
}
（3）使用Vscode常见报错
报错1
Cannot find module './App.vue' or its corresponding type declarations.ts(2307)
解决方法
修改项目根目录中 “env.d.ts” 文件，添加如下内容：
/// <reference types="vite/client" />
declare module "*.vue" {
import { DefineComponent } from "vue"
const component: DefineComponent<{}, {}, any>
export default component
}
报错2
Module xx.vue has no default export.Vetur(1192)
解决方法
卸载vetur插件，主要针对vue2项目；
安装Volar插件，全称Vue Language Features (Volar)，主要针对vue3和TypeScript项目；
报错3
使用“@”无法导入ts文件，错误符号指向“@”，Cannot find module 'XXXXs' or its corresponding type declarations
解决方法
- 安装@types/node，会在项目中生成一个vite-env.d.ts文件
npm i @types/node -D
- 在vite.config.js文件中配置参数
import { defineConfig } from 'vite'
import vue from '@vitejs/plugin-vue'
import path from "path";
// https://vitejs.dev/config/
export default defineConfig({
plugins: [vue()],
resolve: {
// !!!!配置路径别名!!!
alias: {
'@': path.resolve(__dirname,'./src'),
}
},
});
- 在 tsconfig.json中配置代码
"compilerOptions":{
"baseUrl": "./",
"paths":{
"@": ["src"],
"@/*": ["src/*"],
}
}
（4）启动服务
在vscode中打开，在doc上启动服务
npm run dev
1.3 版本要求
flask-socketio的版本兼容
| JavaScript Socket.IO version | Socket.IO protocol revision | Engine.IO protocol revision | Flask-SocketIO version | python-socketio version | python-engineio version |
|---|---|---|---|---|---|
| 0.9.x | 1, 2 | 1, 2 | Not supported | Not supported | Not supported |
| 1.x and 2.x | 3, 4 | 3 | 4.x | 4.x | 3.x |
| 3.x and 4.x | 5 | 4 | 5.x | 5.x | 4.x |
1.4 Flask-SocketIO消息隔离
我认为有三种隔离级别
| 隔离级别 | 对应策略 | 说明 |
|---|---|---|
| 全局 | / | 全局命名空间，连接到此命名空间下的客户端会收到消息，系统默认空间 |
| 空间 | namespace | 特定命名空间，连接到此命名空间的客户端会收到消息，不同命名空间相互隔离 |
| 组 | room | 在某命名空间下的用户组，不同用户组仅仅接收自己所在组的消息，不同组消息隔离 |
2 Flask项目
2.1 项目布局
2.2 源代码
main.py
from blueprint_user import init_blueprint
from config_app import socketio, app
# 初始化蓝本
from socket_comm import init_socket
init_blueprint()
init_socket()
if __name__ == '__main__':
# 注意：应用会重重加载两次，应该是flask加载1次，socketio又加载1次。
socketio.run(app, host='0.0.0.0', port=5000, debug=True)
config_app.py
from flask import Flask
from flask_cors import CORS
from flask_socketio import SocketIO
def create_app():
flask_app = Flask(__name__)
# 设置密钥
flask_app.config['token'] = '123456'
CORS(flask_app, supports_credentials=True)
return flask_app
def create_socketio():
flask_socketio = SocketIO()
# 解决跨域问题
flask_socketio.init_app(app, cors_allowed_origins='*')
return flask_socketio
app = create_app()
socketio = create_socketio()
name_space_user = "/user"
blueprint_user.py
from flask import Blueprint
from config_app import app, name_space_user, socketio
user = Blueprint("user", __name__)
# 注册蓝本
def init_blueprint():
app.register_blueprint(user, url_prefix='/user')
@user.route('/')
def index():
return "Success"
# 使用方法传递参数
@user.route('/broad')
def broad_event():
event_name = "data_res"
broad_casted_data = {'data': "test message!"}
# 发送消息
socketio.emit(event_name, broad_casted_data, namespace=name_space_user)
return "success"
socket_comm.py
from flask_socketio import emit, send, join_room, leave_room
from config_app import socketio, name_space_user
def init_socket():
pass
# 可以用“@socketio.event”替代“@socketio.on('connect')”
# “@socketio.event”可以用来装饰socketio默认的事件，例如：connect等
@socketio.on('connect', namespace=name_space_user)
def connected_msg(auth:dict):
print(auth)
print('client connected.')
if(auth.get("token") == "123"):
return True
else:
return False
@socketio.on('disconnect', namespace=name_space_user)
def disconnect_msg():
print('client disconnected.')
@socketio.on('data_event', namespace=name_space_user)
def test_event(message):
print(message)
# emit(): 发送到指定活动上，对应前端的data_res
"""
# 对应前端的代码
socket.on('data_res', (data:string) => {
console.log('监听消息：');
console.log(data);
});
"""
emit('data_res', message, broadcast=True)
@socketio.on('data_msg', namespace=name_space_user)
def test_msg(message):
print(message)
# send(): 发送到message,对应前端的message
"""
# 对饮前端的代码
socket.on("message", function(data: string){
console.log("Message" + data)
})
"""
send(message, namespace=name_space_user, broadcast=True)
@socketio.on('join', namespace=name_space_user)
def on_join(data):
username = data['username']
room = data['room']
join_room(room)
print(data)
send(username + ' has entered the room.', to=room)
@socketio.on('leave', namespace=name_space_user)
def on_leave(data):
username = data['username']
room = data['room']
leave_room(room)
print(data)
send(username + ' has left the room.', to=room)
@socketio.on('send_room', namespace=name_space_user)
def send_room(data):
username = data['username']
room = data['room']
print(data)
send(data, to=room)
"""
解决flask-socketio使用线程或者while循环，前端无法实时接收到数据的解决方法有两个。
方法1：在socketio.init_app方法中添加async_mode='threading'参数（flask-socketio在启动时使用的是协程；）；
flask_socketio = SocketIO()
flask_socketio.init_app(flask_app, cors_allowed_origins='*', async_mode='threading')
方法2：使用socketio.start_background_task(target=background_thread)函数，不过此可以解决while问题，但无法解决线程问题。
"""
3 Vue项目
构建项目使用的是“组合式 API (Composition API)”编写，不是“选项式 API (Options API)”
3.1 项目布局
3.2 源代码
main.ts
import { createApp } from 'vue'
import './style.css'
import App from './App.vue'
createApp(App).mount('#app')
vite.config.ts
import { defineConfig } from 'vite'
import vue from '@vitejs/plugin-vue'
import path from "path";
// https://vitejs.dev/config/
export default defineConfig({
// 配置插件列表
plugins: [vue()],
resolve: {
// !!!!配置路径别名!!!
alias: {
'@': path.resolve(__dirname,'./src'),
},
},
// 打包配置
build: {
target: 'modules',
// 设置输出路径
outDir: 'dist'
},
// 本地运行配置，及反向代理配置
server: {
// 默认启用并允许任何源
cors: true,
// 使用默认浏览器中打开应用程序
open: true,
// 设置本地端口
port: 4000,
// 在本地开发环境中，设置反向代理配置，注意配置rewrite
proxy: {
// '/api': {
// // 设置代理接口访问实际地址
// target: 'http://localhost/5000',
// changeOrigin: true,
// // 允许websocket代理
// ws: true,
// // 将api替换为空
// rewrite: (path) => path.replace(/^\/api/, '')
// },
// '/socket.io': {
// target: `ws://127.0.0.1:5000`,
// ws: true,
// changeOrigin: true
// },
}
}
})
App.vue
<script setup lang="ts">
import HelloWorld from './components/HelloWorld.vue'
</script>
<template>
<div class="app">
<HelloWorld msg="app" />
</div>
</template>
<style scoped>
.app{
width: 100%;
height: 100%;
}
</style>
在”config/ws“目录下：
index.ts
/*
import { io } from 'socket.io-client';
// 第一种连接方法
// 参考地址：https://socket.io/docs/v4/client-api/#io
export function create_socket(token: string) {
const socket = io('http://127.0.0.1:5000', {
// 指定传输方式，如WebSocket
transports: ['websocket'],
// 是否自动连接
autoConnect: true,
// 是否自动重新连接
reconnection: true,
// 重新连接尝试次数
reconnectionAttempts: 3,
// 重新连接延迟时间（毫秒）
reconnectionDelay: 1000,
// 自定义查询参数
// query: {
// token: token
// },
// 其他可选参数...
});
return socket;
}
*/
// 第二种连接方法
// 参考地址：https://socket.io/docs/v4/client-api/#manager
import { Manager } from "socket.io-client";
export function create_socket(token: string) {
const manager = new Manager("ws://127.0.0.1:5000", {
// 指定传输方式，如WebSocket
transports: ['websocket'],
// 是否自动连接
autoConnect: true,
// 是否自动重新连接
reconnection: true,
// 重新连接尝试次数
reconnectionAttempts: 3,
// 重新连接延迟时间（毫秒）
reconnectionDelay: 1000,
// 自定义查询参数
// query: {
// token: token
// },
// 其他可选参数...
});
const socket = manager.socket("/user", {
auth: {
token: token
}
});
return socket
}
在”components“目录下：
HelloWorld.vue
<script setup lang="ts">
import { ref, onMounted } from 'vue'
import { create_socket } from '@/config/ws/index.ts';
const socket = create_socket("123");
defineProps<{ msg: string }>()
function re_open_conn(){
// // 第一种方法连接
let connect_res = socket.open();
console.log(connect_res)
// console.log(connect_res.connected)
// 第二种方法连接
// 默认通道 connect是通道名称
// socket.on('connect', () => {
// console.log('连接成功');
// });
}
function close_conn(){
// 关闭连接，disconnect_res是一个Socket对象
// socket.disconnect()和socket.close()作用相同
// 对应后端的代码
// @socketio.on('disconnect', namespace=name_space_user)
let disconnect_res = socket.disconnect()
console.log(disconnect_res)
// 下面这种方法后端没反应
// socket.on("disconnect", () => {
// console.log("关闭连接")
// });
}
// 响应式状态
let msg = ref("测试")
function send_msg(){
console.log("data_event" + msg.value)
socket.emit("data_event", "Event " + msg.value)
console.log(" data_msg "+ msg.value);
socket.emit("data_msg", "Message " + msg.value)
}
// 更改状态、触发更新的函数
function listen_event() {
socket.on('data_res', (data:string) => {
console.log('监听事件(event)-对应后端emit()方法');
console.log(data);
});
socket.on("message", function(data:string){
console.log("监听消息(message)-对应后端send()方法:")
console.log(data)
});
}
onMounted(()=>{
listen_event()
});
function join_room(){
socket.emit("join", {"username":"zhangsan", "room":"room1"})
}
function leave_room(){
socket.emit("leave", {"username":"zhangsan", "room":"room1"})
}
function send_room(){
socket.emit("send_room", {"username":"zhangsan", "room":"room1"})
}
</script>
<template>
<div class="hello">
<button v-on:click="re_open_conn">打开连接</button>
<button v-on:click="close_conn">关闭连接</button>
<button v-on:click="send_msg">发送消息</button>
<input v-model="msg"/>
<br/>
<button v-on:click="join_room">进入房间</button>
<button v-on:click="leave_room">离开房间</button>
<button v-on:click="send_room">房间发送消息</button>
</div>
</template>
<style scoped>
.hello {
color: #888;
}
</style>
index.html
<!doctype html>
<html lang="en">
<head>
<meta charset="UTF-8" />
<link rel="icon" type="image/svg+xml" href="/vite.svg" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>Vite + Vue + TS</title>
</head>
<body>
<div id="app"></div>
<script type="module" src="/src/main.ts"></script>
</body>
</html>
4 运行结果
前端
后端
前往低代码交流专区
更多推荐
9
0- 0
已为社区贡献7条内容
所有评论(0)