Django Channels를 이용한 WebSocket 실시간 통신 연동 기초

Django의 HTTP 요청-응답 모델을 넘어 WebSocket을 통한 실시간 양방향 통신을 구현하는 Django Channels의 기본 개념과 구현 방법

WebSocket과 Django Channels가 필요한 이유

HTTP 통신의 한계

• 전통적인 Django는 WSGI 기반으로 요청-응답 사이클 후 연결 종료

• 실시간 채팅, 알림, 라이브 대시보드 등에는 부적합

WebSocket의 장점

• 단일 TCP 연결 위에서 양방향 통신

• 연결 지향적이며 실시간성 보장

• 최초 핸드셰이크 이후 최소한의 데이터만 전송

Django Channels

• Django가 WebSocket 등 HTTP 이외 프로토콜 처리 가능하게 확장

• ASGI(Asynchronous Server Gateway Interface) 표준 사용

• HTTP 요청은 Django View로, WebSocket 연결은 Consumer로 분기

프로젝트 설정

라이브러리 설치

pip install channels daphne channels-redis

settings.py 설정

INSTALLED_APPS = [
    "daphne",  # 가장 위에 추가
    "channels",
    # 기타 앱들
]

ASGI_APPLICATION = "myproject.asgi.application"

CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {
            "hosts": [("127.0.0.1", 6379)],
        },
    },
}

asgi.py 설정

from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
import chat.routing

application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    "websocket": AuthMiddlewareStack(
        URLRouter(chat.routing.websocket_urlpatterns)
    ),
})

Consumer와 Routing 구현

Routing 설정

# chat/routing.py
from django.urls import re_path
from . import consumers

websocket_urlpatterns = [
    re_path(r"ws/chat/(?P<room_name>\w+)/$", consumers.ChatConsumer.as_asgi()),
]

Consumer 작성

# chat/consumers.py
class ChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.room_name = self.scope["url_route"]["kwargs"]["room_name"]
        self.room_group_name = f"chat_{self.room_name}"
        
        await self.channel_layer.group_add(
            self.room_group_name, self.channel_name
        )
        await self.accept()

    async def disconnect(self, close_code):
        await self.channel_layer.group_discard(
            self.room_group_name, self.channel_name
        )

    async def receive(self, text_data):
        data = json.loads(text_data)
        message = data["message"]
        
        await self.channel_layer.group_send(
            self.room_group_name,
            {
                "type": "chat_message",
                "message": message,
                "user": self.scope["user"].username
            }
        )

    async def chat_message(self, event):
        await self.send(text_data=json.dumps({
            "message": event["message"],
            "user": event["user"]
        }))

실무 고려사항

배포 아키텍처

• Nginx가 HTTP는 Gunicorn으로, WebSocket은 Daphne로 라우팅

• Redis를 채널 레이어 백엔드로 사용

보안

• AuthMiddlewareStack으로 인증 연동

• 운영 환경에서는 WSS(SSL) 사용 필수

• Origin 헤더 검증으로 Cross-Origin 요청 방지

성능 및 안정성

• Consumer 로직은 가볍게 유지

• 무거운 작업은 Celery 등 별도 태스크 큐 활용

• 클라이언트에서 재연결 로직 구현

• ORM 사용 시 database_sync_to_async 래퍼 필요