Django Channels 实现点对点实时聊天与消息推送

今天早睡

发布时间：2026-10-13| 6人浏览过

原创

本文详解如何利用 Django Channels 结合 Redis 后端，实现基于 WebSocket 的点对点实时聊天与后台消息推送功能。通过构建 ChatConsumer 和 PushConsumer，处理连接生命周期与组广播，解决传统 HTTP 轮询带来的服务器压力与流量消耗问题，提供完整的代码配置与路由示例。

简介

简介：在许多实际项目开发场景中，实时功能的需求日益增长。本文将演示如何利用 Django Channels 快速实现点对点聊天及消息推送功能。

假设当前项目需要后台消息推送及用户间一对一在线聊天功能。例如，当用户 A 评论用户 B 的帖子时，系统需向用户 B 发送通知。若采用传统页面刷新并查询数据库的方式，不仅会增加后台服务器负载，对于手机客户端而言还会造成不必要的流量浪费。因此，考虑使用 WebSocket 建立长连接以优化此流程。

原生 Django 并不直接支持 WebSocket，经过调研，我们选择了 django-channels 项目。它允许 Django 项目不仅处理 HTTP 请求，还能处理 WebSockets、MQTT、聊天机器人及业余无线电等需要长期保持连接的协议。

笔者接触 Channels 时间尚短，为完成上述两个功能，查阅了大量文档并经历了不少调试过程。最终实现了基础功能，特此记录供参考。

安装 Channels

一：安装 Channels

若使用的 Django 版本为 1.9 或更高，在通过 pip 安装 Channels 时无需添加 -U 参数。

pip install channels

安装完成后，需将 Channels 作为应用添加到 Django 项目中。请在 settings.py 文件中添加相关配置。


INSTALLED_APPS = [
  'django.contrib.admin',
  'django.contrib.auth',
  'django.contrib.contenttypes',
  'django.contrib.sessions',
  'django.contrib.messages',
  'django.contrib.staticfiles',
  'Your-app',
  'channels',
]

此处选用 Redis 作为 Channels 的通道后端，以支持更多高级功能（具体功能将在后文提及）。因此，还需安装依赖包以保障其正常运行。

pip install channels_redis

随后在 settings.py 文件中添加通道层配置。


CHANNEL_LAYERS = {
  "default": {
    "BACKEND": "channels_redis.core.RedisChannelLayer",
    "CONFIG": {
      "hosts": [('127.0.0.1', 6379)],
    },
    # 配置路由的路径
    # "ROUTING": "exmchannels.routing.channel_routing",
  },
}
ASGI_APPLICATION = 'exmchannels.routing.application'

点对点聊天实现

二：点对点聊天实现

在项目目录下新建 channel 文件夹用于存放 Channels 相关代码。在 channel 目录下创建 consumers.py 文件，并定义 ChatConsumer 类以处理聊天相关的 WebSocket 请求。与普通聊天室不同，此处需在 ChatConsumer 中维护一个 chats 字典，用于记录每个 group 中的连接数量。通过判断该数量，可确定聊天双方是否均已加入同一聊天组。

聊天组的命名规则设定为 user_a 的 id 加上下划线 _ 再加 user_b 的 id，其中 id 值较小的用户 ID 排在前面，例如：195752_748418。


class ChatConsumer(AsyncJsonWebsocketConsumer):
  chats = dict()
  async def connect(self):
    self.group_name = self.scope['url_route']['kwargs']['group_name']
    await self.channel_layer.group_add(self.group_name, self.channel_name)
    # 将用户添加至聊天组信息chats中
    try:
      ChatConsumer.chats[self.group_name].add(self)
    except:
      ChatConsumer.chats[self.group_name] = set([self])
    #print(ChatConsumer.chats)
    # 创建连接时调用
    await self.accept()
  async def disconnect(self, close_code):
    # 连接关闭时调用
    # 将关闭的连接从群组中移除
    await self.channel_layer.group_discard(self.group_name, self.channel_name)
    # 将该客户端移除聊天组连接信息
    ChatConsumer.chats[self.group_name].remove(self)
    await self.close()

ChatConsumer 中的 chats 字典用于记录每个 group 内的连接数。当客户端访问正确的 WebSocket URL 时，将调用 connect() 函数，把该客户端加入其 URL 指向的 group，并在 chats 中记录该客户端信息。当客户端断开连接时，调用 disconnect() 函数，将其从 group 中移除，并删除 chats 中的对应记录。

完成连接与断开逻辑后，接下来处理信息接收。


 async def receive_json(self, message, **kwargs):
    # 收到信息时调用
    to_user = message.get('to_user')
    # 信息发送
    length = len(ChatConsumer.chats[self.group_name])
    if length == 2:
      await self.channel_layer.group_send(
        self.group_name,
        {
          "type": "chat.message",
          "message": message.get('message'),
        },
      )
    else:
      await self.channel_layer.group_send(
        to_user,
        {
          "type": "push.message",
          "event": {'message': message.get('message'), 'group': self.group_name}
        },
      )
  async def chat_message(self, event):
    # Handles the "chat.message" event when it's sent to us.
    await self.send_json({
      "message": event["message"],
    })

在上述函数中，接收到客户端 WebSocket 信息后，首先判断该聊天组内的客户端连接数量。若数量为 2，说明双方均已连接至该聊天组，可直接向该 group 发送信息，对方即可直接接收；若数量为 1，说明接收方尚未进入聊天组，此时向其推送一条包含 group_name 和信息内容的消息。

至此，点对点聊天系统的基本功能已实现。

消息推送功能

三：消息推送功能

消息推送的工作原理与聊天类似，即每个用户拥有独立的 WebSocket 连接。此处可使用用户的 username 作为 group_name。当其他用户的某些行为触发推送条件时，后台向该用户所在的 group 发送信息，从而完成消息推送服务。

特别说明：Channels 同样支持单通道发送，即每个客户端连接时生成专用通道名称。但此处全部使用 group 发送，原因在于个人偏好，且通过客户端连接时的用户认证，可确保每个用户仅能连接至自己的 group。此处仅展示简单消息推送的实现，不涉及其他复杂代码。


# 推送consumer
class PushConsumer(AsyncWebsocketConsumer):
  async def connect(self):
    self.group_name = self.scope['url_route']['kwargs']['username']
    await self.channel_layer.group_add(
      self.group_name,
      self.channel_name
    )
    await self.accept()
  async def disconnect(self, close_code):
    await self.channel_layer.group_discard(
      self.group_name,
      self.channel_name
    )
    # print(PushConsumer.chats)
  async def push_message(self, event):
    print(event)
    await self.send(text_data=json.dumps({
      "event": event['event']
    }))

由于消息推送是后台向客户端推送信息，无需处理来自客户端的信息接收操作。因此，只需改写 connect() 和 disconnect() 函数，并添加一个处理发送信息的 push_message() 函数。

接着编写 push() 函数，供项目其他部分调用。这也是第一步中使用 Redis 作为 Channels 通道后端的原因。


from channels.layers import get_channel_layer

def push(username, event):
  channel_layer = get_channel_layer()
  async_to_sync(channel_layer.group_send)(
    username,
    {
      "type": "push.message",
      "event": event
    }
  )

该函数定义在 PushConsumer 类之外，因为在项目其他部分调用时，不会使用 self.channel_layer 获取通道层，故单独编写为函数，并通过 get_channel_layer 检索。

因此，在需要消息推送的地方，直接调用 push() 函数，传入被推送用户的用户名及推送信息即可。

Routing 配置及其他配置

四：Routing 配置及其他配置

同样，在 channel 文件夹下新建 routing.py 文件，并添加以下内容。其工作原理与 Django 的 urls.py 一致，定义了 WebSocket 的连接路径。


from . import consumers

websocket_urlpatterns = [
  url(r'^ws/chat/(?P<group_name>[^/]+)/$', consumers.ChatConsumer),
  url(r'^push/(?P<username>[0-9a-z]+)/$', consumers.PushConsumer),
]

然后在 settings.py 同级目录下新建 routing.py 文件，添加以下代码。


from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
import example.routing

application = ProtocolTypeRouter({
  # (http->django views is added by default)
  'websocket': AuthMiddlewareStack(
    URLRouter(
      example.routing.websocket_urlpatterns
    )
  ),
})

如此，客户端即可成功连接至 WebSocket，功能得以简单实现。