Django Channels 实现点对点实时聊天和消息推送功能

简介在很多实际的项目开发中，我们需要实现很多实时功能；而在这篇文章中，我们就利用django channels简单地实现了点对点聊天和消息推送功能。

手边有一个项目需要用到后台消息推送和用户之间一对一在线聊天的功能。例如用户A评论了用户B的帖子，这时候用户B就应该收到一条通知，显示自己的帖子被评论了。这个功能可以由最基本的刷新页面后访问数据库来完成，但是这样会增加对后台服务器的压力，同时如果是手机客户端的话，也会造成流量的损失。于是，我们考虑使用websocket建立一个连接来完成这个功能。

但是django并不支持websocket，因此在一番寻找之后发现了django-channels这个项目，它允许Django项目不仅可以处理HTTP，还可以处理需要长时间连接的协议 - WebSockets，MQTT，chatbots，业余无线电等等。

作者本人也接触channels没多久，为了搞这两个功能看channels文档看到自闭，最终简单实现了这两个功能，特地记录一下

一：安装channels

如果使用的是django 1.9 及以上，在pip安装channels时可以不加-U参数

pip install channels

安装结束后，我们把channels作为一个app添加进入我们的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'

二：点对点聊天

在项目目录下新建一个文件，用来存放我们的channels代码，为channel。在channel中新建一个comsumers.py文件，在其中新建一个ChatComsumer类用来处理我们聊天时的websocket请求。相对于建立一个聊天室，在这里不同的是我们在ChatComsumer中添加了一个chats来记录每一个group中的连接数。以此根据这个连接数来判断，聊天双方是否都已连接进入该个聊天group。

同时，我们设定聊天组的命名形式为user_a的id加上下划线_加上user_b的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()

ChatComsumer中的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便可以完成相应的功能，只要在客户端连接时添加用户认证，便能保证每个用户只能连接上自己的那个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
   }
 )

这个函数写在PushComsumer之外，因为我们在项目的其他地方调用时，不会使用self.self.channel_layer来获取通道层，因此单独写做一个函数，然后使用get_channel_layer来检索它。

因此，在我们需要使用消息推送的地方，只要直接调用push()函数，传入被推送用户的用户名和推送的信息就OK了。

四：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了，功能简单实现。

总结

以上所述是小编给大家介绍的Django Channels 实现点对点实时聊天和消息推送功能,网站的支持！
如果你觉得本文对你有帮助，欢迎转载，烦请注明出处，谢谢！

来源：https://blog.thinker.ink/passage/14/