「AIPyGo」FastAPI Vue实现socketio实时通信教程

安装依赖

pip install fastapi pip install uvicorn[standard] pip install socketio

main.py

import socketio from fastapi import FastAPI, Request from fastapi.responses import HTMLResponse from fastapi.templating import Jinja2Templates from fastapi.staticfiles import StaticFiles fast_app = FastAPI( openapi_url='/api/chat/openapi.json', docs_url='/api/chat/docs', redoc_url='/api/chat/redoc' ) mgr = socketio.AsyncRedisManager('redis://') # 使用redis作为消息队列 sio = socketio.AsyncServer( # 使用socketio async_mode='asgi', cors_allowed_origins='*', client_manager=mgr ) app = socketio.ASGIApp( # 整合socketio和fastapi socketio_server=sio, other_asgi_app=fast_app, socketio_path='/socket.io' ) # 挂载目录 fast_app.mount("/static", StaticFiles(directory="static"), name="static") # 模板 templates = Jinja2Templates(directory="templates") @fast_app.get("/", response_class=HTMLResponse) async def get(request: Request): return templates.TemplateResponse("index.html", {"request": request, "id": "hello world"}) @sio.on('join') async def handle_join(sid, *args, **kwargs): print("与客户端建立了连接") print('sid:', sid) print('args:', args) print('kwargs:', kwargs) user_dict = args[0] user_dict.update({'sid': sid}) print("用户字典：", user_dict) await sio.emit('chat', '恭喜您登录成功') # 服务器也可以响应字典 await sio.emit('chat', user_dict) @sio.on('chat') async def handle_chat(sid, *args, **kwargs): print("接收到客户端的会话。。。") print("sid:", sid) print("args:", args) print("kwargs:", kwargs) await sio.emit('chat', f'服务器响应回来的{args[0]}')

运行

uvicorn main:app --reload创建客户端

核心代码

import io from "socket.io-client"; let socket = io.connect("http://localhost:8000"); socket.on('connect', function (data) { // 连接成功以后向服务器发送一个join登录的事件 // json支持的数据，socketio都支持 socket.emit('join', {username: 'lxgzhw', password: 'lxgzhw', age: 22, gender: true}); });

创建项目

npm install yarn -g vue create vue_demo cd vue_demo yarn add socket.io-client

修改main.js

import Vue from 'vue' import App from './App.vue' import router from './router' import store from './store' import io from "socket.io-client"; let socket = io.connect("http://localhost:8000"); socket.on('connect', function (data) { // 连接成功以后向服务器发送一个join登录的事件 // json支持的数据，socketio都支持 socket.emit('join', {username: 'lxgzhw', password: 'lxgzhw', age: 22, gender: true}); }); for (let i = 0; i < 10; i ) { socket.emit('chat', {username: 'lxgzhw' i, password: 'lxgzhw', age: 22, gender: true}); } Vue.config.productionTip = false new Vue({ router, store, render: h => h(App) }).$mount('#app')

运行

yarn serve连接创建连接

服务器启动，就创建了一个连接。

客户端通过指定服务器的端口和空间，来建立双向连接。

import io from "socket.io-client"; let socket = io.connect("http://localhost:8000/chat"); socket.on('connect', function (data) { console.log('与服务器建立连接，开始向服务器提交事件') socket.emit('join', {username: 'lxgzhw', password: 'lxgzhw', age: 22, gender: true}, function (a, b) { console.log("事件的回调数据：", a, b) }); ... });

客户端通过向服务器发送一个事件，服务器监听该事件，来判断双向连接是否建立成功。

@sio.on('join', namespace='/chat') async def handle_join(sid, *args, **kwargs): print("与客户端建立了连接") print(sid, args, kwargs) await sio.emit('client_chat', {'data': 'foobar'}) return 'ok', 123断开连接

客户端调用方法来断开连接

socket.disconnect()

服务器监听客户端的端口连接事件，来主动断开客户端的连接。以确保双向连接全部断开。

@sio.on('disconnect', namespace='/chat') async def disconnect(sid, *args, **kwargs): # 进入房间 print(f'用户{sid}断开连接') # 服务器断开连接 await sio.disconnect(sid)空间

使用空间namespace

服务器：只需要给监听的事件指定namespace

@sio.on('join', namespace='/chat') async def handle_join(sid, *args, **kwargs): ... @sio.on('chat', namespace='/chat') async def handle_chat(sid, *args, **kwargs): ...客户端

客户端：只需要在连接的时候以路径的方式指定空间名

import io from "socket.io-client"; // 语法：http://localhost:8000/空间名 let socket = io.connect("http://localhost:8000/chat"); socket.on('connect', function (data) { ... });房间进入房间

客户端通过提交一个事件进入房间

socket.on('connect', function (data) { // 进入房间 socket.emit('begin_chat'); }); socket.on('chat', function (data) { console.log("服务器的响应：", data) });

服务器通过监听该事件让用户进入指定的房间

@sio.on('begin_chat', namespace='/chat') async def begin_chat(sid, *args, **kwargs): # 进入房间 sio.enter_room(sid, 'chat_users', '/chat') print(f'用户{sid}进入房间') await sio.emit('chat', '恭喜你进入房间', namespace='/chat')离开房间

客户端通过提交事件来通知服务器要退出房间

socket.emit("exit_chat", "aaa")

服务器通过监听该事件来让客户端退出房间

@sio.on('exit_chat', namespace='/chat') async def exit_chat(sid, *args, **kwargs): # 进入房间 sio.leave_room(sid, 'chat_users') print(f'用户{sid}离开房间') await sio.emit('chat', '你离开了房间', namespace='/chat')提交事件事件回调

服务器端

@sio.on('join', namespace='/chat') async def handle_join(sid, *args, **kwargs): print("与客户端建立了连接") print(sid, args, kwargs) await sio.emit('chat', '恭喜您登录成功') return 'ok', 123

客户端

socket.on('connect', function (data) { console.log('与服务器建立连接，开始向服务器提交事件') socket.emit('join', {username: 'lxgzhw', password: 'lxgzhw', age: 22, gender: true}, function (a, b) { console.log("事件的回调数据：", a, b) }); });监听事件监听namespace事件

如果连接的时候指定了namespace，name监听的时候，也是监听namespace下的事件。这就要求在传递的时候，必须要指定namespace进行传递。

服务器代码

import socketio from fastapi import FastAPI fast_app = FastAPI( openapi_url='/api/chat/openapi.json', docs_url='/api/chat/docs', redoc_url='/api/chat/redoc' ) mgr = socketio.AsyncRedisManager('redis://') # 使用redis作为消息队列 sio = socketio.AsyncServer( # 使用socketio async_mode='asgi', cors_allowed_origins='*', client_manager=mgr ) app = socketio.ASGIApp( # 整合socketio和fastapi socketio_server=sio, other_asgi_app=fast_app, socketio_path='/socket.io' ) @sio.on('join', namespace='/chat') async def handle_join(sid, *args, **kwargs): print("与客户端建立了连接") print(sid, args, kwargs) await sio.emit('client_chat', {'data': 'foobar'}) return 'ok', 123 @sio.on('chat', namespace='/chat') async def handle_chat(sid, *args, **kwargs): print("监听到了客户端的事件：", sid, args, kwargs) await sio.emit('chat', f'服务器响应回来的{args[0]}', namespace='/chat')

客户端代码

import io from "socket.io-client"; let socket = io.connect("http://localhost:8000/chat"); socket.on('connect', function (data) { console.log('与服务器建立连接，开始向服务器提交事件') socket.emit('join', {username: 'lxgzhw', password: 'lxgzhw', age: 22, gender: true}, function (a, b) { console.log("事件的回调数据：", a, b) }); }); socket.emit("chat", "aaa") socket.on('chat', function (data) { console.log("服务器的响应：", data) });用户会话

@sio.event async def connect(sid, environ): username = authenticate_user(environ) await sio.save_session(sid, {'username': username}) @sio.event async def message(sid, data): session = await sio.get_session(sid) print('message from ', session['username'])