Django使用Channels实现WebSocket消息通知功能

Channels 用于websocket通信，websocket是啥？主要实现在无客户端请求的情况下向客户端发送数据。

主要概念：频道，websocket是面向频道的，就好比微信群，一个微信群就是一个频道，你向频道发送内容，微信群内的所有人(连接到这个频道的客户端)都可以收到你的消息

本文主要写快速实现ws服务端

需要django3.0+，最好是在linux下进行测试开发，windows下安装时问题一大堆

• 安装：

  pip install django>=3.0
  pip install channels
  

• 创建project

  django-admin startproject TestVideos
  

• 创建app

  django-admin startapp chat
  

• 创建chat/routing.py

  #django的路由叫urls.py
  #对于channels有新的路由文件
  from django.urls import re_path
  
  from . import consumers  #等同于views.py 稍后创建
  
  websocket_urlpatterns = [
      re_path(r‘ws/chat/(?P<room_name>\w+)/$‘, consumers.ChatConsumer.as_asgi()),
  ]
  

• 创建 chat/consumers.py

  #等同于django的views.py
  #对于channels叫consumers.py
  #下面的内容作用：将新的ws客户端加入到一个频道，将其发送到ws服务端的内容广播至频道所有人
  class ChatConsumer(AsyncWebsocketConsumer):
      async def connect(self):
          self.room_name = self.scope[‘url_route‘][‘kwargs‘][‘room_name‘]
          self.room_group_name = ‘chat_%s‘ % self.room_name
  
          # Join room group
          await self.channel_layer.group_add(
              self.room_group_name,
              self.channel_name
          )
  
          await self.accept()
  
      async def disconnect(self, close_code):
          # Leave room group
          await self.channel_layer.group_discard(
              self.room_group_name,
              self.channel_name
          )
  
      # Receive message from WebSocket
      async def receive(self, text_data):
          message = dict(json.loads(text_data))
  
          # Send message to room group
          await self.channel_layer.group_send(
              self.room_group_name,
              {
                  ‘type‘: ‘chat_message‘,
                  ‘message‘: message
              }
          )
  
      # Receive message from room group
      async def chat_message(self, event):
          message = event[‘message‘]
  
          # Send message to WebSocket
          await self.send(text_data=json.dumps({
              ‘message‘: message
          }))
  
  

• 修改settings.py

  INSTALLED_APPS = [
      ‘django.contrib.admin‘,
      ‘django.contrib.auth‘,
      ‘django.contrib.contenttypes‘,
      ‘django.contrib.sessions‘,
      ‘django.contrib.messages‘,
      ‘django.contrib.staticfiles‘,
      ...,
      ‘channels‘,   #ws服务端
      ‘chat‘,       #ws   app
  ]
  
  
  #channels
  ASGI_APPLICATION = "TestVideos.asgi.application"    #入口信息
  CHANNEL_LAYERS = {    #频道后端，这里采用内存存储，默认是redis
      "default": {
          "BACKEND": "channels.layers.InMemoryChannelLayer"
      }
  }
  

• 添加入口：

  配置project/asgi.py(django3.0以前仅wsgi.py。和settings.py同级)

  asgi.py原内容

  """
  ASGI config for TestVideos project.
  
  It exposes the ASGI callable as a module-level variable named ``application``.
  
  For more information on this file, see
  https://docs.djangoproject.com/en/3.1/howto/deployment/asgi/
  """
  
  import os
  
  from django.core.asgi import get_asgi_application
  
  os.environ.setdefault(‘DJANGO_SETTINGS_MODULE‘, ‘TestVideos.settings‘)
  
  application = get_asgi_application()
  
  

  asgi.py新内容

  """
  ASGI config for main_project project.
  
  It exposes the ASGI callable as a module-level variable named ``application``.
  
  For more information on this file, see
  https://docs.djangoproject.com/en/3.1/howto/deployment/asgi/
  """
  
  import os
  
  from django.core.asgi import get_asgi_application
  
  os.environ.setdefault(‘DJANGO_SETTINGS_MODULE‘, ‘BoatControlProject.settings‘)
  
  # application = get_asgi_application()  #原本内容
  
  
  from channels.routing import ProtocolTypeRouter, URLRouter  #channels
  from channels.auth import AuthMiddlewareStack
  import chat.routing    #我们创建的app
  
  
  application = ProtocolTypeRouter({
      "http": get_asgi_application(),
  
      "websocket": AuthMiddlewareStack(
          URLRouter(
              chat.routing.websocket_urlpatterns
          )
      ),
  })
  
  
  

• 如何访问我们的ws，首先准备一个ws测试工具（用于向频道发送内容）。

  使用这个：https://github.com/easy-swoole/wstool
  

• 在浏览器打开两个测试ws测试工具，都连接至同一个地址，client 1 发送数据， client 2 接收数据

  

  

其他的一些功能

• 当你有http请求或者其他地方需要向ws客户端发送内容时的写法：

  from asgiref.sync import async_to_sync
  from channels.layers import get_channel_layer
  
  @method_decorator(csrf_exempt, name="dispatch")
  class SendRoom(View):
      def dispatch(self, request, *args, **kwargs):
  
          return super(SendRoom, self).dispatch(request, *args, **kwargs)
  
      #测试频道广播
      def get(self, request, *args, **kwargs):
          event = {}
          event["message"] = {"latitudenum":37.553534,"longitudenum":122.083534,"yaw":90}
  
          channel_layer = get_channel_layer()     #重要代码
          async_to_sync(channel_layer.group_send)("chat_videos_stram", {"type": "chat_message","message": event["message"]})   #重要代码
  
          message = "ws send ok"
          return JsonResponse({"message": message})
  
      def post(self, request, *args, **kwargs):
          message = "post ok"
  
          return JsonResponse({"message": message})
  

Django使用Channels实现WebSocket数据推送功能

原文：https://www.cnblogs.com/lisicn/p/14596411.html