"Django Channels 是什麼？跟 Django 原生的 ASGI 支援有什麼不同？"

"Django Channels 是 Django 的官方擴充套件，由 Django Migrations 的作者 Andrew Godwin 開發，讓 Django 能處理 WebSocket、長連接等非同步協定。雖然 Django 3.0+ 原生支援 ASGI，但僅提供基礎的非同步 HTTP 處理；Django Channels 在此基礎上加入了 Consumer、Channel Layers、Groups 等完整的 WebSocket 處理架構，是 Django 實作即時通訊的標準方案。"

"WebSocket 和 HTTP 有什麼差別？為什麼需要 WebSocket？"

"HTTP 是單向的請求/回應模式，客戶端發送請求後伺服器才能回應，無法主動推送資料。WebSocket 則是全雙工（Full-Duplex）的持久連線協定，連線建立後客戶端與伺服器可以隨時互相傳送資料，非常適合聊天室、即時通知、即時儀表板等需要即時雙向通訊的場景。"

"Channel Layers 一定要用 Redis 嗎？開發環境可以用什麼替代？"

"生產環境強烈建議使用 Redis 作為 Channel Layer 的後端（channels_redis），因為它支援跨進程、跨伺服器的訊息傳遞。開發環境中可以使用 InMemoryChannelLayer，但它僅支援單一進程，不適合生產使用。如果不需要 Group 廣播功能（例如只做一對一的 WebSocket 回應），甚至可以不設定 Channel Layer。"

Django Channels 與 WebSocket 即時通訊實戰 | Django 教學

Django Channels 是讓 Django 從傳統的 HTTP 請求/回應模式擴展到 WebSocket 即時雙向通訊的官方解決方案。它基於 ASGI（Asynchronous Server Gateway Interface，非同步伺服器閘道介面）標準，提供 Consumer、Channel Layers、Groups 等核心元件，讓你在 Django 的完整生態系內就能實作聊天室、即時通知、協作編輯等即時功能。本文將從 WebSocket 與 HTTP 的差異出發，帶你完成 Django Channels 的安裝設定，並實作一個完整的即時聊天室範例。

WebSocket vs HTTP：為什麼需要即時通訊？

傳統的 HTTP 協定採用「請求/回應」模式——客戶端發送請求，伺服器處理後回傳回應，一次交互就結束了。這種模式對於網頁瀏覽、表單提交等場景完全足夠，但當你需要即時推送資料時就會遇到瓶頸。

想像一個聊天室場景：使用者 A 發送了一則訊息，使用者 B 要怎麼即時收到？在純 HTTP 的世界裡，通常有這幾種方案：

輪詢（Polling）：客戶端每隔幾秒向伺服器發送 GET 請求，詢問「有沒有新訊息」。缺點是浪費頻寬與伺服器資源
長輪詢（Long Polling）：客戶端發送請求後，伺服器「掛住」直到有新資料才回應。比輪詢好一些，但連線管理複雜
Server-Sent Events（SSE）：伺服器可以主動推送資料給客戶端，但只能單向（伺服器 -> 客戶端）

WebSocket 協定徹底解決了這個問題。它在 HTTP 握手後升級為持久的 全雙工（Full-Duplex） 連線，讓客戶端和伺服器可以隨時互相傳送資料：

特性	HTTP	WebSocket
通訊模式	單向（請求/回應）	雙向（全雙工）
連線方式	每次請求建立新連線	一次握手，持久連線
伺服器推送	不支援（需輪詢）	原生支援
協定標頭	每次請求都帶完整 Header	握手後無額外 Header 開銷
適用場景	網頁瀏覽、API 呼叫	聊天室、即時通知、遊戲
URL Scheme	`http://` / `https://`	`ws://` / `wss://`

ASGI vs WSGI：Django 的協定演進

Django 傳統上使用 WSGI（Web Server Gateway Interface，網頁伺服器閘道介面）作為應用伺服器的標準介面。WSGI 是同步的，每個請求佔用一個執行緒，處理完畢後連線就結束，天生不支援長連接。

ASGI（Asynchronous Server Gateway Interface）是 WSGI 的非同步繼任者，專門為支援長連接和非同步處理而設計：

特性	WSGI	ASGI
協定支援	僅 HTTP	HTTP、WebSocket、HTTP/2、SSE
執行模型	同步（一請求一執行緒）	非同步（事件迴圈）
長連接支援	不支援	原生支援
Django 支援	Django 1.0+	Django 3.0+（原生）+ Channels
部署伺服器	Gunicorn、uWSGI	Daphne、Uvicorn、Hypercorn

Django 3.0 開始內建 ASGI 支援，但僅提供基礎的非同步 HTTP 處理。要支援 WebSocket，需要安裝 Django Channels，它在 ASGI 之上建構了完整的 WebSocket 處理架構。

安裝 Django Channels 與相關套件

首先安裝 Django Channels 和 Redis Channel Layer 後端：

# 安裝 Django Channels 與 Redis Channel Layer
pip install channels channels-redis

接著在 settings.py 中註冊 Channels：

# settings.py

INSTALLED_APPS = [
    'daphne',  # ASGI 伺服器（Channels 4.x 推薦）
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    # 第三方套件
    'channels',
    # 你的 App
    'chat',
]

# 指定 ASGI 應用程式的路徑
ASGI_APPLICATION = 'myproject.asgi.application'

daphne 放在 INSTALLED_APPS 的第一個位置，是因為 Channels 4.x 版本使用 Daphne 作為預設的 ASGI 開發伺服器，會自動取代 Django 的 runserver 指令。

設定 ASGI Application

Django 專案建立時會自動產生 asgi.py，我們需要修改它來整合 Channels 的路由系統：

# myproject/asgi.py

import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')

# 必須在 import routing 之前呼叫 django.setup()
django_asgi_app = get_asgi_application()

from chat.routing import websocket_urlpatterns  # noqa: E402

application = ProtocolTypeRouter({
    # HTTP 請求仍走傳統 Django View
    'http': django_asgi_app,
    # WebSocket 請求走 Channels Consumer
    'websocket': AuthMiddlewareStack(
        URLRouter(websocket_urlpatterns)
    ),
})

這裡有幾個關鍵元件：

ProtocolTypeRouter：根據協定類型（HTTP 或 WebSocket）分派到不同的處理程式
URLRouter：類似 Django 的 urls.py，將 WebSocket URL 路徑對應到特定的 Consumer
AuthMiddlewareStack：自動將 Django 的認證資訊注入到 scope['user']，讓 Consumer 能取得登入使用者

Consumer：WebSocket 的 View

Consumer 是 Django Channels 處理 WebSocket 連線的核心元件，你可以把它理解為 WebSocket 版本的 Django View。每個 WebSocket 連線都會建立一個 Consumer 實例，透過三個主要方法處理連線的生命週期：

Django View	Channels Consumer
`def view(request)`	`class ChatConsumer(WebsocketConsumer)`
`request.GET`	`self.scope['query_string']`
`request.user`	`self.scope['user']`
請求/回應後結束	持久連線（connect / disconnect / receive）

Django Channels 提供三種主要的 Consumer 類型：

WebsocketConsumer：同步版本，適合簡單場景，內部使用執行緒處理
AsyncWebsocketConsumer：非同步版本，效能更佳，推薦在大多數場景使用
JsonWebsocketConsumer：自動處理 JSON 的序列化與反序列化

同步 Consumer 範例

# chat/consumers.py

import json
from channels.generic.websocket import WebsocketConsumer


class EchoConsumer(WebsocketConsumer):
    """同步 WebSocket Consumer，收到什麼就回傳什麼。"""

    def connect(self):
        """客戶端發起 WebSocket 連線時觸發。"""
        self.accept()  # 接受連線，若不呼叫則連線被拒絕

    def disconnect(self, close_code):
        """客戶端斷開連線時觸發。"""
        pass

    def receive(self, text_data):
        """收到客戶端傳來的訊息時觸發。"""
        data = json.loads(text_data)
        message = data['message']

        # 將訊息原封不動回傳給客戶端
        self.send(text_data=json.dumps({
            'message': f'Echo: {message}'
        }))

非同步 Consumer 範例

# chat/consumers.py

import json
from channels.generic.websocket import AsyncWebsocketConsumer


class AsyncEchoConsumer(AsyncWebsocketConsumer):
    """非同步 WebSocket Consumer，效能更佳的版本。"""

    async def connect(self):
        await self.accept()

    async def disconnect(self, close_code):
        pass

    async def receive(self, text_data):
        data = json.loads(text_data)
        message = data['message']

        await self.send(text_data=json.dumps({
            'message': f'Echo: {message}'
        }))

非同步版本使用 async/await 語法，在處理大量並發連線時不會阻塞事件迴圈，效能遠優於同步版本。

Routing：WebSocket 的 URLconf

Consumer 寫好後，需要設定路由來告訴 Channels 哪個 URL 要交給哪個 Consumer 處理：

# 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()),
]

注意幾個重點：

WebSocket URL 慣例以 ws/ 作為前綴，方便在反向代理（如 Nginx）中區分 HTTP 與 WebSocket 請求
使用 re_path 搭配正規表達式（Regular Expression）擷取 URL 參數（如聊天室名稱）
Consumer 必須呼叫 .as_asgi() 方法，類似 Django CBV 的 .as_view()

Channel Layers 與 Redis

到目前為止，每個 Consumer 實例都是獨立的，無法互相溝通。如果使用者 A 發送訊息，使用者 B 的 Consumer 不會收到通知。這就是 Channel Layers（頻道層）要解決的問題。

Channel Layer 是 Consumer 之間的通訊層，讓不同的 Consumer 實例（甚至不同的 Worker 進程）能互相傳遞訊息。它的核心概念包含：

Channel：每個 Consumer 實例都有一個唯一的 channel_name，就像信箱地址
Group：一組 Channel 的集合，用於廣播訊息（如聊天室中的所有成員）

設定 Redis Channel Layer

生產環境建議使用 Redis 作為 Channel Layer 的後端：

# settings.py

CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels_redis.core.RedisChannelLayer',
        'CONFIG': {
            'hosts': [('127.0.0.1', 6379)],
            'capacity': 1500,    # 每個 Channel 的訊息緩衝上限
            'expiry': 10,        # 訊息過期時間（秒）
        },
    },
}

開發環境如果不想安裝 Redis，可以使用記憶體內的 Channel Layer（僅限單一進程）：

# settings.py（開發環境）

CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels.layers.InMemoryChannelLayer',
    },
}

Groups：實現群組廣播

Groups（群組）是 Channel Layer 中最常用的功能。它允許你將多個 Consumer 加入同一個群組，然後透過 group_send 向群組內所有成員廣播訊息。

三個核心方法：

group_add(group_name, channel_name)：將 Consumer 加入群組
group_discard(group_name, channel_name)：將 Consumer 從群組移除
group_send(group_name, message)：向群組內所有 Consumer 廣播訊息

group_send 的 message 字典中必須包含 type 鍵，其值對應 Consumer 中的處理方法名稱（點號 . 會自動轉換為底線 _）。

實作完整聊天室

現在讓我們把所有概念串在一起，實作一個完整的即時聊天室。

Consumer：聊天室核心邏輯

# chat/consumers.py

import json
from channels.generic.websocket import AsyncWebsocketConsumer


class ChatConsumer(AsyncWebsocketConsumer):
    """非同步 WebSocket Consumer，實作群組聊天功能。"""

    async def connect(self):
        """客戶端發起 WebSocket 連線時觸發。"""
        # 從 URL 路徑中取得聊天室名稱
        self.room_name = self.scope['url_route']['kwargs']['room_name']
        self.room_group_name = f'chat_{self.room_name}'

        # 將此 Consumer 加入聊天室 Group
        await self.channel_layer.group_add(
            self.room_group_name,
            self.channel_name,  # 此連線的唯一識別
        )

        # 接受 WebSocket 連線
        await self.accept()

    async def disconnect(self, close_code):
        """客戶端斷開連線時觸發。"""
        # 將此 Consumer 從聊天室 Group 移除
        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']
        username = self.scope['user'].username

        # 透過 Channel Layer 廣播訊息給聊天室所有成員
        await self.channel_layer.group_send(
            self.room_group_name,
            {
                'type': 'chat_message',  # 對應下方的 chat_message() 方法
                'message': message,
                'username': username,
            }
        )

    async def chat_message(self, event):
        """處理從 group_send 收到的訊息，傳送給 WebSocket 客戶端。

        方法名稱必須與 group_send 中的 type 值一致（點號轉底線）。
        """
        await self.send(text_data=json.dumps({
            'message': event['message'],
            'username': event['username'],
        }))

Routing：WebSocket URL 配置

# 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()),
]

ASGI 設定

# myproject/asgi.py

import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
django_asgi_app = get_asgi_application()

from chat.routing import websocket_urlpatterns  # noqa: E402

application = ProtocolTypeRouter({
    'http': django_asgi_app,
    'websocket': AuthMiddlewareStack(
        URLRouter(websocket_urlpatterns)
    ),
})

前端 JavaScript 客戶端

// 取得聊天室名稱（假設從 URL 或頁面變數取得）
const roomName = 'general';

// 建立 WebSocket 連線（使用 wss:// 對應 HTTPS，ws:// 對應 HTTP）
const wsScheme = window.location.protocol === 'https:' ? 'wss' : 'ws';
const ws = new WebSocket(
    `${wsScheme}://${window.location.host}/ws/chat/${roomName}/`
);

// 連線建立成功
ws.onopen = function(event) {
    console.log('WebSocket 連線成功');
};

// 收到伺服器推送的訊息
ws.onmessage = function(event) {
    const data = JSON.parse(event.data);
    const chatLog = document.getElementById('chat-log');
    const messageEl = document.createElement('div');
    messageEl.textContent = `${data.username}: ${data.message}`;
    chatLog.appendChild(messageEl);
    // 自動捲動到最新訊息
    chatLog.scrollTop = chatLog.scrollHeight;
};

// 連線關閉
ws.onclose = function(event) {
    console.warn('WebSocket 連線已關閉，代碼：', event.code);
};

// 連線錯誤
ws.onerror = function(event) {
    console.error('WebSocket 發生錯誤');
};

// 發送訊息
document.getElementById('send-btn').addEventListener('click', function() {
    const input = document.getElementById('message-input');
    const message = input.value.trim();
    if (message) {
        ws.send(JSON.stringify({ message: message }));
        input.value = '';  // 清空輸入框
    }
});

// 按 Enter 發送訊息
document.getElementById('message-input').addEventListener('keypress', function(e) {
    if (e.key === 'Enter') {
        document.getElementById('send-btn').click();
    }
});

Django Template

<!-- chat/templates/chat/room.html -->

{% extends "base.html" %}

{% block content %}
<h2>聊天室：{{ room_name }}</h2>

<div id="chat-log" style="height: 400px; overflow-y: auto; border: 1px solid #ccc; padding: 10px; margin-bottom: 10px;">
    <!-- 訊息會透過 JavaScript 動態插入 -->
</div>

<div>
    <input id="message-input" type="text" placeholder="輸入訊息..." style="width: 80%;">
    <button id="send-btn">發送</button>
</div>

<!-- 將聊天室名稱傳給 JavaScript -->
{{ room_name|json_script:"room-name" }}

<script>
    const roomName = JSON.parse(document.getElementById('room-name').textContent);
    // ... 上方的 JavaScript 程式碼
</script>
{% endblock %}

Django View 與 URL

# chat/views.py

from django.shortcuts import render


def chat_room(request, room_name):
    """聊天室頁面 View。"""
    return render(request, 'chat/room.html', {
        'room_name': room_name,
    })

# chat/urls.py

from django.urls import path
from . import views

urlpatterns = [
    path('chat/<str:room_name>/', views.chat_room, name='chat-room'),
]

在非同步 Consumer 中存取 Django ORM

Django ORM 是同步的，在 AsyncWebsocketConsumer 中直接呼叫 ORM 操作會引發錯誤。需要使用 database_sync_to_async 裝飾器將同步的 ORM 操作包裝為非同步函式：

# chat/consumers.py

from channels.db import database_sync_to_async
from chat.models import Message


class ChatConsumer(AsyncWebsocketConsumer):

    @database_sync_to_async
    def save_message(self, room_name, username, message):
        """在執行緒池中同步執行 ORM 操作。"""
        return Message.objects.create(
            room=room_name,
            author=username,
            content=message,
        )

    async def receive(self, text_data):
        data = json.loads(text_data)
        message = data['message']
        username = self.scope['user'].username

        # 非同步地儲存訊息到資料庫
        await self.save_message(self.room_name, username, message)

        # 廣播訊息給聊天室成員
        await self.channel_layer.group_send(
            self.room_group_name,
            {
                'type': 'chat_message',
                'message': message,
                'username': username,
            }
        )

WebSocket 認證處理

Django Channels 的 AuthMiddlewareStack 會自動從 Session Cookie 取得認證使用者並注入到 scope['user']。但如果你的前端使用 Token 認證（如 JWT），WebSocket 無法透過標準的 HTTP Header 傳遞 Token，需要自訂 Middleware：

# chat/middleware.py

from channels.middleware import BaseMiddleware
from channels.db import database_sync_to_async
from django.contrib.auth.models import AnonymousUser
from rest_framework.authtoken.models import Token


class TokenAuthMiddleware(BaseMiddleware):
    """從 WebSocket URL 查詢參數中取得 Token 並認證使用者。"""

    async def __call__(self, scope, receive, send):
        # 從 query string 解析 token 參數
        query_string = scope.get('query_string', b'').decode()
        params = dict(
            p.split('=') for p in query_string.split('&') if '=' in p
        )
        token_key = params.get('token')

        if token_key:
            scope['user'] = await self.get_user(token_key)
        else:
            scope['user'] = AnonymousUser()

        return await super().__call__(scope, receive, send)

    @database_sync_to_async
    def get_user(self, token_key):
        """透過 Token 取得對應的使用者。"""
        try:
            token = Token.objects.select_related('user').get(key=token_key)
            return token.user
        except Token.DoesNotExist:
            return AnonymousUser()

在 Consumer 中可以拒絕未認證的連線：

async def connect(self):
    if not self.scope['user'].is_authenticated:
        await self.close(code=4001)  # 使用自訂關閉代碼
        return
    # ... 正常連線邏輯
    await self.accept()

前端連線時在 URL 中附帶 Token：

// Token 認證方式的 WebSocket 連線
const token = 'your-auth-token';
const ws = new WebSocket(
    `ws://${window.location.host}/ws/chat/${roomName}/?token=${token}`
);

測試 WebSocket Consumer

Django Channels 提供 WebsocketCommunicator 用於測試 Consumer：

# chat/tests/test_consumers.py

import pytest
from channels.testing import WebsocketCommunicator
from channels.layers import get_channel_layer
from myproject.asgi import application


@pytest.mark.asyncio
async def test_chat_consumer_connect():
    """測試 WebSocket Consumer 的連線。"""
    communicator = WebsocketCommunicator(
        application, '/ws/chat/test-room/'
    )

    # 測試連線是否成功
    connected, subprotocol = await communicator.connect()
    assert connected

    # 測試斷開連線
    await communicator.disconnect()


@pytest.mark.asyncio
async def test_chat_consumer_send_receive():
    """測試 WebSocket Consumer 的訊息傳遞。"""
    communicator = WebsocketCommunicator(
        application, '/ws/chat/test-room/'
    )
    await communicator.connect()

    # 傳送訊息
    await communicator.send_json_to({'message': 'Hello'})

    # 接收並驗證回應
    response = await communicator.receive_json_from()
    assert response['message'] == 'Hello'

    await communicator.disconnect()

部署注意事項

部署使用 WebSocket 的 Django 專案時，需要使用支援 ASGI 的伺服器：

# 使用 Daphne（Channels 官方推薦）
daphne -b 0.0.0.0 -p 8000 myproject.asgi:application

# 使用 Uvicorn（效能更佳）
uvicorn myproject.asgi:application --host 0.0.0.0 --port 8000 --workers 4

Nginx 反向代理設定需要特別支援 WebSocket 升級：

# Nginx 設定（/etc/nginx/sites-available/myproject）
upstream channels-backend {
    server localhost:8000;
}

server {
    listen 80;
    server_name example.com;

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

    # WebSocket 路由需要額外的 Header 設定
    location /ws/ {
        proxy_pass http://channels-backend;
        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;
    }
}

關鍵是 /ws/ 路徑下要加入 Upgrade 和 Connection Header，讓 Nginx 支援 WebSocket 協定升級。proxy_read_timeout 設為較大的值，避免長時間閒置的 WebSocket 連線被中斷。

總結

本文從即時通訊的需求出發，完整介紹了 Django Channels 的核心概念與實作方式：

WebSocket vs HTTP：WebSocket 提供全雙工的持久連線，是即時通訊的最佳選擇
ASGI vs WSGI：ASGI 是支援非同步與長連接的新一代伺服器介面標準
安裝與設定：安裝 channels 和 channels_redis，設定 ASGI_APPLICATION 與 CHANNEL_LAYERS
Consumer：WebSocket 的 View，透過 connect、disconnect、receive 三個方法處理連線生命週期
Routing：使用 ProtocolTypeRouter 和 URLRouter 設定 WebSocket 路由
Channel Layers 與 Groups：透過 Redis 實現跨 Consumer 的訊息廣播，group_add、group_send、group_discard 是三個核心方法
完整聊天室範例：整合 Consumer、Routing、前端 JavaScript、Django Template 的端到端實作

Django Channels 讓你不必離開 Django 的生態系就能實現強大的即時通訊功能。在下一篇中，我們將學習 Django 的測試框架，確保包含 WebSocket 在內的所有功能都有完善的測試覆蓋。