FastAPI 中的 WebSockets - ConnectionClosedOK：收到 1000（正常）

我有 3 个客户端，他们定期向我的服务器发送数据。我使用的例子来自FastAPI 的文档 https://fastapi.tiangolo.com/advanced/websockets/.

这是我的服务器代码：

class ConnectionManager:
    def __init__(self):
        self.active_connections: list[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def send_response(self, message: dict, websocket: WebSocket):
        await websocket.send_json(message)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/chargeStationState/{client_id}")
async def websocket_endpoint(websocket: WebSocket,
                             client_id: int,
                             db: Session = Depends(deps.get_db)):

    await manager.connect(websocket)
    try:
        while True:
            message = await websocket.receive_json()
            logging.info(message)
          
            ## read data from db
            response = {
                             "stations": "repsonse",
                             "timestamp": int(time.time())
            }
            await manager.send_response(response, websocket)
            #await manager.broadcast(f"Client #{client_id} says: {data}")

    except WebSocketDisconnect:
        manager.disconnect(websocket)

这是客户端代码：

async for websocket in websockets.connect("ws://127.0.0.1:8001/chargeStationState/1"):
            message = {'name':'station1'}
            await websocket.send(json.dumps(message))
            p = await asyncio.wait_for(websocket.recv(), timeout=10)
            print(p)
            await asyncio.sleep(2)

所以，我想要 5 个客户端，他们将与我的服务器通信并发送传感器数据，但 5 分钟后我收到以下错误

websockets.exceptions.ConnectionClosedOK: received 1000 (OK) 

然后接收1000 (OK)并且无法确定问题出在哪里。

测试您提供的代码，似乎无法重现您所指的问题。因此，问题可能出在代码的其他地方。下面根据您问题中提供的代码提供了一个工作示例。相关例子可以找到here https://stackoverflow.com/a/70996841/17865804， 也here https://stackoverflow.com/a/71415451/17865804 and here https://stackoverflow.com/a/71871482/17865804.

工作示例

app.py

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import uvicorn
import time

class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def send_personal_message(self, message: str, websocket: WebSocket):
        await websocket.send_json(message)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_json(message)


app = FastAPI()
manager = ConnectionManager()

@app.websocket('/chargeStationState')
async def websocket_endpoint(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_json()
            response = {'station': data['station'], 'timestamp': time.ctime()}
            await manager.send_personal_message(response, websocket)
    except WebSocketDisconnect:
        manager.disconnect(websocket)

test.py

import websockets
import asyncio
import json

async def main():
    url = 'ws://127.0.0.1:8000/chargeStationState'
    data = json.dumps({'station':'1'})
    
    async for websocket in websockets.connect(url):
        await websocket.send(data)
        print(await asyncio.wait_for(websocket.recv(), timeout=10))
        await asyncio.sleep(2)
        
asyncio.run(main())

test.py（修改）

在之前的test.py代码示例中，每次将数据发送到服务器时，websocket 连接都会关闭，并会建立一个新的连接（使用async for表达）。除非那是最适合您的，否则下面的代码示例将通过使用while里面的循环语句async for环形。这意味着首先建立的（相同的）连接将保持打开状态，除非循环体中发生异常，该异常的处理方式如下connect()达到，以便重新连接下一次迭代（否则，您可能会让异常冒泡并跳出循环）。如果建立连接时发生错误，connect()使用指数退避重试。退避延迟从三秒开始，最多增加一分钟（请参阅相关文件 https://websockets.readthedocs.io/en/stable/reference/client.html#opening-a-connection).

import websockets
import asyncio
import json

async def main():
    url = 'ws://127.0.0.1:8000/chargeStationState'
    data = json.dumps({'station':'1'})
    
    async for websocket in websockets.connect(url):
        try:
            while True:
                await websocket.send(data)
                print(await asyncio.wait_for(websocket.recv(), timeout=10))
                await asyncio.sleep(2)
        except websockets.ConnectionClosed:
            continue
        
asyncio.run(main())