使用FastAPI同时搭建WebSocket服务端和HTTP服务端_fatspi如何实现websocket

使用FastAPI搭建WebSocket服务端和HTTP服务端

1. WebSocket协议

WebSocket协议是一种在单个TCP连接上进行全双工通信的网络协议。它提供了双向通信的能力，允许服务器和客户端之间进行实时数据传输。与HTTP不同，WebSocket在连接建立后保持打开状态，可以在任何时间点双向传输数据，而不需要每次请求都建立新的连接。在 WebSocket 协议握手阶段，客户端和服务器之间会通过 HTTP 协议进行握手升级，从而建立 WebSocket 连接。

WebSocket协议的关键特点包括：

• 全双工通信: 服务器和客户端可以同时向对方发送和接收数据。
• 持久连接: 连接在建立后保持打开状态，允许实时通信而无需重复建立连接。
• 轻量级消息格式: WebSocket消息的开销相对较小，因为它们不需要HTTP的头部信息。
• 低延迟: 由于保持了持久连接，可以实现实时性要求较高的应用，如聊天应用和实时游戏。

2. FastAPI

FastAPI 是一个现代、快速（高性能）的 web 框架，用于构建基于 Python 的 API。它是建立在 Starlette 和 Pydantic 之上的，这两者都是非常流行的 Python 库。FastAPI 旨在提供简单易用的 API 开发体验，同时保持高性能和强大的功能。以下是 FastAPI 的一些特点和优势：

• 快速高效：FastAPI 极其快速，并且具有异步请求处理的能力，这使得它成为处理高并发请求的理想选择。
• 基于标准：FastAPI 基于标准的 Python 类型提示和标准的 Python 异步框架，这使得它易于学习和使用，并且能够充分利用 Python 的强大功能。
• 自动文档生成：FastAPI 可以自动生成交互式 API 文档，支持 Swagger UI 和 ReDoc，这样可以方便地查看和测试 API。
• 数据验证和序列化：FastAPI 集成了 Pydantic，可以方便地进行数据验证和序列化，提供了强大的数据校验功能。
• 依赖注入系统：FastAPI 内置了依赖注入系统，可以轻松管理 API 中的依赖关系。
• 简单易用：FastAPI 设计简洁、直观，提供了易于理解的 API 开发体验，使得开发者能够快速构建出高效、稳健的 API。

总的来说，FastAPI 是一个功能强大、性能优越的 Python web 框架，适用于构建各种类型的 API 服务。它的简单易用性、高性能和标准化特性使得它成为了众多开发者的首选框架之一。

3. 使用FastAPI搭建WebSocket服务端

使用FastAPI + uvicorn可以快速搭建HTTP服务端，而WebSocket是基于HTTP协议的协议，所以FastAPI支持同时搭建HTTP和WebSocket服务。也就是说，你可以在一个FastAPI的应用中，同时提供HTTP和WebSocket服务，以下是代码示例。

import uvicorn
from fastapi import FastAPI, WebSocket
from loguru import logger


class WebsocketServer:
    def __init__(self):
        self.wbs_client = None
        self.wbs_server: FastAPI = FastAPI()
        self.wbs_server.add_api_websocket_route(path='/test_websocket_path', endpoint=self.test_wbs_path_connection)
        self.wbs_server.add_api_route(path='/test_http_path', endpoint=self.test_http_path_connection, methods=['GET'])

    async def test_wbs_path_connection(self, websocket: WebSocket):
        self.wbs_client = websocket
        await websocket.accept()
        while True:
            data = await websocket.receive_text()
            logger.info(f'收到消息：{data}')
            await self.wbs_client.send_text(data)
            logger.info(f'发送消息：{data}')

    async def test_http_path_connection(self):
        return 'test http'

    def start_serving(self):
        uvicorn.run(self.wbs_server, host='0.0.0.0', port=8001, loop='asyncio')


def main():
    websocket_server = WebsocketServer()
    websocket_server.start_serving()


if __name__ == '__main__':
    main()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

其中，代码self.wbs_server.add_api_websocket_route(path='/test_websocket_path', endpoint=self.test_wbs_path_connection)添加了一个WebSocket路径，在客户端，使用地址ws://127.0.0.1:8001/test_websocket_path即可连接。

代码self.wbs_server.add_api_route(path='/test_http_path', endpoint=self.test_http_path_connection, methods=['GET'])添加了一个HTTP GET服务路径，在客户端，使用地址http://127.0.0.1:8001/test_http_path即可进行访问。