실시간 통신 구현 방법

우리 챗봇 서비스 보면 글자 하나하나가 실시간으로 찍히는 것을 볼 수 있다.

일반적인 통신은 요청-응답의 형태를 가지는데,

챗봇에 경우 요청-응답-갱신갱신갱신의 형태를 가지는 것이다.

이러한 통신 방식을 SSE (Server-Sent Events)라고 하는데,

서버가 통신 유지하고 계속 클라이언트한테 데이터를 밀어 넣는 방식이다.

Polling: 일반적인 요청-응답으로 HTTP 프로토콜 기반

SSE: 한번 HTTP 통신으로 연결 수립하고 서버에서 단방향으로 데이터 push

WebSocket: 우리가 아는 실시간 채팅 양방향 통신

구현 방식

FastAPI와 React를 기준으로 일반적인 polling 통신 구조를 보자

@app.get("/poll")
async def poll_data():
    return {"value": random.randint(1, 100)}

서버는 API 하나 오픈하고 오면 데이터 넘겨주면 끝이다.

const fetchData = async () => {
    const res = await fetch('http://localhost:8000/poll');
    const json = await res.json();
};
fetchData();

React도 그냥 통신 한번 시도하면 끝이다.

그렇다면 SSE는?

async def event_generator():
    while True:
        data = "실시간 데이터..."
        yield f"data: {data}\n\n"
        await asyncio.sleep(1)

@app.get("/sse")
async def sse_endpoint(request: Request):
    return StreamingResponse(event_generator(), media_type="text/event-stream")

어차피 다 지원해 준다. StreamingResponse 함수로 매개변수에 실행할 함수만 넣어주면 된다.

실행 함수 내부에서는 yield 형식으로 데이터를 전송하면 된다.

const eventSource = new EventSource("http://localhost:8000/sse");

eventSource.onmessage = (event) => {
    setData(event.data);
};

React에서는 onmessage를 통해서 데이터가 들어올 때마다 처리해주면 된다.

연결은 어떻게 종료함?

만약 데이터 통신이 끝나서, FastAPI에서 loop를 빠져나오면(예시는 while True 지만) 통신은 종료되고 리소스가 반환된다.

하지만 클라이언트가 먼저 끊어버리는 경우가 문제다.

eventSource.close();

클라이언트가 아직 연결을 붙잡고 있는지를 확인하기 위해 FastAPI에서는 매 Loop 마다 request에 대한 검증을 추천한다.

if await request.is_disconnected():
    print("Client disconnected! Stopping stream.")
    break

주의 사항

연결 종료 뿐만 아니라 처리가 오래 걸려서 데이터 전송 없이 연결만 유지될 경우

로드 밸런서나 방화벽이 죽은 연결로 판단해서 끊어버릴 수 있다.

서비스에는 늘 장애가 있을 수 있음을 명시하자.

왜냐면 FastAPI 공식문서에 연결종료(is_disconnected) 검색하는데 사이트가 터졌기 때문이다.

우린 할 수 있는게 없다.

팝콘이나 가져와라