Django Channels Redis 使用教程

42 0 0

Django Channels Redis 使用教程

channels_redisRedis channel layer backend for Django Channels项目地址:https://gitcode.com/gh_mirrors/ch/channels_redis

项目介绍

Django Channels Redis 是一个用于 Django Channels 的 Redis 后端，它允许 Django 项目处理 WebSocket、长轮询和其他异步请求。Django Channels 扩展了 Django 的能力，使其能够处理传统的 HTTP 请求以及 WebSocket 等异步协议。Redis 作为一个高性能的键值存储系统，被用作 Channels 的通道层，以支持实时通信和消息传递。

项目快速启动

安装依赖

首先，确保你已经安装了 Django 和 Django Channels。然后，通过 pip 安装 Django Channels Redis：

pip install channels_redis

配置 Django 项目

在你的 Django 项目的 settings.py 文件中，添加以下配置：




INSTALLED_APPS = [


    ...


    'channels',


]


 


ASGI_APPLICATION = 'your_project_name.asgi.application'


 


CHANNEL_LAYERS = {


    'default': {


        'BACKEND': 'channels_redis.core.RedisChannelLayer',


        'CONFIG': {


            "hosts": [('127.0.0.1', 6379)],


        },


    },


}

创建 ASGI 文件

在项目的根目录下创建一个 asgi.py 文件，内容如下：




import os


from django.core.asgi import get_asgi_application


from channels.routing import ProtocolTypeRouter, URLRouter


from channels.auth import AuthMiddlewareStack


import your_app.routing


 


os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project_name.settings')


 


application = ProtocolTypeRouter({


    "http": get_asgi_application(),


    "websocket": AuthMiddlewareStack(


        URLRouter(


            your_app.routing.websocket_urlpatterns


        )


    ),


})

配置路由

在你的应用目录下创建一个 routing.py 文件，内容如下：




from django.urls import re_path


from . import consumers


 


websocket_urlpatterns = [


    re_path(r'ws/some_path/$', consumers.SomeConsumer.as_asgi()),


]

创建消费者

在你的应用目录下创建一个 consumers.py 文件，内容如下：




import json


from channels.generic.websocket import AsyncWebsocketConsumer


 


class SomeConsumer(AsyncWebsocketConsumer):


    async def connect(self):


        await self.channel_layer.group_add(


            "some_group",


            self.channel_name


        )


        await self.accept()


 


    async def disconnect(self, close_code):


        await self.channel_layer.group_discard(


            "some_group",


            self.channel_name


        )


 


    async def receive(self, text_data):


        text_data_json = json.loads(text_data)


        message = text_data_json['message']


 


        await self.channel_layer.group_send(


            "some_group",


            {


                'type': 'chat_message',


                'message': message


            }


        )


 


    async def chat_message(self, event):


        message = event['message']


 


        await self.send(text_data=json.dumps({


            'message': message


        }))