Django Channels 实现 WebSocket 通信完整指南

苏见山

发布时间：2026-10-14| 3人浏览过

原创

本文详细介绍如何在 Django 项目中集成 Channels 库以实现 WebSocket 实时通信。内容涵盖依赖安装、settings.py 配置、Redis 服务部署、路由文件 routing.py 与消费者 consumers.py 的编写，以及通过 Django Shell 验证通信链路的完整步骤，帮助开发者快速搭建基于 Django 的即时通讯应用。

Django Channels 环境准备与配置

首先，需要安装 Channels 及其 Redis 后端依赖包。


pip install channels channels-redis

在 Django 项目的 settings.py 文件中，将 'channels' 添加到 INSTALLED_APPS 列表中，以启用 Channels 支持。


INSTALLED_APPS = [
  'django.contrib.admin',
  'django.contrib.auth',
  'django.contrib.contenttypes',
  'django.contrib.sessions',
  'django.contrib.messages',
  'django.contrib.staticfiles',
  'MyWeb.apps.MywebConfig',
  "channels",
]

核心文件创建与路由映射

创建 Django 项目后，在自定义应用目录下生成 consumers.py 和 routing.py 两个核心配置文件。

consumers.py 文件的作用类似于 Django 传统的 views.py，用于处理所有通过 WebSocket 路由过来的逻辑和函数。

routing.py 文件负责定义 WebSocket URL 与具体处理函数之间的映射关系，类似于 Django 的 urls.py。当 WebSocket 请求进入时，系统会根据映射关系触发 consumers.py 中对应的方法。

Redis 服务部署与通道层配置

接下来，需要安装并配置 Redis 服务，作为 Channels 的通道层后端。


redis 安装配置默认密码

yum install -y redis

[root@localhost ~]# vim /etc/redis.conf 开启远程
bind 0.0.0.0 
protected-mode no

redis-cli -h 192.168.1.20 -p 6379

在 settings.py 文件的末尾，添加 CHANNEL_LAYERS 配置，指定 Redis 作为通道层后端，并配置 ASGI_APPLICATION 指向路由应用。


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

ASGI_APPLICATION = "MyWeb.routing.application"

路由应用配置与通信测试

在 routing.py 文件中，配置 ProtocolTypeRouter，处理 HTTP 和 WebSocket 请求的路由分发。


from channels.routing import ProtocolTypeRouter

application = ProtocolTypeRouter({
  # Empty for now (http->django views is added by default)
})

最后，进入 Django Shell 环境，测试是否能成功连接到 Redis 数据库并发送接收消息。


(venv) C:\Users\LyShark\PycharmProjects\MyProject>manage.py shell
Python 3.7.4 (tags/v3.7.4:e09359112e, Jul 8 2019, 20:34:20) [MSC v.1916 64 bit (AMD64)] on win32
Type "help", "copyright", "credits" or "license" for more information.
(InteractiveConsole)
>>> import channels.layers
>>> channel_layer = channels.layers.get_channel_layer()
>>> from asgiref.sync import async_to_sync
>>> async_to_sync(channel_layer.send)('test_channel', {'type': 'hello'})
>>> async_to_sync(channel_layer.receive)('test_channel')
{'type': 'hello'}
>>>

以上即为 Django 使用 Channels 实现通信的完整示例，更多关于 Channels 实现通信的资料请关注本站其它相关文章！