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

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 接收数据

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

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

其他的一些功能

  • 当你有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数据推送功能

上一篇:linux ssh公钥免密码登录


下一篇:CSS之字体属性