DocsPricingBlog
Contact Us
Log inSign up for FREE
Back to Tech Articles

Python `websockets`와 ASGI를 이용한 초고속 독립형 WebSocket 서버 구축

Aug 16, 2025
# Misc

Lukas Schneider

DevOps Engineer · Leapcell

This post is also available in English, 日本語, and Deutsch
Cover of "Python `websockets`와 ASGI를 이용한 초고속 독립형 WebSocket 서버 구축"

실시간 통신 소개

오늘날 상호 연결된 디지털 환경에서 실시간 통신은 더 이상 사치가 아니라 기본적인 기대치입니다. 공동 문서 편집 및 라이브 채팅 애플리케이션부터 금융 거래 플랫폼 및 IoT 장치 모니터링에 이르기까지 정보를 즉각적이고 지속적으로 교환하는 능력은 매우 중요합니다. 다양한 사용 사례에 훌륭한 기존 HTTP 요청-응답 주기는 지속적이고 낮은 지연 시간의 양방향 통신이 필요한 경우 종종 부족합니다. 바로 여기서 WebSocket이 빛을 발합니다. WebSocket은 단일 TCP 연결을 통해 전이중 통신 채널을 제공하여 반복적인 HTTP 폴링에 비해 오버헤드를 크게 줄입니다. 많은 Python 웹 프레임워크에서 WebSocket 통합을 제공하지만, 완전한 웹 프레임워크에서 분리된 고성능 독립형 WebSocket 서버가 최적의 솔루션인 시나리오가 있습니다. 이 글에서는 Python의 강력한 websockets 라이브러리와 Asynchronous Server Gateway Interface (ASGI) 사양을 사용하여 이러한 서버를 구축하는 방법을 자세히 살펴보고 효율적인 실시간 기능을 활용합니다.

핵심 구성 요소 이해

구현 세부 사항을 살펴보기 전에 고성능 WebSocket 서버의 기반이 되는 핵심 기술을 명확히 합시다.

WebSockets: 언급했듯이 WebSocket은 단일 TCP 연결을 통해 전이중 통신을 가능하게 합니다. 이는 클라이언트와 서버가 각 교환을 위해 새 연결을 설정할 필요 없이 동시에 메시지를 송수신할 수 있음을 의미합니다. 이 지속적인 연결은 지연 시간과 오버헤드를 크게 줄여 실시간 상호 작용에 이상적입니다.

ASGI (Asynchronous Server Gateway Interface): ASGI는 Python 비동기 웹 서버, 프레임워크 및 애플리케이션을 위한 사양입니다. 비동기 웹 서버(Uvicorn 또는 Hypercorn와 같은)와 비동기 Python 웹 애플리케이션 간의 통신을 위한 표준 인터페이스를 정의합니다. ASGI 애플리케이션은 본질적으로 요청 세부 정보가 포함된 스코프(scope) 사전과 send 및 receive 함수를 통해 이벤트를 수신/송신하는 비동기 호출 가능 객체입니다. 이러한 표준화는 다양한 ASGI 서버 및 프레임워크 간의 상호 운용성을 허용하여 강력하고 유연한 생태계를 촉진합니다.

websockets 라이브러리: 이는 WebSocket 서버 및 클라이언트를 구축하기 위한 깨끗하고 강력한 API를 제공하는 환상적인 Python 라이브러리입니다. 핸드셰이크, 프레이밍 및 오류 처리를 포함한 저수준 WebSocket 프로토콜 세부 정보를 처리하여 개발자가 애플리케이션 로직에 집중할 수 있도록 합니다. 비동기 특성은 ASGI 및 최신 Python의 asyncio 패러다임과 완벽하게 일치합니다.

이들을 함께 사용하는 원리는 ASGI 서버를 진입점으로 사용하여 WebSocket 요청을 websockets 애플리케이션으로 디스패치하는 것입니다. websockets 라이브러리 자체도 독립형 서버로 작동할 수 있지만, Uvicorn과 같은 ASGI 서버와 통합하면 특히 다른 ASGI 호환 프로토콜을 처리하거나 ASGI 미들웨어와 통합해야 할 때 더 큰 유연성을 얻을 수 있습니다.

고성능 WebSocket 서버 구축

우리의 목표는 수많은 동시 WebSocket 연결을 효율적으로 처리할 수 있는 서버를 만드는 것입니다. 여기에는 비동기 프로그래밍과 신중한 리소스 관리가 포함됩니다.

간단한 에코 서버

기본 에코 서버부터 시작해 보겠습니다. 클라이언트가 메시지를 보내면 서버는 단순히 메시지를 다시 보냅니다. 이는 핵심 송수신 기능을 보여줍니다.

# echo_server.py
import asyncio
import websockets

async def echo(websocket, path):
    """
    WebSocket 연결을 위한 비동기 핸들러.
    클라이언트에서 수신된 메시지를 다시 에코합니다.
    """
    print(f"Client connected: {websocket.remote_address}")
    try:
        async for message in websocket:
            print(f"Received message from {websocket.remote_address}: {message}")
            await websocket.send(f"Echo: {message}")
    except websockets.exceptions.ConnectionClosedOK:
        print(f"Client {websocket.remote_address} disconnected gracefully")
    except websockets.exceptions.ConnectionClosedError as e:
        print(f"Client {websocket.remote_address} disconnected with error: {e}")
    finally:
        print(f"Client disconnected: {websocket.remote_address}")

async def main():
    """
    WebSocket 서버를 시작합니다.
    """
    # localhost, 포트 8765에서 WebSocket 서버 시작
    async with websockets.serve(echo, "localhost", 8765):
        await asyncio.Future()  # 무기한 실행

if __name__ == "__main__":
    print("Starting WebSocket echo server on ws://localhost:8765")
    asyncio.run(main())

이것을 실행하려면 echo_server.py로 저장하고 python echo_server.py를 실행하면 됩니다. 그런 다음 브라우저 콘솔의 간단한 JavaScript 클라이언트를 사용하여 테스트할 수 있습니다.

const ws = new WebSocket("ws://localhost:8765");
ws.onopen = () => console.log("Connected");
ws.onmessage = (event) => console.log("Received:", event.data);
ws.send("Hello, WebSocket!");

향상된 유연성을 위한 ASGI 통합

websockets.serve 함수는 독립형 WebSocket 애플리케이션에 훌륭하지만, Uvicorn과 같은 ASGI 서버와 통합하면 다음과 같은 이점을 얻을 수 있습니다.

  • 동일한 서버에서 WebSocket과 함께 다른 ASGI 애플리케이션(예: REST API) 실행
  • ASGI 미들웨어 활용
  • 프로덕션 준비된 ASGI 서버가 제공하는 더 나은 프로세스 관리 및 확장 기능

websockets 핸들러를 ASGI 애플리케이션으로 래핑하는 방법은 다음과 같습니다. websockets 라이브러리는 이를 단순화하기 위해 websockets.ASGIHandler를 제공합니다.

# asgi_websocket_server.py
import asyncio
import websockets
from websockets.exceptions import ConnectionClosedOK, ConnectionClosedError
from websockets.sync.server import serve
import uvicorn

async def websocket_application(scope, receive, send):
    """
    우리의 에코 서버를 위한 ASGI 호환 WebSocket 애플리케이션.
    """
    if scope['type'] == 'websocket':
        async def handler(websocket):
            print(f"ASGI Client connected: {websocket.remote_address}")
            try:
                async for message in websocket:
                    print(f"ASGI Received message from {websocket.remote_address}: {message}")
                    await websocket.send(f"ASGI Echo: {message}")
            except ConnectionClosedOK:
                print(f"ASGI Client {websocket.remote_address} disconnected gracefully")
            except ConnectionClosedError as e:
                print(f"ASGI Client {websocket.remote_address} disconnected with error: {e}")
            finally:
                print(f"ASGI Client disconnected: {websocket.remote_address}")

        # ASGI용 websockets.server.serve 프로토콜 사용
        # 이것은 연결을 위한 AsyncWebSocketServerProtocol 인스턴스를 생성합니다.
        await websockets.server.serve_websocket(handler, scope, receive, send)
    else:
        # 필요한 경우 다른 유형의 요청 처리(예: 상태 확인을 위한 HTTP)
        # 순수 WebSocket 서버의 경우 이것은 오류일 수 있습니다.
        response_start = {'type': 'http.response.start', 'status': 404, 'headers': []}
        response_body = {'type': 'http.response.body', 'body': b'Not Found'}
        await send(response_start)
        await send(response_body)

# Uvicorn으로 실행하려면:
# uvicorn asgi_websocket_server:websocket_application --port 8000 --ws websockets

이 ASGI 호환 서버를 실행하려면:

  1. Uvicorn 설치: pip install uvicorn websockets
  2. 명령 실행: uvicorn asgi_websocket_server:websocket_application --port 8000 --ws websockets

--ws websockets 플래그는 Uvicorn에 websockets를 사용하여 WebSocket 연결을 처리하도록 지시하여 우리 애플리케이션과의 호환성을 보장합니다. 이제 JavaScript 클라이언트는 ws://localhost:8000을 가리켜야 합니다.

실제 예: 간단한 채팅방

여러 클라이언트를 처리하고 메시지를 브로드캐스트하는 기능을 시연하기 위해 에코 서버를 기본 채팅방으로 확장해 보겠습니다.

# chat_server.py
import asyncio
import websockets
from websockets.exceptions import ConnectionClosedOK, ConnectionClosedError
import json

CONNECTED_CLIENTS = set() # 활성 WebSocket 연결 저장

async def register(websocket):
    """새 클라이언트 연결을 등록합니다."""
    CONNECTED_CLIENTS.add(websocket)
    print(f"Client connected: {websocket.remote_address}. Total clients: {len(CONNECTED_CLIENTS)}")

async def unregister(websocket):
    """클라이언트 연결을 등록 해제합니다."""
    CONNECTED_CLIENTS.remove(websocket)
    print(f"Client disconnected: {websocket.remote_address}. Total clients: {len(CONNECTED_CLIENTS)}")

async def broadcast_message(message):
    """모든 연결된 클라이언트에게 메시지를 보냅니다."""
    if CONNECTED_CLIENTS:  # 보낼 클라이언트가 있는지 확인
        await asyncio.wait([client.send(message) for client in CONNECTED_CLIENTS])

async def chat_handler(websocket, path):
    """
    채팅방에서 개별 클라이언트 연결을 처리합니다.
    """
    await register(websocket)
    try:
        user_name = None
        async for message_str in websocket:
            try:
                message_data = json.loads(message_str)
                message_type = message_data.get("type")

                if message_type == "join":
                    user_name = message_data.get("name", "Anonymous")
                    join_msg = json.dumps({"type": "status", "message": f"{user_name} joined the chat."})
                    await broadcast_message(join_msg)
                elif message_type == "chat" and user_name:
                    chat_msg = message_data.get("message", "")
                    full_msg = json.dumps({"type": "chat", "sender": user_name, "message": chat_msg})
                    await broadcast_message(full_msg)
                else:
                    await websocket.send(json.dumps({"type": "error", "message": "Invalid message format or not joined." }))
            except json.JSONDecodeError:
                print(f"Received invalid JSON from {websocket.remote_address}: {message_str}")
                await websocket.send(json.dumps({"type": "error", "message": "Invalid JSON format." }))
            except Exception as e:
                print(f"Error handling message from {websocket.remote_address}: {e}")
                await websocket.send(json.dumps({"type": "error", "message": f"Server error: {e}" }))
    except ConnectionClosedOK:
        print(f"Client {websocket.remote_address} disconnected gracefully")
    except ConnectionClosedError as e:
        print(f"Client {websocket.remote_address} disconnected with error: {e}")
    finally:
        if user_name:
            leave_msg = json.dumps({"type": "status", "message": f"{user_name} left the chat."})
            await broadcast_message(leave_msg)
        await unregister(websocket)

async def main_chat_server():
    """채팅 WebSocket 서버를 시작합니다."""
    async with websockets.serve(chat_handler, "localhost", 8766):
        await asyncio.Future() # 무기한 실행

if __name__ == "__main__":
    print("Starting WebSocket chat server on ws://localhost:8766")
    asyncio.run(main_chat_server())

이 채팅 서버는 사용자가 이름으로 참가하고 모든 사람에게 브로드캐스트되는 메시지를 보낼 수 있도록 합니다. 활성 연결을 추적하기 위한 전역 집합 CONNECTED_CLIENTS와 효율적인 브로드캐스트를 위해 asyncio.wait를 사용합니다.

애플리케이션 시나리오

websockets와 ASGI를 사용하여 구축된 독립형 고성능 WebSocket 서버는 다음과 같은 경우에 이상적입니다.

  • 실시간 대시보드: 라이브 데이터 업데이트(주가, 센서 판독값, 분석) 표시.
  • 멀티플레이어 게임: 게임 상태 동기화를 위한 낮은 지연 시간 통신.
  • 라이브 채팅 및 메시징: 프레임워크 오버헤드 없이 사용자 정의 채팅 애플리케이션 구축.
  • IoT 장치 통신: 연결된 장치에서 실시간 데이터 스트림 수신.
  • 알림 시스템: 클라이언트에 즉시 알림 푸시.

결론: Python 실시간 애플리케이션 강화

websockets 라이브러리의 견고함과 ASGI의 유연성 및 표준화를 결합함으로써 Python 개발자는 까다로운 실시간 통신 요구 사항을 처리할 수 있는 강력한 독립형 WebSocket 서버를 만들 수 있습니다. 이 접근 방식은 세밀한 제어, 뛰어난 성능 및 확장 경로를 제공하여 현대적이고 대화형 웹 서비스에 매우 유용한 패턴이 됩니다. 이러한 도구를 활용하면 Python이 실시간 애플리케이션 영역에서 강력한 경쟁자로 자리매김할 수 있습니다.

Share this article

HackerNews

More Posts from Leapcell

Popular Posts