Django Channels: WebSocket и асинхронность в Django

Django Channels — официальный пакет проекта Django, который добавляет к обычным view WebSocket и другие долгоживущие соединения. Он работает поверх ASGI (асинхронного интерфейса между сервером и приложением): HTTP-запросы по-прежнему обрабатывает Django, а WebSocket-соединения — классы-обработчики, которые называются consumers. Чтобы сообщение из одного соединения дошло до других (чат, уведомления), Channels использует channel layer — общую шину сообщений, в продакшне на Redis.

Быстрый ответ: установите channels[daphne] и channels-redis, поставьте daphne первым в INSTALLED_APPS, укажите ASGI_APPLICATION, оберните asgi.py в ProtocolTypeRouter, опишите consumer и маршрут в routing.py, настройте CHANNEL_LAYERS. Для разработки хватает python manage.py runserver, в продакшне — Daphne или Uvicorn за Nginx. Весь код ниже проверен на Channels 4.3.2, Daphne 4.2.3 и channels-redis 4.3.0 с Django 5.2 LTS, 6.0 и 6.1 (Python 3.12). Проект в примерах называется config, приложение — chat.

Зачем нужен Django Channels

Обычный Django работает по схеме «запрос — ответ»: браузер спрашивает, сервер отвечает, соединение закрывается. Сервер не может сам отправить данные клиенту, когда что-то изменилось. WebSocket решает эту задачу: соединение открывается один раз и остаётся открытым, а сообщения идут в обе стороны в любой момент.

Под WSGI (Gunicorn, классический runserver) долгоживущие соединения не поддерживаются. Django умеет работать под ASGI и поддерживает асинхронные view, но сам WebSocket не обрабатывает. Эту часть берёт на себя Channels:

• маршрутизация по протоколам — HTTP идёт в Django, WebSocket — в consumers;
• consumers — классы с методами connect, receive, disconnect, синхронные и асинхронные;
• channel layers и группы — рассылка сообщений между соединениями, процессами и серверами;
• аутентификация — scope["user"] из обычной сессии Django;
• тестовые помощники — WebsocketCommunicator для тестов без запуска сервера.

Типичные задачи: чат, уведомления в реальном времени, живые дашборды, прогресс долгой операции, совместное редактирование, многопользовательские игры. Если сервер только иногда отправляет события клиенту, Channels может оказаться избыточным — варианты проще описаны ниже, в разделе об альтернативах.

Как устроен Channels

Путь WebSocket-соединения через проект:

Браузер (ws:// или wss://)
  → Nginx (проксирует Upgrade: websocket)
  → ASGI-сервер (Daphne или Uvicorn)
  → config/asgi.py: ProtocolTypeRouter
      ├─ "http"      → обычный Django (views, админка, DRF)
      └─ "websocket" → AllowedHostsOriginValidator → AuthMiddlewareStack → URLRouter
                        → ChatConsumer (по одному объекту на соединение)
                             ↕ channel layer (Redis): группы и рассылка

Компонент	Что делает	Аналог в обычном Django
ASGI-сервер	Принимает HTTP и WebSocket, держит соединения	Gunicorn
ProtocolTypeRouter	Разводит трафик по протоколам	—
URLRouter и routing.py	Сопоставляет путь WebSocket с consumer	urls.py
Consumer	Обрабатывает события одного соединения	view
scope	Данные о соединении: путь, заголовки, пользователь	request
Channel layer	Передаёт сообщения между consumers и процессами	—

Установка Channels, Daphne и channels-redis

Установите пакеты в виртуальное окружение проекта:

# Windows
python -m venv venv
venv\Scripts\activate

# macOS / Linux
python3 -m venv venv
source venv/bin/activate

python -m pip install "django>=5.2" "channels[daphne]" channels-redis

• channels — сам фреймворк. Требует Django 4.2+ и Python 3.9+.
• [daphne] — дополнительно ставит ASGI-сервер Daphne. С версии Channels 4 он не устанавливается автоматически.
• channels-redis — channel layer на Redis для продакшна. Для первых шагов не обязателен.

На Windows 11 с Python 3.12 все пакеты ставятся из готовых wheel, компилятор C++ не нужен.

Добавьте приложения и настройки в config/settings.py:

# config/settings.py
INSTALLED_APPS = [
    "daphne",  # первым: подменяет runserver на ASGI-версию
    "django.contrib.admin",
    "django.contrib.auth",
    "django.contrib.contenttypes",
    "django.contrib.sessions",
    "django.contrib.messages",
    "django.contrib.staticfiles",
    "channels",
    "chat",
]

ASGI_APPLICATION = "config.asgi.application"

# Для разработки и тестов: сообщения живут в памяти одного процесса
CHANNEL_LAYERS = {
    "default": {"BACKEND": "channels.layers.InMemoryChannelLayer"},
}

daphne должен стоять выше django.contrib.staticfiles и других приложений со своей командой runserver, иначе запустится обычный WSGI-сервер. Проверьте вывод python manage.py runserver:

Django version 6.0.9, using settings 'config.settings'
Starting ASGI/Daphne version 4.2.3 development server at http://127.0.0.1:8000/

Если вместо этого видна строка Starting development server at ..., Daphne не подключился, и WebSocket-запросы будут получать 404.

Redis для channel layer

InMemoryChannelLayer не передаёт сообщения между процессами. Как только у вас несколько воркеров сервера или сообщения отправляет отдельный процесс (воркер Celery, Telegram-бот, management-команда), нужен Redis:

# config/settings.py
import os

CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {
            "hosts": [os.environ.get("REDIS_URL", "redis://127.0.0.1:6379/0")],
        },
    },
}

В hosts можно передать URL Redis строкой или кортеж ("127.0.0.1", 6379). Сервер Redis для разработки проще всего поднять в Docker: docker run -d -p 6379:6379 redis:7.

asgi.py и маршрутизация WebSocket

Корневое ASGI-приложение описывается в config/asgi.py. Оно разводит протоколы: HTTP уходит в стандартный Django, WebSocket — в маршруты Channels.

# config/asgi.py
import os

from django.core.asgi import get_asgi_application

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings")
# Сначала инициализируем Django, потом импортируем код с моделями
django_asgi_app = get_asgi_application()

from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.security.websocket import AllowedHostsOriginValidator

from chat.routing import websocket_urlpatterns

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

• get_asgi_application() вызывается до импорта consumers. Иначе импорт моделей упадёт с AppRegistryNotReady.
• AuthMiddlewareStack читает cookie сессии и кладёт пользователя в scope["user"].
• AllowedHostsOriginValidator отклоняет соединения, у которых заголовок Origin не входит в ALLOWED_HOSTS. Это защита от подключения к вашему сокету с чужого сайта с cookie пользователя.

Маршруты WebSocket хранятся отдельно от urls.py — обычно в routing.py приложения:

# chat/routing.py
from django.urls import path

from . import consumers

websocket_urlpatterns = [
    path("ws/echo/", consumers.EchoConsumer.as_asgi()),
    path("ws/chat/<slug:room>/", consumers.ChatConsumer.as_asgi()),
    path("ws/notify/", consumers.NotifyConsumer.as_asgi()),
]

Синтаксис такой же, как в urls.py: path, re_path, конвертеры вроде <slug:room>. У consumer вызывается .as_asgi() — по аналогии с .as_view() у классовых view. Маршруты WebSocket, добавленные в urls.py, работать не будут.

Consumers: синхронные и асинхронные

Consumer — это класс, объект которого создаётся на каждое WebSocket-соединение и живёт, пока соединение открыто. Готовые базовые классы лежат в channels.generic.websocket:

Класс	Код методов	Формат сообщений	Когда брать
WebsocketConsumer	def	текст или байты	Простая синхронная логика, много вызовов ORM
AsyncWebsocketConsumer	async def	текст или байты	Много соединений, ожидание сети, рассылка по группам
JsonWebsocketConsumer	def	JSON (receive_json, send_json)	Синхронный обмен JSON
AsyncJsonWebsocketConsumer	async def	JSON	Асинхронный обмен JSON — самый частый выбор

Основные методы:

• connect() — клиент открывает соединение. Здесь проверяют права и вызывают accept() или close().
• receive() или receive_json() — пришло сообщение от клиента.
• send() или send_json() — отправить сообщение этому клиенту.
• disconnect(code) — соединение закрыто; здесь убирают канал из групп.
• self.scope — словарь с путём, параметрами маршрута (scope["url_route"]["kwargs"]), заголовками и пользователем.

Минимальный пример — эхо-сервер на JsonWebsocketConsumer:

# chat/consumers.py
from channels.generic.websocket import JsonWebsocketConsumer


class EchoConsumer(JsonWebsocketConsumer):
    def connect(self):
        self.accept()

    def receive_json(self, content):
        self.send_json({"echo": content})

Клиент отправляет {"ping": 1} и получает {"echo": {"ping": 1}}.

Синхронный или асинхронный consumer

Синхронный consumer выполняется в пуле потоков, поэтому в нём можно спокойно вызывать ORM. Методы channel layer у него приходится оборачивать в async_to_sync, потому что слой асинхронный:

from asgiref.sync import async_to_sync
from channels.generic.websocket import WebsocketConsumer


class SyncChatConsumer(WebsocketConsumer):
    def connect(self):
        self.group_name = "chat_sync"
        async_to_sync(self.channel_layer.group_add)(self.group_name, self.channel_name)
        self.accept()

    def disconnect(self, code):
        async_to_sync(self.channel_layer.group_discard)(self.group_name, self.channel_name)

    def receive(self, text_data=None, bytes_data=None):
        async_to_sync(self.channel_layer.group_send)(
            self.group_name, {"type": "chat.message", "text": text_data}
        )

    def chat_message(self, event):
        self.send(text_data=event["text"])

Асинхронный consumer не занимает поток на каждое соединение и лучше подходит для большого числа открытых сокетов. Цена — дисциплина: внутри async def нельзя вызывать блокирующий код (time.sleep(), requests, синхронный ORM). Если основа async-кода для вас новая, начните со статьи «async и await в Python».

ORM в асинхронном consumer

Прямой вызов ORM внутри async def Django остановит ошибкой:

django.core.exceptions.SynchronousOnlyOperation: You cannot call this from an async context - use a thread or sync_to_async.

Два рабочих способа:

• вынести запрос в метод с декоратором database_sync_to_async из channels.db. Он выполняет код в потоке и закрывает устаревшие соединения с БД, как это делает Django после каждого запроса;
• использовать асинхронные методы ORM: await Message.objects.acreate(...), aget(), acount(), async for по QuerySet.

Для нескольких связанных запросов удобнее первый способ: весь блок выполняется в одном потоке, в том числе внутри transaction.atomic().

Channel layers и группы: чат на Channels

У каждого consumer есть уникальное имя канала self.channel_name. Группа объединяет несколько каналов под одним именем: group_add добавляет канал, group_discard удаляет, group_send рассылает событие всем участникам. Ключ type в событии определяет метод-обработчик: "chat.message" вызывает chat_message (точки заменяются подчёркиваниями).

Модель для истории сообщений:

# chat/models.py
from django.db import models


class Message(models.Model):
    room = models.CharField(max_length=50)
    author = models.CharField(max_length=150)
    text = models.TextField()
    created = models.DateTimeField(auto_now_add=True)

Consumer комнаты чата:

# chat/consumers.py
from channels.db import database_sync_to_async
from channels.generic.websocket import AsyncJsonWebsocketConsumer

from .models import Message


class ChatConsumer(AsyncJsonWebsocketConsumer):
    async def connect(self):
        self.room = self.scope["url_route"]["kwargs"]["room"]
        self.group_name = f"chat_{self.room}"
        await self.channel_layer.group_add(self.group_name, self.channel_name)
        await self.accept()

    async def disconnect(self, code):
        await self.channel_layer.group_discard(self.group_name, self.channel_name)

    async def receive_json(self, content):
        user = self.scope["user"]
        author = user.username if user.is_authenticated else "гость"
        await self.save_message(author, content["text"])
        await self.channel_layer.group_send(
            self.group_name,
            {
                "type": "chat.message",  # вызовет метод chat_message
                "author": author,
                "text": content["text"],
            },
        )

    async def chat_message(self, event):
        await self.send_json({"author": event["author"], "text": event["text"]})

    @database_sync_to_async
    def save_message(self, author, text):
        return Message.objects.create(room=self.room, author=author, text=text)

Сообщение от одного клиента сохраняется в базе, уходит в группу chat_lobby и возвращается всем подключённым к комнате, включая отправителя. После добавления модели выполните python manage.py makemigrations и migrate — чтобы создать таблицу.

Имя группы может содержать только латинские буквы, цифры, дефис, подчёркивание и точку и должно быть короче 100 символов. Иначе group_add выбросит TypeError: Group name must be a valid unicode string.... Поэтому в маршруте стоит конвертер <slug:room>: кириллическое имя комнаты до consumer просто не дойдёт.

Клиент в браузере — стандартный WebSocket API:

// static/chat/chat.js
const room = 'lobby';
const scheme = location.protocol === 'https:' ? 'wss' : 'ws';
const socket = new WebSocket(`${scheme}://${location.host}/ws/chat/${room}/`);

socket.onopen = () => socket.send(JSON.stringify({ text: 'Привет' }));
socket.onmessage = (event) => {
  const data = JSON.parse(event.data);
  console.log(`${data.author}: ${data.text}`);
};
socket.onclose = (event) => console.log('Соединение закрыто', event.code);

Страница с этим скриптом должна открываться с того же домена, что и сокет: тогда браузер передаст cookie сессии, а Origin пройдёт проверку. Браузер не переподключается сам: если соединение оборвалось, клиент должен открыть его заново, лучше с нарастающей паузой.

Отправка сообщений из view, сигнала или задачи Celery

Channel layer доступен вне consumers через get_channel_layer(). Его методы асинхронные, поэтому из синхронного кода их вызывают через async_to_sync:

# chat/notify.py
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer


def notify_user(user_id, message):
    """Отправить событие в WebSocket пользователя из синхронного кода."""
    channel_layer = get_channel_layer()
    async_to_sync(channel_layer.group_send)(
        f"user_{user_id}", {"type": "notify", "message": message}
    )

Consumer подписывает каждого пользователя на личную группу. Так уведомление дойдёт во все его вкладки и устройства:

# chat/consumers.py
class NotifyConsumer(AsyncJsonWebsocketConsumer):
    async def connect(self):
        user = self.scope["user"]
        if not user.is_authenticated:
            await self.close(code=4401)
            return
        self.group_name = f"user_{user.pk}"
        await self.channel_layer.group_add(self.group_name, self.channel_name)
        await self.accept()

    async def disconnect(self, code):
        if hasattr(self, "group_name"):
            await self.channel_layer.group_discard(self.group_name, self.channel_name)

    async def notify(self, event):
        await self.send_json({"message": event["message"]})

notify_user(order.user_id, "Заказ оплачен") можно вызвать из view, обработчика сигнала post_save или задачи Celery — например, чтобы показать прогресс долгой обработки. Если отправитель работает в другом процессе (воркер Celery), channel layer обязательно должен быть на Redis.

Аутентификация в WebSocket

С AuthMiddlewareStack пользователь берётся из сессии Django — той же, что создаёт обычный вход на сайт (см. статью об аутентификации в Django). В consumer он доступен как self.scope["user"], для гостя это AnonymousUser. Права проверяют в connect():

async def connect(self):
    user = self.scope["user"]
    if not user.is_authenticated:
        await self.close(code=4401)
        return
    await self.accept()

close() до accept() отклоняет рукопожатие: Daphne и Uvicorn отвечают клиенту HTTP 403, и браузер увидит закрытие с кодом 1006. Если клиенту нужен свой код закрытия (например, 4401 «требуется вход»), сначала вызовите accept(), затем close(code=4401).

Токен для SPA и мобильных приложений

Браузерный WebSocket API не позволяет задать заголовок Authorization. Если фронтенд живёт на другом домене или это мобильное приложение, токен передают в query string или первым сообщением. Пример middleware, который принимает короткоживущий подписанный токен:

# chat/middleware.py
from urllib.parse import parse_qs

from channels.db import database_sync_to_async
from channels.middleware import BaseMiddleware
from django.contrib.auth import get_user_model
from django.contrib.auth.models import AnonymousUser
from django.core import signing


@database_sync_to_async
def get_user_by_token(token):
    try:
        user_id = signing.loads(token, salt="ws", max_age=60)
        return get_user_model().objects.get(pk=user_id)
    except (signing.BadSignature, get_user_model().DoesNotExist):
        return AnonymousUser()


class TokenAuthMiddleware(BaseMiddleware):
    async def __call__(self, scope, receive, send):
        params = parse_qs(scope["query_string"].decode())
        token = params.get("token", [None])[0]
        if token:
            scope["user"] = await get_user_by_token(token)
        return await super().__call__(scope, receive, send)

Токен выдаёт обычный HTTP-эндпоинт для вошедшего пользователя: signing.dumps(request.user.pk, salt="ws"). Клиент подключается к wss://example.com/ws/notify/?token=.... Middleware ставят внутрь стека, чтобы токен переопределял пользователя из сессии: AuthMiddlewareStack(TokenAuthMiddleware(URLRouter(...))). Query string попадает в логи сервера, поэтому делайте токен короткоживущим — в примере он действует 60 секунд. Для API на JWT логика та же: middleware проверяет токен и кладёт пользователя в scope.

Тестирование: WebsocketCommunicator

WebsocketCommunicator из channels.testing подключается к ASGI-приложению напрямую, без сервера и сети. С InMemoryChannelLayer тесты не требуют Redis. Удобнее всего писать их на pytest — подробнее о настройке в статье «Тестирование Django с pytest»:

python -m pip install pytest pytest-django pytest-asyncio

# pytest.ini
[pytest]
DJANGO_SETTINGS_MODULE = config.settings
python_files = tests.py test_*.py

# chat/tests.py
import pytest
from asgiref.sync import sync_to_async
from channels.routing import URLRouter
from channels.testing import WebsocketCommunicator
from django.contrib.auth.models import AnonymousUser, User

from chat.models import Message
from chat.notify import notify_user
from chat.routing import websocket_urlpatterns
from config.asgi import application

# Consumers закрывают старые соединения с БД, поэтому тестам нужен доступ к базе
pytestmark = [pytest.mark.asyncio, pytest.mark.django_db(transaction=True)]

app = URLRouter(websocket_urlpatterns)  # без проверки Origin и сессий


async def test_chat_broadcast():
    alice = WebsocketCommunicator(app, "/ws/chat/lobby/")
    bob = WebsocketCommunicator(app, "/ws/chat/lobby/")
    alice.scope["user"] = bob.scope["user"] = AnonymousUser()
    assert (await alice.connect())[0]
    assert (await bob.connect())[0]

    await alice.send_json_to({"text": "Привет"})
    expected = {"author": "гость", "text": "Привет"}
    assert await alice.receive_json_from() == expected
    assert await bob.receive_json_from() == expected
    assert await Message.objects.acount() == 1

    await alice.disconnect()
    await bob.disconnect()


async def test_notify_from_sync_code():
    user = await User.objects.acreate(username="anna")
    ws = WebsocketCommunicator(app, "/ws/notify/")
    ws.scope["user"] = user
    assert (await ws.connect())[0]

    await sync_to_async(notify_user)(user.pk, "Заказ оплачен")
    assert await ws.receive_json_from() == {"message": "Заказ оплачен"}
    await ws.disconnect()


async def test_anonymous_is_rejected():
    ws = WebsocketCommunicator(
        application, "/ws/notify/", headers=[(b"origin", b"http://localhost")]
    )
    connected, code = await ws.connect()
    assert not connected
    assert code == 4401


async def test_foreign_origin_is_rejected():
    ws = WebsocketCommunicator(
        application, "/ws/chat/lobby/", headers=[(b"origin", b"https://evil.example")]
    )
    connected, _ = await ws.connect()
    assert not connected

$ python -m pytest chat/tests.py -v
chat/tests.py::test_chat_broadcast PASSED                                [ 25%]
chat/tests.py::test_notify_from_sync_code PASSED                         [ 50%]
chat/tests.py::test_anonymous_is_rejected PASSED                         [ 75%]
chat/tests.py::test_foreign_origin_is_rejected PASSED                    [100%]

============================== 4 passed in 0.55s ==============================

На что обратить внимание:

• connect() возвращает кортеж: (True, None) при успехе или (False, код_закрытия) при отказе.
• Пользователя можно подставить напрямую через communicator.scope["user"], если тестируете маршруты без AuthMiddlewareStack.
• Полное приложение из asgi.py проверяет Origin: без заголовка origin из ALLOWED_HOSTS соединение будет отклонено. В режиме DEBUG с пустым ALLOWED_HOSTS разрешены localhost и 127.0.0.1.
• Без отметки django_db тест падает при отключении с RuntimeError: Database access not allowed: consumers Channels 4.3 обращаются к соединению с БД при обработке каждого события.
• Всегда вызывайте await communicator.disconnect(), иначе pytest выдаст предупреждения о незавершённых корутинах.

Тот же сценарий работает и со стандартным python manage.py test: в django.test.TestCase достаточно объявить тестовый метод через async def.

Деплой: Daphne или Uvicorn за Nginx

Gunicorn с обычным WSGI-воркером WebSocket не обслуживает. Проект с Channels запускают под ASGI-сервером, а Nginx перед ним раздаёт статику, держит HTTPS и проксирует заголовки Upgrade. Общая схема деплоя, systemd и certbot разобраны в статье «Деплой Django: Gunicorn и Nginx»; ниже — что меняется для Channels.

	Daphne	Uvicorn
Установка	channels[daphne]	uvicorn[standard]
Несколько процессов	Нет --workers: несколько экземпляров через systemd или supervisor	--workers 4
Ограничение жизни соединения	--websocket_timeout, по умолчанию 86400 с (сутки), -1 — без ограничения	Нет такой настройки
Keepalive-пинги	Каждые 20 с простоя	Каждые 20 с (--ws-ping-interval)
Заголовки прокси	--proxy-headers	Включены по умолчанию для 127.0.0.1

# Daphne
daphne -b 127.0.0.1 -p 8001 --proxy-headers config.asgi:application

# Uvicorn: нужен extra [standard] или пакет websockets
python -m pip install "uvicorn[standard]"
uvicorn config.asgi:application --host 127.0.0.1 --port 8001 --workers 4

С голым pip install uvicorn поддержки WebSocket нет: сервер запустится, HTTP будет работать, а на запросы к сокету в логе появится No supported WebSocket library detected и ответ 404. Подробнее о параметрах сервера — в статье «Uvicorn: запуск ASGI-приложений».

Конфигурация Nginx, когда весь трафик идёт через ASGI-сервер:

# /etc/nginx/sites-available/mysite
map $http_upgrade $connection_upgrade {
    default upgrade;
    ''      close;
}

upstream django_asgi {
    server 127.0.0.1:8001;
}

server {
    server_name example.com;

    location /static/ {
        alias /srv/mysite/staticfiles/;
    }

    location /ws/ {
        proxy_pass http://django_asgi;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection $connection_upgrade;
        proxy_set_header Host $host;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        proxy_read_timeout 1h;
    }

    location / {
        proxy_pass http://django_asgi;
        proxy_set_header Host $host;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}

• Без proxy_http_version 1.1 и заголовков Upgrade/Connection рукопожатие WebSocket не пройдёт.
• proxy_read_timeout по умолчанию 60 секунд. Пинги сервера обычно держат соединение живым, но запас не помешает.
• На сайте с HTTPS клиент подключается по wss://. Сертификат Nginx тот же, отдельная настройка не нужна.
• Домен должен быть в ALLOWED_HOSTS, иначе AllowedHostsOriginValidator отклонит соединения с ответом 403.

Если переводить весь проект на ASGI пока не хочется, можно оставить HTTP на Gunicorn, а на ASGI-сервер направлять только location /ws/. При разделении по пути сайт и сокет остаются на одном домене, и cookie сессии у них общие; при выносе сокета на поддомен этого не будет.

В продакшне channel layer — только Redis: с InMemoryChannelLayer и несколькими воркерами сообщение, отправленное в одном процессе, не дойдёт до клиентов в другом. Состояние комнат (кто онлайн, история) по той же причине храните в Redis или базе, а не в глобальных переменных модуля. В Docker Compose схема та же: отдельные сервисы для приложения, Redis и Nginx.

Когда Channels не нужен: альтернативы

Channels добавляет ASGI-сервер, Redis и асинхронный код. Если двусторонний обмен в реальном времени не нужен, задача решается проще.

Подход	Направление	Что нужно на сервере	Подходит для
Опрос (polling), в том числе HTMX hx-trigger="every 10s"	Клиент спрашивает	Обычный Django	Обновления раз в десятки секунд
SSE через асинхронный view	Сервер → клиент	ASGI-сервер, без Channels	Уведомления, прогресс, ленты событий
django-eventstream	Сервер → клиент (SSE)	Daphne или другой ASGI; Redis для нескольких процессов	Готовые каналы событий и send_event()
Django Channels	В обе стороны	ASGI-сервер, Redis	Чаты, совместная работа, игры
Отдельный сервис (FastAPI, Centrifugo, python-socketio)	В обе стороны	Ещё один сервис	Real-time как отдельная часть системы

Server-Sent Events (SSE) — односторонний поток событий поверх обычного HTTP. Браузер получает его через EventSource и сам переподключается. Под ASGI-сервером Django отдаёт SSE асинхронным view без дополнительных пакетов:

# chat/views.py
import asyncio

from django.http import StreamingHttpResponse


async def events(request):
    async def stream():
        for i in range(3):
            yield f"data: тик {i}\n\n"
            await asyncio.sleep(1)

    return StreamingHttpResponse(stream(), content_type="text/event-stream")

$ curl -N http://127.0.0.1:8000/events/
data: тик 0

data: тик 1

data: тик 2

django-eventstream берёт на себя каналы и доставку SSE: подключаете его URL и вызываете send_event("test", "message", {"text": "hello"}) из любого места кода. Для нескольких экземпляров приложения нужен Redis или прокси Pushpin.

Отдельный сервис имеет смысл, когда real-time — самостоятельная часть системы или команда уже работает с FastAPI. Сервис держит соединения, а Django публикует события через Redis или HTTP. FastAPI поддерживает WebSocket из коробки. Centrifugo — готовый сервер реального времени, которому Django отправляет сообщения по HTTP API. python-socketio можно смонтировать рядом с Django в одном ASGI-приложении, если клиенты уже используют Socket.IO.

Для частичного обновления страниц без WebSocket посмотрите HTMX и Django: у HTMX есть и опрос, и расширения для SSE и WebSocket.

Нужен ли Channels для Telegram-бота

Обычно нет. Telegram-бот — отдельный процесс, который сам опрашивает серверы Telegram (long polling) или принимает от них HTTPS-запросы (webhook). WebSocket для этого не нужен. Бот пишут на aiogram или python-telegram-bot, а с Django его связывает общая база: бот запускается как management-команда и импортирует модели.

Channels пригодится в одном случае: события из бота нужно показать в браузере в реальном времени, например сообщения из Telegram в чате на сайте. Тогда бот отправляет событие в группу через channel layer, а consumer доставляет его клиентам. Пример на aiogram 3:

# chat/management/commands/runbot.py
import asyncio

from aiogram import Bot, Dispatcher
from aiogram.filters import CommandStart
from aiogram.types import Message as TgMessage
from channels.layers import get_channel_layer
from django.conf import settings
from django.core.management.base import BaseCommand

from chat.models import Message

dp = Dispatcher()


@dp.message(CommandStart())
async def start(message: TgMessage):
    await message.answer("Пишите — сообщения появятся в чате на сайте.")


@dp.message()
async def to_site(message: TgMessage):
    author = message.from_user.full_name
    text = message.text or ""
    await Message.objects.acreate(room="telegram", author=author, text=text)
    # Опционально: показать сообщение в браузере через Channels (нужен Redis-слой)
    await get_channel_layer().group_send(
        "chat_telegram", {"type": "chat.message", "author": author, "text": text}
    )
    await message.answer("Отправлено на сайт")


class Command(BaseCommand):
    help = "Запускает Telegram-бота (long polling)"

    def handle(self, *args, **options):
        bot = Bot(token=settings.TELEGRAM_BOT_TOKEN)
        asyncio.run(dp.start_polling(bot))

Бот запускается командой python manage.py runbot отдельно от веб-сервера. Токен храните в переменной окружения и читайте в settings.py. Обработчики aiogram асинхронные, поэтому ORM вызывается асинхронными методами (acreate, aget). Бот и веб-сервер — разные процессы, поэтому событие дойдёт до браузера только через Redis-слой. Без group_send этот же код — обычный бот, которому Channels не нужен. Для webhook хватает обычного Django-view, который принимает POST от Telegram.

Частые ошибки

WebSocket возвращает 404

Сокет обслуживает WSGI-сервер. Проверьте, что daphne первым стоит в INSTALLED_APPS и runserver пишет Starting ASGI/Daphne, а в продакшне запущен Daphne или uvicorn[standard], а не Gunicorn с WSGI. Ещё одна причина — маршрут добавлен в urls.py вместо routing.py.

Соединение отклоняется с кодом 403

Сработал AllowedHostsOriginValidator: домен страницы не входит в ALLOWED_HOSTS. Другая причина — consumer вызвал close() до accept(), например для неавторизованного пользователя.

SynchronousOnlyOperation в consumer

ORM вызван напрямую в async def. Оберните запрос в database_sync_to_async или используйте acreate(), aget() и другие асинхронные методы.

Сообщения не доходят до других пользователей

Проверьте имя настройки: CHANNEL_LAYERS, без лишней буквы. Если сервер работает в несколько процессов или отправитель — отдельный процесс, InMemoryChannelLayer не подойдёт, нужен Redis. Убедитесь также, что обработчик назван по type события: "chat.message" → chat_message.

Соединения обрываются через минуту или через сутки

Через минуту обычно обрывает Nginx с proxy_read_timeout по умолчанию, через сутки — Daphne с --websocket_timeout 86400. Увеличьте таймауты и добавьте на клиенте переподключение.

Частые вопросы

Можно ли работать с WebSocket в Django без Channels?

Можно, но придётся писать низкоуровневое ASGI-приложение для протокола websocket и самому решать рассылку между процессами. Другие варианты — отдельный сервис (FastAPI, Centrifugo, python-socketio), который держит соединения и получает события от Django через Redis или HTTP. Для одностороннего потока событий хватает SSE в асинхронном view.

Чем Channels отличается от Celery?

Celery выполняет фоновые задачи вне HTTP-запроса: письма, отчёты, обработку файлов. Channels держит открытые соединения с клиентами и доставляет им события. Их часто используют вместе: задача Celery работает в фоне и через async_to_sync(channel_layer.group_send) сообщает браузеру о прогрессе.

Нужен ли Redis для разработки?

Нет. InMemoryChannelLayer подходит для runserver и тестов, пока всё работает в одном процессе. Redis нужен, как только появляются несколько воркеров сервера или события отправляет другой процесс: Celery, бот, management-команда.

Работает ли Channels вместе с Django REST Framework?

Да. HTTP-запросы по-прежнему обрабатывают view DRF, а WebSocket — consumers в том же проекте. Если API использует токены или JWT, для сокета нужен собственный middleware, который проверяет токен и кладёт пользователя в scope, как в примере выше.

Как отправить сообщение конкретному пользователю?

Подпишите каждое соединение пользователя на личную группу, например user_42, и отправляйте событие в неё через group_send. Сообщение получат все его вкладки и устройства. Слой Channels не умеет перечислять участников группы, поэтому список онлайн-пользователей храните отдельно, например в Redis.

Итог

Django Channels добавляет в проект WebSocket и рассылку событий в реальном времени. Настройка сводится к четырём шагам: daphne и channels в INSTALLED_APPS, ProtocolTypeRouter в asgi.py, consumers с маршрутами в routing.py, channel layer на Redis для продакшна. Асинхронный consumer с группами покрывает чат и уведомления, database_sync_to_async убирает конфликт с синхронным ORM, а WebsocketCommunicator позволяет тестировать сокеты без сервера. В продакшне проект запускают под Daphne или uvicorn[standard] за Nginx с заголовками Upgrade. Для одностороннего потока событий проще SSE, а Telegram-боту Channels нужен, только если его события нужно показать в браузере в реальном времени. Остальные основы фреймворка собраны в обзоре «Django для начинающих».