FastAPI WebSocket ping/pong 超时

在本文中，我们将介绍如何在 FastAPI 中处理 WebSocket 的 ping/pong 超时问题。

FastAPI 简介

FastAPI 是一个现代、快速（高性能）的 Web 框架，用于构建 Web API。它基于 Python 3.7+ 标准并使用了新的 Python 特性，如类型注解和异步语法。FastAPI 高度集成了 Starlette 框架，从而提供了一种高性能、易于使用的开发方式。

WebSocket 超时问题

WebSocket 是一种在客户端和服务器之间进行全双工通信的协议，能够在已经建立连接的情况下保持连接状态。为了保持连接的活动状态，WebSocket 协议定义了 ping/pong 消息。当服务器发送 ping 消息给客户端时，客户端需要及时回复一个 pong 消息以表明连接是活跃的。如果客户端没有在指定的时间内发送 pong 消息，服务器将认为连接已断开并关闭连接。这个时间限制就是 ping/pong 超时时间。

在 FastAPI 中，默认的 WebSocket 超时时间是 5 秒。如果你的应用有更高的性能要求或者需要处理更大的数据量，可能需要增加超时时间。

修改 WebSocket 超时时间

要修改 FastAPI 中的 WebSocket 超时时间，可以使用 websocket_timeout 参数。例如：

from fastapi import FastAPI

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()

上述代码中的 websocket 参数是一个 WebSocket 类型的对象，它有一个 websocket_timeout 属性可以用于设置超时时间。

要将 WebSocket 超时时间设置为 10 秒，可以这样修改代码：

from fastapi import FastAPI
from starlette.websockets import WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    websocket.websocket_timeout = 10

上述代码将 WebSocket 超时时间设置为 10 秒。

示例

下面是一个完整的示例，用于演示如何在 FastAPI 中修改 WebSocket 的 ping/pong 超时时间：

from fastapi import FastAPI
from starlette.websockets import WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    websocket.websocket_timeout = 10

    while True:
        try:
            message = await websocket.receive_text()
            await websocket.send_text(f"Received message: {message}")
        except WebSocketDisconnect:
            break

上述代码中，WebSocket 连接建立后，设置了超时时间为 10 秒。然后，通过 websocket.receive_text() 方法接收来自客户端的消息，并用 websocket.send_text() 方法将消息发送回客户端。

总结

本文介绍了在 FastAPI 中处理 WebSocket 的 ping/pong 超时问题的方法。通过修改 WebSocket 对象的 websocket_timeout 属性，可以更改超时时间。使用示例代码可以帮助你快速上手。使用 FastAPI，你可以轻松构建高性能的 WebSocket API。

希望本文对你理解和使用 FastAPI 中的 WebSocket 功能有所帮助！