Django Channels 2026: WebSockets, Real-Time และคำถามสัมภาษณ์

Django Channels 4.x ขยายความสามารถของ Django เกินกว่า HTTP เพื่อรองรับ WebSockets ทำให้สามารถสร้างฟีเจอร์ real-time เช่น chat, การแจ้งเตือน และ live dashboard บทเรียนนี้ครอบคลุมรูปแบบ consumer, channel layers กับ Redis และคำถามสัมภาษณ์ที่แยกแยะผู้สมัครระดับ mid-level จาก senior

ทำไม Django ต้องการ Channels สำหรับ WebSockets

สถาปัตยกรรมของ Django สร้างบน WSGI โปรโตคอล request-response แบบ synchronous WSGI ไม่มีแนวคิดเรื่อง persistent connections Django 4.1 เพิ่ม async views แต่ async views ยังคงทำงานตามรูปแบบ request-response ปิดการเชื่อมต่อหลังจากทุก response

การรองรับ WebSocket ต้องการ ASGI ซึ่งจัดการการเชื่อมต่อระยะยาวและการสื่อสารแบบสองทิศทาง Django Channels 4.x ให้เลเยอร์ ASGI นี้ ห่อหุ้มการรองรับ async native ของ Django ด้วย routing และ consumer abstractions

# 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', 'project.settings')

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

ProtocolTypeRouter กระจาย connections ที่เข้ามาตามประเภทโปรโตคอล HTTP requests ไปยัง handler ASGI มาตรฐานของ Django ขณะที่ WebSocket connections ถูก route ผ่าน URLRouter ไปยัง consumer ที่เหมาะสม

AsyncWebsocketConsumer: Building Block หลัก

Consumers คือส่วนที่เทียบเท่ากับ Django views สำหรับ WebSocket Class AsyncWebsocketConsumer ให้ hooks สามตัว: connect, disconnect และ receive แต่ละตัวทำงานเป็น coroutine ทำให้สามารถทำ I/O operations แบบ non-blocking ได้

# consumers.py
import json
from channels.generic.websocket import AsyncWebsocketConsumer

class NotificationConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        # Extract user from session via AuthMiddlewareStack
        self.user = self.scope["user"]
        if self.user.is_anonymous:
            await self.close()
            return
        
        # Create user-specific channel group
        self.group_name = f"notifications_{self.user.id}"
        await self.channel_layer.group_add(
            self.group_name,
            self.channel_name
        )
        await self.accept()

    async def disconnect(self, close_code):
        # Clean up group membership
        await self.channel_layer.group_discard(
            self.group_name,
            self.channel_name
        )

    async def receive(self, text_data):
        # Handle incoming messages from client
        data = json.loads(text_data)
        await self.send(text_data=json.dumps({
            "type": "ack",
            "id": data.get("id")
        }))

    async def notification_message(self, event):
        # Handler for messages sent via channel layer
        await self.send(text_data=json.dumps({
            "type": "notification",
            "payload": event["payload"]
        }))

Dictionary scope เก็บ metadata ของการเชื่อมต่อ รวมถึง user ที่ authenticated เมื่อใช้ AuthMiddlewareStack channel_name คือ identifier เฉพาะสำหรับการเชื่อมต่อนี้ ขณะที่ groups ทำให้สามารถ broadcast ไปยังหลาย connections ได้

การ Route WebSocket Connections

Routing แมป URL paths ไปยัง consumers คล้ายกับการตั้งค่า URL ของ Django Method as_asgi() คืนค่า instance ของ ASGI application สำหรับแต่ละ consumer class

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

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

URL parameters ที่จับได้จาก regex groups จะปรากฏใน self.scope["url_route"]["kwargs"] ทำให้สามารถ route แบบ dynamic ได้ เช่น การเข้าร่วมห้อง chat ต่างๆ ตาม URL path

Redis Channel Layer สำหรับ Production

In-memory channel layer ทำงานได้ในการพัฒนาแต่ล้มเหลวใน production แต่ละ process รักษา layer ของตัวเอง ป้องกันการสื่อสารข้าม process Package channels-redis ให้โซลูชันระดับ production

# settings.py
CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {
            "hosts": [("redis", 6379)],
            "capacity": 1500,
            "expiry": 10,
        },
    },
}

การตั้งค่า capacity จำกัด message queue ต่อ channel (ค่าเริ่มต้น: 100) การตั้งค่า expiry ควบคุมว่าข้อความจะรอนานเท่าไรก่อนถูกทิ้ง (ค่าเริ่มต้น: 60 วินาที) สำหรับแอปพลิเคชัน high-throughput ให้เพิ่ม capacity และลด expiry เพื่อป้องกันการสะสมของหน่วยความจำ

การส่งข้อความจาก Django Views

Channel layers ทำให้สามารถส่งข้อความ WebSocket จากที่ไหนก็ได้ในแอปพลิเคชัน รวมถึง Django views แบบ synchronous และ Celery tasks

# views.py
from channels.layers import get_channel_layer
from asgiref.sync import async_to_sync

def create_order(request):
    order = Order.objects.create(user=request.user, **form.cleaned_data)
    
    # Send notification to user's WebSocket connection
    channel_layer = get_channel_layer()
    async_to_sync(channel_layer.group_send)(
        f"notifications_{request.user.id}",
        {
            "type": "notification_message",
            "payload": {
                "title": "Order Created",
                "order_id": order.id
            }
        }
    )
    return redirect("order_detail", pk=order.id)

ฟิลด์ type แมปไปยัง handler method บน consumer notification_message กลายเป็น notification_message() หลังจากแทนที่จุดด้วย underscore

การเข้าถึงฐานข้อมูลใน Async Consumers

ORM ของ Django เป็น synchronous โดยค่าเริ่มต้น การเรียก ORM methods แบบ synchronous จาก async consumer จะบล็อก event loop ทำให้ประสิทธิภาพลดลง มีสองโซลูชัน: database_sync_to_async และ async ORM methods ของ Django

# consumers.py
from channels.db import database_sync_to_async
from .models import Message

class ChatConsumer(AsyncWebsocketConsumer):
    @database_sync_to_async
    def save_message(self, content):
        return Message.objects.create(
            room=self.room,
            user=self.user,
            content=content
        )
    
    async def receive(self, text_data):
        data = json.loads(text_data)
        message = await self.save_message(data["content"])
        
        await self.channel_layer.group_send(
            self.room_group_name,
            {
                "type": "chat_message",
                "content": data["content"],
                "user_id": self.user.id,
                "message_id": message.id
            }
        )

Django 4.1+ ให้ async ORM methods ที่มี prefix a: aget(), acreate(), afilter() สิ่งนี้ขจัดความจำเป็นของ database_sync_to_async ในหลายกรณี

# Using Django's native async ORM (Django 4.1+)
async def receive(self, text_data):
    data = json.loads(text_data)
    message = await Message.objects.acreate(
        room=self.room,
        user=self.user,
        content=data["content"]
    )

การ Deploy ด้วย Uvicorn และ Nginx

Daphne คือ ASGI server ดั้งเดิมสำหรับ Channels Uvicorn ร่วมกับ uvloop ให้ประสิทธิภาพที่ดีกว่าสำหรับ WebSocket workloads และมีชุมชนที่ใหญ่กว่า

# Production deployment
uvicorn project.asgi:application \
    --host 0.0.0.0 \
    --port 8000 \
    --workers 4 \
    --ws websockets \
    --loop uvloop

Nginx ต้องการการตั้งค่าเฉพาะสำหรับ WebSocket proxying Headers Upgrade และ Connection ต้องถูกส่งต่อเพื่อเปิดใช้งานการเปลี่ยนโปรโตคอลจาก HTTP เป็น WebSocket

# nginx.conf
upstream channels {
    server 127.0.0.1:8000;
}

server {
    listen 80;
    server_name example.com;

    location /ws/ {
        proxy_pass http://channels;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_read_timeout 86400;
    }

    location / {
        proxy_pass http://channels;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
    }
}

การตั้งค่า proxy_read_timeout ป้องกัน Nginx จากการปิด WebSocket connections ที่ไม่ได้ใช้งาน ตั้งค่าให้ตรงกับระยะเวลาการเชื่อมต่อที่คาดหวังของแอปพลิเคชัน

คำถามสัมภาษณ์: Django Channels

คำถามเหล่านี้ปรากฏบ่อยในการสัมภาษณ์ senior Python developer คำตอบที่คาดหวังต้องเกินกว่านิยามพื้นฐาน

ทำไม Django ไม่สามารถจัดการ WebSockets แบบ native ได้?

Django ใช้ WSGI โปรโตคอล synchronous ที่ทุก request ได้รับ response และการเชื่อมต่อถูกปิด WebSockets ต้องการการเชื่อมต่อที่ยาวนานและสองทิศทาง WSGI ไม่มี specification สำหรับสิ่งนี้ Async views ของ Django (เพิ่มใน 4.1) ยังคงทำงานตาม semantics ของ request-response Channels เพิ่มการรองรับ ASGI ซึ่งจัดการการเชื่อมต่อระยะยาวและการสื่อสารแบบ event-driven

ความแตกต่างระหว่าง channel และ channel layer คืออะไร?

Channel คือ named queue ที่ข้อความรอ consumer ทุก WebSocket connection ได้รับ channel name ที่ไม่ซ้ำกัน Channel layer คือ transport backend (Redis, in-memory) ที่ route ข้อความระหว่าง channels Layer จัดการการสื่อสารข้าม process, group membership และการ serialize ข้อความ

เมื่อไหร่ควรใช้ database_sync_to_async เทียบกับ async ORM ของ Django?

ใช้ async ORM methods ของ Django (aget, acreate, afilter) สำหรับการดำเนินการง่ายๆ เพราะรวมเข้ากับ async code ได้ดี ใช้ database_sync_to_async เมื่อเรียก synchronous code ที่ไม่สามารถแปลงได้ง่าย เช่น third-party libraries, QuerySet chains ที่ซับซ้อน หรือ methods ที่ trigger synchronous operations เพิ่มเติมเช่น signals

วิธี scale WebSocket connections แบบ horizontal?

Redis channel layer เปิดใช้งาน horizontal scaling ทุก Uvicorn workers และทุก server instances สื่อสารผ่าน Redis Channel name ของแต่ละ connection ไม่ซ้ำกัน และ Redis ติดตาม group membership Sticky sessions ไม่จำเป็นเพราะ channel layer ไม่ใช่ server ที่รักษา connection state Consumer ต้องการเพียง channel name เพื่อส่งข้อความ

จะเกิดอะไรขึ้นถ้า Redis ล่ม?

Connections ใหม่สำเร็จเพราะ connect() ทำงานก่อน group membership Connections ที่มีอยู่ยังคงเปิดอยู่แต่สูญเสีย group messaging group_send throw exception หรือล้มเหลวแบบเงียบๆ ขึ้นอยู่กับการตั้งค่า สำหรับแอปพลิเคชันที่สำคัญ ให้ implement การตรวจสอบสุขภาพของการเชื่อมต่อและการเชื่อมต่อใหม่อย่างราบรื่นที่ฝั่ง client พิจารณา Redis Sentinel หรือ Redis Cluster สำหรับ high availability

สำหรับการเตรียมสัมภาษณ์ Django เพิ่มเติม ดู module Django middleware และ module Django signals บน SharpSkill

Checklist สำหรับ Production Django Channels

• ใช้ channels-redis กับ Redis 6+ สำหรับ channel layer In-memory layer ใช้สำหรับ development เท่านั้น
• ตั้งค่า capacity ตาม message throughput ที่คาดหวัง 100 (ค่าเริ่มต้น) ทำงานได้สำหรับแอปพลิเคชันส่วนใหญ่
• รัน Uvicorn ด้วย workers หลายตัว: --workers 4 สำหรับ server 4-core
• ตั้งค่า Nginx ด้วย proxy_read_timeout ที่ตรงกับการเชื่อมต่อที่ยาวที่สุดที่คาดหวัง
• Implement ลอจิกการเชื่อมต่อใหม่ที่ฝั่ง client ด้วย exponential backoff
• ตรวจสอบการใช้หน่วยความจำ Redis ปริมาณข้อความสูงกับ consumers ที่ช้าทำให้หน่วยความจำเพิ่มขึ้น
• ใช้ async ORM methods ของ Django เมื่อเป็นไปได้เพื่อหลีกเลี่ยง overhead ของ thread pool
• ทดสอบด้วยจำนวน connections ที่เป็นจริง WebSocket connections ใช้ทรัพยากรมากกว่า HTTP requests