FastAPI websocket 无法处理大量数据传入？

Posted 2023-02-16

【中文标题】FastAPI websocket 无法处理大量数据传入？

【英文标题】：FastAPI websocket can't handle large incoming of data?

【发布时间】：2020-09-28 09:37:38

【问题描述】：

我已经使用 python 和 FastAPI 创建了一个 websocket 服务器：https://fastapi.tiangolo.com/

而且.. 完成了，当然最后，我发现了一个让我有些恐慌的新“错误”。

当连接的客户端刚刚发送大量数据时，他会突然断开连接。我不确定是websocket代码本身，是asyncio，还是uvicorn，docker，还是linux服务器本身无法处理大量传入的数据？

这是我在 dockercontainer 中运行的服务器代码：

import json
import time

import socketio

from fastapi import FastAPI, WebSocket
from fastapi.responses import htmlResponse

from starlette.websockets import WebSocketDisconnect


from typing import List


app = FastAPI()


sio = socketio.AsyncServer(async_mode='asgi')
socket_app = socketio.ASGIApp(sio, static_files='/': 'app.html')
background_task_started = False


from streaming_class import repetisjons_klasse


import asyncio
from pydantic import BaseModel

class UserClientWebSocket(BaseModel):
    id: str
    ws: WebSocket

    class Config:
        arbitrary_types_allowed = True


class WebPageClientWebSocket(BaseModel):
    id: str
    ws: WebSocket

    class Config:
        arbitrary_types_allowed = True

class ConnectionManager:
    def __init__(self):

        self.active_user_client_connections: List[UserClientWebSocket] = []
        self.collect_user_IDs = []


        self.active_webpage_client_connections: List[WebPageClientWebSocket] = []
        self.collect_webpages_IDs = []


    async def connect_the_user_client(self, websocket: WebSocket, THE_USER_ID):
        await websocket.accept()
        await self.send_message_to_absolutely_everybody(f"User: THE_USER_ID connected to server!")
        print("User:  ".format(THE_USER_ID) + " Connected")
        if THE_USER_ID not in self.collect_user_IDs:
            self.collect_user_IDs.append(THE_USER_ID)
        else:
            await self.send_message_to_absolutely_everybody(f"Somebody connected with the same ID as client: THE_USER_ID")
            await self.send_message_to_absolutely_everybody("but Vlori is a nice and kind guy, so he wil not get kicked :)")
            self.collect_user_IDs.append(THE_USER_ID)
        self.active_user_client_connections.append(UserClientWebSocket(ws=websocket, id=THE_USER_ID)) #SJEEKK DENNE LINJA !!!!!
        await self.show_number_of_clients()



    async def connect_the_webpage_client(self, websocket: WebSocket, the_webpage_id):
        await websocket.accept()
        await self.send_message_to_absolutely_everybody(f"User: the_webpage_id connected to server through webbrowser!")
        print("User:  ".format(the_webpage_id) + " Connected")
        if the_webpage_id not in self.collect_webpages_IDs:
            self.collect_webpages_IDs.append(the_webpage_id)
        else:
            await self.send_message_to_absolutely_everybody(f"Somebody connected with the same ID as client: the_webpage_id")
            await self.send_message_to_absolutely_everybody("but Vlori is a nice and kind guy, so he wil not get kicked :)")
            self.collect_webpages_IDs.append(the_webpage_id)
        self.active_webpage_client_connections.append(WebPageClientWebSocket(ws=websocket, id=the_webpage_id)) #SJEEKK DENNE LINJA !!!!!
        await self.show_number_of_clients()

    async def disconnect_the_webpage_client(self, websocket: WebSocket, the_webpage_id):
        await websocket.close(code=1000)
        self.collect_webpages_IDs.remove(the_webpage_id)
        self.active_webpage_client_connections.remove(WebPageClientWebSocket(ws=websocket, id=the_webpage_id))
        await self.show_number_of_clients()


    async def disconnect_the_user_client(self, websocket: WebSocket, THE_USER_ID):
        await websocket.close(code=1000)
        self.collect_user_IDs.remove(THE_USER_ID)
        self.active_user_client_connections.remove(UserClientWebSocket(ws=websocket, id=THE_USER_ID))
        await self.show_number_of_clients()





"""
PROBLEM: THE USER GETS DISCONNECTED WHEN SENDING TO MUCH DATA IN REAL TIME!
"""

@app.websocket("/ws/testchannel")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            print("Received data:  ".format(data))
            #await websocket.send_text(f"you sent message: data")
            ¤await connection_manager.send_message_to_absolutely_everybody(data)

      
    except WebSocketDisconnect:
        print("client left chat.")

以下是我在 dockerfile 中运行的设置（可能是这里的东西吗？我不确定）：

FROM ubuntu:latest
FROM python:3

MAINTAINER raxor2k "xxx.com"

RUN apt-get update -y

RUN apt-get install -y python3-pip build-essential python3-dev

COPY . /app
WORKDIR /app

RUN pip3 install --upgrade pip
RUN pip3 install -r requirements.txt
RUN pip3 install fastapi uvicorn #dennekanfjernes?

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "80", "--reload"]

所以，这是我在本地计算机上运行的 client.py（运行良好）：

#client.py

from websocket import create_connection
import json
import time

ws = create_connection("ws://134.122.76.213:5080/ws/testchannel")

time.sleep(1)


def send_json_all_the_time(position):
    generate_json =  "machineID":"001", "RepSensor": position
    send_json = json.dumps(generate_json)
    print("JSON SENT FROM IoT SENSOR: ".format(send_json))
    time.sleep(0.1)
    ws.send(json.dumps(send_json))
    time.sleep(0.1)


while True:
    for x in range(1000):
        send_json_all_the_time(x)

    for x in range(100, -1, -1):
        ws.send("pause a little bit, starting again soon!")
        send_json_all_the_time(x)

所以我很困惑..我的服务器代码是否有错误？是linux服务器的问题吗？是dockercontainer里面的问题吗？

任何帮助将不胜感激！

【问题讨论】：

通常不单独运行 uvicorn，因为it's documentation 建议您使用 gunicorn 进行生产部署（我建议您使用 tiangolo' official image）。除此之外，您的应用程序可能面临多种原因会导致它断开连接。请包含您的客户日志以供进一步分析。

您好。我的服务器没有任何问题，问题出在客户端，因为 websocket 客户端没有包含 ping-pong 同步。现在一切正常:)

很高兴知道！然后考虑回答您自己的问题，因此我们添加到 Stack Overflow 的知识库

【参考方案1】：

编辑：经过 FastAPI“论坛”的一些研究、测试和反馈：https://gitter.im/tiangolo/fastapi

由于缺少乒乓同步，问题出在客户端上。服务器端没有任何问题。

【讨论】：

我遇到了同样的问题，你能详细说明一下这个解决方案吗？

哦，我修好了，客户端也需要实现乒乓同步。