Django 실시간 채팅 구현기 — Channels와 WebSocket으로 1:1 채팅을 만들었다 (2026)

매칭 기반 서비스를 만들면서 1:1 실시간 채팅이 필요했습니다. 매칭이 성사되면 두 사람이 바로 대화할 수 있어야 하는데, 일반적인 HTTP 요청-응답 방식으로는 “실시간”이 불가능합니다. 메시지를 보내면 상대방이 새로고침을 해야 보이니까요.

그래서 Django Channels + WebSocket으로 Django 실시간 채팅을 구현했습니다. 메시지를 보내는 순간 상대방 화면에 바로 뜨는, 카카오톡과 같은 경험을 만드는 게 목표였습니다. 이 글에서는 Django 실시간 채팅을 처음부터 구현한 과정과 구조를 공유합니다.
왜 Django로 채팅을 만들었나
“채팅은 Node.js로 만들어야 하는 거 아니야?” 라는 말을 많이 들었습니다. 맞는 말이긴 합니다. Node.js는 비동기 처리에 강하고, Socket.io라는 검증된 라이브러리가 있습니다.
그런데 이 서비스의 백엔드가 이미 Django로 구축되어 있었습니다. 회원가입, 로그인, 매칭 시스템, 포인트 시스템, 프로필 관리 — 전부 Django의 인증 시스템과 ORM 위에서 돌아가고 있었습니다. 채팅 하나 때문에 Node.js를 별도로 띄우면 인증 동기화, DB 이중 관리, 서버 비용 증가 문제가 생깁니다.
Django Channels를 쓰면 기존 Django 프로젝트에 WebSocket을 추가할 수 있습니다. 인증도 Django의 세션/토큰을 그대로 쓸 수 있고, DB도 Django ORM으로 통일됩니다. 채팅 때문에 별도 서버를 띄울 필요가 없었습니다.
Django 실시간 채팅의 전체 구조
Django 실시간 채팅의 아키텍처는 이렇게 생겼습니다.
[사용자 A 브라우저] ←WebSocket→ [Daphne (ASGI 서버)]
↕
[Django Channels Consumer]
↕
[Redis (Channel Layer)]
↕
[Django Channels Consumer]
↕
[사용자 B 브라우저] ←WebSocket→ [Daphne (ASGI 서버)]
일반적인 Django는 WSGI 서버(Gunicorn 등) 위에서 HTTP 요청만 처리합니다. Django Channels는 ASGI 서버(Daphne)를 사용해서 HTTP + WebSocket을 동시에 처리합니다.
핵심 구성 요소를 정리하면:
| 구성 요소 | 역할 |
|---|---|
| Django Channels | Django에 WebSocket 지원을 추가하는 라이브러리 |
| Daphne | ASGI 서버. HTTP + WebSocket 동시 처리 |
| Consumer | WebSocket 연결을 처리하는 핸들러 (View의 WebSocket 버전) |
| Redis | Channel Layer. 서로 다른 Consumer 간 메시지 중계 |
| Group | 채팅방 개념. 같은 Group에 속한 Consumer끼리 메시지 공유 |
Django 실시간 채팅 구현 — 핵심 코드
1단계: 패키지 설치
pip install channels channels-redis daphne
2단계: ASGI 설정
Django의 기본 WSGI 설정을 ASGI로 전환합니다.
# settings.py
INSTALLED_APPS = [
'daphne', # 반드시 맨 위에
'channels',
'django.contrib.auth',
# ... 기존 앱들
'chat',
]
ASGI_APPLICATION = 'myproject.asgi.application'
# Redis를 Channel Layer로 사용
CHANNEL_LAYERS = {
'default': {
'BACKEND': 'channels_redis.core.RedisChannelLayer',
'CONFIG': {
'hosts': [('127.0.0.1', 6379)],
},
},
}
# asgi.py
import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
import chat.routing
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
application = ProtocolTypeRouter({
'http': get_asgi_application(),
'websocket': AuthMiddlewareStack(
URLRouter(chat.routing.websocket_urlpatterns)
),
})
AuthMiddlewareStack이 핵심입니다. WebSocket 연결에도 Django의 인증 정보가 자동으로 넘어옵니다. 로그인한 사용자만 채팅을 할 수 있게 제어할 수 있습니다.
3단계: Consumer 작성 — WebSocket 핸들러
Consumer는 View의 WebSocket 버전입니다. 연결, 메시지 수신, 연결 해제를 처리합니다.
# chat/consumers.py
import json
from channels.generic.websocket import AsyncWebsocketConsumer
from channels.db import database_sync_to_async
from .models import Message, ChatRoom
class ChatConsumer(AsyncWebsocketConsumer):
async def connect(self):
self.room_id = self.scope['url_route']['kwargs']['room_id']
self.room_group_name = f'chat_{self.room_id}'
self.user = self.scope['user']
# 인증 확인 — 로그인 안 한 사용자는 연결 거부
if not self.user.is_authenticated:
await self.close()
return
# 채팅방 그룹에 참가
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']
# DB에 메시지 저장
await self.save_message(message)
# 같은 채팅방의 모든 사용자에게 메시지 전송
await self.channel_layer.group_send(
self.room_group_name,
{
'type': 'chat_message',
'message': message,
'sender': self.user.username,
'timestamp': str(datetime.now()),
}
)
async def chat_message(self, event):
# 클라이언트에게 메시지 전달
await self.send(text_data=json.dumps({
'message': event['message'],
'sender': event['sender'],
'timestamp': event['timestamp'],
}))
@database_sync_to_async
def save_message(self, message):
room = ChatRoom.objects.get(id=self.room_id)
Message.objects.create(
room=room,
sender=self.user,
content=message
)
이 코드가 하는 일을 순서대로 설명하면:
- connect — WebSocket 연결 시 인증 확인 후 채팅방 그룹에 참가
- receive — 메시지를 받으면 DB에 저장하고 같은 방의 모든 사용자에게 전송
- chat_message — 그룹에서 받은 메시지를 클라이언트 WebSocket으로 전달
- disconnect — 연결 해제 시 그룹에서 나가기
4단계: URL 라우팅
# chat/routing.py
from django.urls import re_path
from . import consumers
websocket_urlpatterns = [
re_path(r'ws/chat/(?P<room_id>\d+)/$', consumers.ChatConsumer.as_asgi()),
]
5단계: 프론트엔드 WebSocket 연결
// JavaScript — 채팅 화면
const roomId = '{{ room.id }}';
const ws = new WebSocket(
'wss://' + window.location.host + '/ws/chat/' + roomId + '/'
);
// 메시지 수신
ws.onmessage = function(e) {
const data = JSON.parse(e.data);
appendMessage(data.sender, data.message, data.timestamp);
};
// 메시지 전송
function sendMessage() {
const input = document.getElementById('message-input');
ws.send(JSON.stringify({
'message': input.value
}));
input.value = '';
}
// 연결 끊김 시 재연결
ws.onclose = function(e) {
console.log('WebSocket 연결이 끊어졌습니다. 재연결 시도...');
setTimeout(function() {
location.reload();
}, 3000);
};
Django 실시간 채팅에서 가장 중요한 것 — 인증 연동
채팅에서 보안은 절대 타협할 수 없는 부분입니다. “로그인한 사용자만 채팅 가능”하고, “매칭된 상대와만 대화 가능”해야 합니다.
Django Channels의 AuthMiddlewareStack을 사용하면 WebSocket 연결에도 Django 세션이 자동으로 연결됩니다. Consumer 안에서 self.scope['user']로 현재 로그인한 사용자를 바로 가져올 수 있습니다.
# Consumer 안에서 인증 + 권한 확인
async def connect(self):
self.user = self.scope['user']
# 1. 로그인 확인
if not self.user.is_authenticated:
await self.close()
return
# 2. 이 채팅방에 참여할 권한이 있는지 확인
room = await self.get_room()
if self.user not in [room.user1, room.user2]:
await self.close()
return
await self.accept()
이걸 Node.js로 구현하려면 Django와 별도의 인증 토큰 교환 로직을 만들어야 합니다. Django Channels는 Django의 인증을 그대로 쓰니까 이런 복잡함이 없습니다. 이게 Django 실시간 채팅의 가장 큰 장점입니다.
Django 실시간 채팅 — 메시지 모델 설계
채팅 메시지를 DB에 저장해야 나중에 대화 기록을 불러올 수 있습니다.
# chat/models.py
from django.db import models
from django.contrib.auth.models import User
class ChatRoom(models.Model):
user1 = models.ForeignKey(User, on_delete=models.CASCADE, related_name='rooms_as_user1')
user2 = models.ForeignKey(User, on_delete=models.CASCADE, related_name='rooms_as_user2')
created_at = models.DateTimeField(auto_now_add=True)
class Meta:
unique_together = ('user1', 'user2') # 같은 조합 중복 방지
class Message(models.Model):
room = models.ForeignKey(ChatRoom, on_delete=models.CASCADE, related_name='messages')
sender = models.ForeignKey(User, on_delete=models.CASCADE)
content = models.TextField()
timestamp = models.DateTimeField(auto_now_add=True)
is_read = models.BooleanField(default=False)
class Meta:
ordering = ['timestamp'] # 시간순 정렬
is_read 필드로 읽음/안읽음 표시도 구현할 수 있습니다. 매칭이 성사되면 ChatRoom을 생성하고, 메시지를 보낼 때마다 Message를 저장하는 구조입니다.
Django 실시간 채팅 — 운영에서 겪은 이슈들
Redis 필수
개발 환경에서는 InMemoryChannelLayer로 테스트할 수 있지만, 프로덕션에서는 반드시 Redis를 써야 합니다. InMemory는 서버가 1대일 때만 작동하고, 프로세스가 여러 개면 메시지가 전달되지 않습니다.
# Redis 설치 (Ubuntu)
sudo apt install redis-server
sudo systemctl enable redis-server
WebSocket 연결 끊김 처리
모바일에서는 네트워크가 자주 끊깁니다. Wi-Fi에서 LTE로 전환되거나, 지하철에서 터널을 지나가거나. 연결이 끊어졌을 때 자동으로 재연결하는 로직이 없으면 사용자가 “채팅이 안 돼요”라고 문의합니다.
Daphne vs Uvicorn
ASGI 서버로 Daphne 대신 Uvicorn을 쓸 수도 있습니다. Uvicorn이 성능은 더 좋지만, Django Channels 공식 문서는 Daphne을 기준으로 작성되어 있어서 처음에는 Daphne으로 시작하는 게 편합니다.
마무리 — Django 실시간 채팅, 생각보다 어렵지 않다
Django 실시간 채팅을 처음 구현할 때 “Django로 WebSocket이 된다고?” 싶었습니다. Django Channels를 알기 전에는 채팅 기능 하나 때문에 Node.js 서버를 별도로 띄워야 하나 고민했습니다.
막상 해보니 Django Channels가 Django의 장점을 그대로 살리면서 WebSocket을 추가해주는 거라, 기존 Django 프로젝트에 자연스럽게 붙일 수 있었습니다. 특히 인증 연동이 매끄러운 게 큰 장점이었습니다.
1:1 채팅, 그룹 채팅, 실시간 알림 — WebSocket이 필요한 기능이라면 Django Channels를 고려해보세요. Django를 이미 쓰고 있다면, 별도 서버 없이 채팅을 붙일 수 있습니다.
관련 글도 참고하세요.