做试用网站的原理,网站备案有什么好处理,长春做个人网站做不了,浙江壹设软装设计有限公司高级实时通信#xff1a;基于 Python 的 WebSocket 实现与异步推送解决方案 目录
#x1f7e2; WebSocket 协议概述#x1f535; 在 FastAPI 中实现 WebSocket#x1f7e3; Django Channels 实现异步实时通信#x1f534; 使用 Redis 实现实时推送 #x1f7e2; 1. WebS…高级实时通信基于 Python 的 WebSocket 实现与异步推送解决方案 目录 WebSocket 协议概述 在 FastAPI 中实现 WebSocket Django Channels 实现异步实时通信 使用 Redis 实现实时推送 1. WebSocket 协议概述
WebSocket 是 HTML5 规范中提出的一种新协议旨在实现客户端与服务器之间的全双工通信。与传统的 HTTP 请求/响应模型不同WebSocket 协议允许持久的、双向的连接客户端和服务器可以在任意时刻相互发送数据而无需重新发起 HTTP 请求。WebSocket 可以减少通信延迟和网络资源消耗特别适用于实时应用程序如聊天系统、股票交易、游戏、以及其他需要即时数据更新的场景。 WebSocket 的工作原理
WebSocket 的工作流程可以简单概括如下
初始握手客户端通过 HTTP 发起 WebSocket 连接请求服务器通过特殊的响应头 Upgrade 升级协议建立连接。持久连接一旦连接建立客户端和服务器之间的通信通道就保持打开直到一方主动关闭连接。双方可以同时发送消息消息是基于帧的传输方式。双向通信与传统的 HTTP 轮询不同WebSocket 是全双工的意味着服务器和客户端可以随时相互发送数据。消息可以是文本帧或者二进制帧且通过 WebSocket 协议传输的消息有较小的开销适合频繁的小数据传输场景。
⚡️ WebSocket 和 HTTP 的区别
连接方式HTTP 是短连接每次请求/响应后连接断开而 WebSocket 是持久化连接只有在连接关闭时才断开。通信方向HTTP 是客户端发起请求服务器响应WebSocket 支持双向通信客户端和服务器都可以主动发送消息。协议开销WebSocket 建立连接时会使用 HTTP 协议握手但后续通信数据帧头信息非常小开销比 HTTP 低。
WebSocket 的这些特性使其成为实时应用开发中的重要工具。 2. 在 FastAPI 中实现 WebSocket
FastAPI 是一个现代的、快速的 Web 框架提供了对 WebSocket 支持的简便实现。FastAPI 内部集成了 ASGIAsynchronous Server Gateway Interface这意味着它能够轻松处理异步通信任务比如 WebSocket。以下是如何在 FastAPI 中实现 WebSocket 通信的详细步骤。 FastAPI 中的 WebSocket 代码实现
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import Listapp FastAPI()# 管理多个 WebSocket 连接的管理器
class ConnectionManager:def __init__(self):self.active_connections: List[WebSocket] []# 添加新连接async def connect(self, websocket: WebSocket):await websocket.accept()self.active_connections.append(websocket)# 移除断开的连接async def disconnect(self, websocket: WebSocket):self.active_connections.remove(websocket)# 向所有连接广播消息async def broadcast(self, message: str):for connection in self.active_connections:await connection.send_text(message)manager ConnectionManager()app.websocket(/ws/{client_id})
async def websocket_endpoint(websocket: WebSocket, client_id: int):await manager.connect(websocket)try:while True:# 接收客户端消息data await websocket.receive_text()await manager.broadcast(fClient #{client_id} says: {data})except WebSocketDisconnect:manager.disconnect(websocket)await manager.broadcast(fClient #{client_id} disconnected)代码解析
ConnectionManager 类用于管理 WebSocket 连接。它可以接受新连接、断开旧连接以及广播消息给所有连接的客户端。websocket_endpoint这个路径函数处理 WebSocket 连接。当客户端连接时它将接收并广播消息给其他连接的客户端。它使用 client_id 来标识连接的客户端。广播消息通过 manager.broadcast() 方法可以将来自某个客户端的消息发送给所有连接的客户端。 运行示例
要运行这个示例可以将代码保存为 main.py然后使用命令 uvicorn main:app --reload 启动 FastAPI 服务器。连接的客户端可以通过 /ws/{client_id} 连接并且所有客户端都能实时接收消息。
此示例展示了如何通过 FastAPI 简洁地实现 WebSocket 服务利用 Python 的异步特性来实现高效的实时通信。 3. Django Channels 实现异步实时通信
Django 默认是同步框架但通过 Django Channels可以为其添加异步功能尤其是支持 WebSocket 的实时通信功能。Django Channels 通过将请求分配给适当的消费者来处理异步通信而不需要重新设计整个 Django 应用。以下是如何使用 Django Channels 来实现 WebSocket 通信。 安装 Django Channels
首先需要安装 Django Channels
pip install channels然后在 settings.py 中添加 Channels 配置
INSTALLED_APPS [# 其他应用channels,
]# 指定 ASGI 应用
ASGI_APPLICATION myproject.asgi.application在项目根目录创建一个 asgi.py 文件
import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from myapp import routingos.environ.setdefault(DJANGO_SETTINGS_MODULE, myproject.settings)application ProtocolTypeRouter({http: get_asgi_application(),websocket: AuthMiddlewareStack(URLRouter(routing.websocket_urlpatterns)),
})Django Channels WebSocket 代码实现
创建 consumers.py 来定义 WebSocket 消费者
import json
from channels.generic.websocket import AsyncWebsocketConsumerclass ChatConsumer(AsyncWebsocketConsumer):async def connect(self):self.room_name chatawait self.channel_layer.group_add(self.room_name, self.channel_name)await self.accept()async def disconnect(self, close_code):await self.channel_layer.group_discard(self.room_name, self.channel_name)async def receive(self, text_data):message json.loads(text_data)[message]await self.channel_layer.group_send(self.room_name,{type: chat_message,message: message})async def chat_message(self, event):message event[message]await self.send(text_datajson.dumps({message: message}))在 routing.py 中定义 WebSocket 路由
from django.urls import re_path
from . import consumerswebsocket_urlpatterns [re_path(rws/chat/$, consumers.ChatConsumer.as_asgi()),
]代码解析
ChatConsumer这是 WebSocket 消费者类使用 AsyncWebsocketConsumer 来处理异步 WebSocket 连接。connect() 方法处理客户端连接disconnect() 方法处理断开receive() 用于接收并转发消息。channel_layerDjango Channels 使用频道层来在多个 WebSocket 连接之间共享数据。group_send() 方法用于向一个组中的所有 WebSocket 客户端广播消息。 运行示例
通过配置 Django Channels 后启动 Django 开发服务器并连接到 /ws/chat/ 端点即可开始实时通信。此实现展示了如何利用 Django 的扩展实现异步 WebSocket 通信方便现有 Django 项目无缝添加实时通信功能。 4. 使用 Redis 实现实时推送
Redis 是一个强大的内存数据存储系统支持发布/订阅Pub/Sub模式这使得它非常适合实现实时推送功能。通过将 Redis 集成到 WebSocket 应用中可以轻松地实现高效的实时数据推送服务。以下展示如何使用 Redis 结合 WebSocket 来实现消息的实时推送。 安装 Redis 和依赖
首先安装 Redis 和 aioredis 以在 Python 中使用异步 Redis 客户端
pip install aioredis实现基于 Redis 的 WebSocket 推送
在 WebSocket 服务中通过 Redis 的发布/订阅模式实现消息推送。
import aioredis
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import Listapp FastAPI()class ConnectionManager:def __init__(self):self.active_connections: List[WebSocket] []async def connect(self, websocket: WebSocket):await websocket.accept()self.active_connections.append(websocket)async def disconnect(self, websocket: WebSocket):self.active_connections.remove(websocket)async def broadcast(self, message: str):for connection in self.active_connections:await connection.send_text(message)manager ConnectionManager()app.on_event(startup)
async def startup_event():global redisredis await aioredis.create_redis_pool(redis://localhost)app.websocket(/ws)
async def websocket_endpoint(websocket: WebSocket):await manager.connect(websocket)try:while True:# 接收消息data await websocket.receive_text()await redis.publish(channel:1, data)except WebSocketDisconnect:manager.disconnect(websocket)app.on_event(shutdown)
async def shutdown_event():redis.close()await redis.wait_closed()Redis 订阅消息并推送给 WebSocket 客户端
实现 Redis 消息订阅和推送
import aioredis
from fastapi import FastAPIapp FastAPI()app.on_event(startup)
async def startup_event():redis await aioredis.create_redis_pool(redis://localhost)pubsub redis.pubsub()async def reader():await pubsub.subscribe(channel:1)async for message in pubsub.listen():print(fReceived message: {message[data]})app.loop.create_task(reader())app.on_event(shutdown)
async def shutdown_event():redis.close()await redis.wait_closed()代码解析
Redis 发布通过 redis.publish() 方法消息发布到指定的频道 channel:1这些消息随后会被其他 Redis 客户端订阅。Redis 订阅pubsub.listen() 用于订阅 Redis 频道上的消息收到消息后可以处理或转发给 WebSocket 客户端。 运行示例
启动 Redis 服务器并运行上述代码客户端通过 WebSocket 连接可以实时接收 Redis 频道中的推送消息。此示例展示了如何利用 Redis 高效的发布/订阅机制来实现实时消息推送结合 WebSocket 实现真正的全双工通信。
