django-channels(websocket)

官文:

https://channels.readthedocs.io/en/latest/tutorial/part_3.html

安装:

pip install channels, channels-redis

配置

# settings.py中添加INSTALLED_APPS = [...,'channels',
]
ASGI_APPLICATION = '项目名.asgi.application'注:推荐使用版本大于django3.0(配置)
asgi.py中添加import os
from channels.routing import ProtocolTypeRouter
from django.core.asgi import get_asgi_applicationos.environ.setdefault('DJANGO_SETTINGS_MODULE', '项目名.settings')application = ProtocolTypeRouter({"http": get_asgi_application(),# Just HTTP for now. (We can add other protocols later.)
})django2.2配置(因为没asgi文件,自建一个asgi.py文件)import os
import django
from channels.http import AsgiHandler
from channels.routing import ProtocolTypeRouteros.environ.setdefault('DJANGO_SETTINGS_MODULE', '项目名.settings')
django.setup()application = ProtocolTypeRouter({"http": AsgiHandler(),# Just HTTP for now. (We can add other protocols later.)
})

启动服务,展示下面日志便启动成功(ASGI)

此时已经配置成功

单向通讯(实操)

单向通讯不用redis即可实现

在自己项目下新建consumers.py

# front_end/consumer.pyimport json
from channels.generic.websocket import WebsocketConsumerclass ChatConsumer(WebsocketConsumer):def connect(self):self.accept()def disconnect(self, close_code):passdef receive(self, text_data):text_data_json = json.loads(text_data)message = text_data_json['message']self.send(text_data=json.dumps({'message': message}))

在自己项目下新建routing.py

from django.urls import path
from front_end import consumerswebsocket_urlpatterns = [path(r'ws/chat/', consumers.ChatConsumer.as_asgi()),
]

asgi.py中注册url

2.2版本
import osimport django
from channels.http import AsgiHandler
from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.security.websocket import AllowedHostsOriginValidatorfrom front_end import routingos.environ.setdefault('DJANGO_SETTINGS_MODULE', '项目名.settings')
django.setup()application = ProtocolTypeRouter({"http": AsgiHandler(),# Just HTTP for now. (We can add other protocols later.)"websocket": AllowedHostsOriginValidator(    # 新增部分AuthMiddlewareStack(                         URLRouter(routing.websocket_urlpatterns))),
})3.0及以上
# 项目名/asgi.py
import osfrom channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.security.websocket import AllowedHostsOriginValidator
from django.core.asgi import get_asgi_applicationos.environ.setdefault("DJANGO_SETTINGS_MODULE", "项目名.settings")
django_asgi_app = get_asgi_application()import chat.routingapplication = ProtocolTypeRouter({"http": django_asgi_app,"websocket": AllowedHostsOriginValidator(AuthMiddlewareStack(URLRouter(routing.websocket_urlpatterns))),
})

websocket在线测试
http://www.websocket-test.com/
http://www.jsons.cn/websocket/【简洁】

ws://127.0.0.1:8000/ws/chat/
发送: {"message": "测试发送一下"}

双向通道(多人聊天配置)

settings.py中配置redis

CHANNEL_LAYERS = {'default': {'BACKEND': 'channels_redis.core.RedisChannelLayer','CONFIG': {"hosts": [('127.0.0.1', 6379)]， # 不带密码注: 不带密码在未关掉redis安全保护的情况不能访问公网redis(redis.conf注释掉bind 127.0.0.1, 找到 protected-mode yes 把 yes 改为 no)#  "hosts": ["redis://:密码@IP:端口/1"],  带密码配置},},
}

归档

安装

pip3 install channels

快速上手

在settings中添加配置

  INSTALLED_APPS = ['django.contrib.admin','django.contrib.auth','django.contrib.contenttypes','django.contrib.sessions','django.contrib.messages','django.contrib.staticfiles','channels',]ASGI_APPLICATION = '项目名.asgi.application'

创建websocket应用和路由

  #!/usr/bin/env python# -*- coding:utf-8 -*-from channels.routing import ProtocolTypeRouter, URLRouterfrom django.conf.urls import urlfrom chat import consumersapplication = ProtocolTypeRouter({'websocket': URLRouter([url(r'^chat/$', consumers.ChatConsumer),])})

编写处理websocket逻辑业务

案例一

  #!/usr/bin/env python# -*- coding:utf-8 -*-from channels.generic.websocket import WebsocketConsumerfrom channels.exceptions import StopConsumerclass ChatConsumer(WebsocketConsumer):def websocket_connect(self, message):self.accept()def websocket_receive(self, message):print('接收到消息', message)self.send(text_data='收到了')def websocket_disconnect(self, message):print('客户端断开连接了')raise StopConsumer()

案例二

  #!/usr/bin/env python# -*- coding:utf-8 -*-from channels.generic.websocket import WebsocketConsumerfrom channels.exceptions import StopConsumerclass SimpleChatConsumer(WebsocketConsumer):def connect(self):self.accept()def receive(self, text_data=None, bytes_data=None):self.send(text_data)# 主动断开连接# self.close()def disconnect(self, code):print('客户端要断开了')

案例三

  #!/usr/bin/env python# -*- coding:utf-8 -*-from channels.generic.websocket import WebsocketConsumerfrom channels.exceptions import StopConsumerCLIENTS = []class ChatConsumer(WebsocketConsumer):def connect(self):self.accept()CLIENTS.append(self)def receive(self, text_data=None, bytes_data=None):for item in CLIENTS:item.send(text_data)# 主动断开连接# self.close()def disconnect(self, code):CLIENTS.remove(self)

channel layer

基于内存的channel layer

CHANNEL_LAYERS = {"default": {"BACKEND": "channels.layers.InMemoryChannelLayer",}
}

业务处理

from channels.generic.websocket import WebsocketConsumer
from asgiref.sync import async_to_syncclass ChatConsumer(WebsocketConsumer):def connect(self):async_to_sync(self.channel_layer.group_add)('x1', self.channel_name)self.accept()def receive(self, text_data=None, bytes_data=None):async_to_sync(self.channel_layer.group_send)('x1', {'type': 'xxx.ooo','message': text_data})def xxx_ooo(self, event):message = event['message']self.send(message)def disconnect(self, code):async_to_sync(self.channel_layer.group_discard)('x1', self.channel_name)

基于 redis的channel layer

安装: pip3 install channels-redis

配置

CHANNEL_LAYERS = {"default": {"BACKEND": "channels_redis.core.RedisChannelLayer","CONFIG": {"hosts": [('10.211.55.25', 6379)]},
},
}CHANNEL_LAYERS = {'default': {'BACKEND': 'channels_redis.core.RedisChannelLayer','CONFIG': {"hosts": ["redis://10.211.55.25:6379/1"],},},
}CHANNEL_LAYERS = {'default': {'BACKEND': 'channels_redis.core.RedisChannelLayer','CONFIG': {"hosts": [('10.211.55.25', 6379)],},},
}CHANNEL_LAYERS = {"default": {"BACKEND": "channels_redis.core.RedisChannelLayer","CONFIG": {"hosts": ["redis://:password@10.211.55.25:6379/0"],"symmetric_encryption_keys": [SECRET_KEY],},},
}

业务逻辑

from channels.generic.websocket import WebsocketConsumer
from asgiref.sync import async_to_syncclass ChatConsumer(WebsocketConsumer):def connect(self):async_to_sync(self.channel_layer.group_add)('x1', self.channel_name)self.accept()def receive(self, text_data=None, bytes_data=None):async_to_sync(self.channel_layer.group_send)('x1', {'type': 'xxx.ooo','message': text_data})def xxx_ooo(self, event):message = event['message']self.send(message)def disconnect(self, code):async_to_sync(self.channel_layer.group_discard)('x1', self.channel_name)

异步

import json
from channels.generic.websocket import WebsocketConsumer, AsyncWebsocketConsumer
from asgiref.sync import async_to_sync
from loguru import loggerclass ChatConsumer(AsyncWebsocketConsumer):def get_room_id(self):'''获取组的id'''return self.scope['url_route']['kwargs']['group']async def connect(self):await self.channel_layer.group_add(str(self.get_room_id()), self.channel_name)await self.accept()logger.warning(f'组id:【{self.get_room_id()}】,通道名:【{self.channel_name}】连接成功')logger.success(f'当前全部组:{self.channel_layer}')async def receive(self, text_data=None, bytes_data=None):await self.channel_layer.group_send(str(self.get_room_id()), {'type': 'xxx.ooo','message': text_data})async def xxx_ooo(self, event):print(event)message = event['message']await self.send(message)async def disconnect(self, code):print(code)await self.channel_layer.group_discard(str(self.get_room_id()), self.channel_name)await self.close()

django-channels(websocket)相关推荐

Python+Django+channels实现websocket
Python+Django+channels实现websocket 前言公司需要实现一个长连接,用的Python的Django框架.研究了很长时间,发现Django+channels可以实现webs ...
Django Channels 入门指南
http://www.oschina.NET/translate/in_deep_with_django_channels_the_future_of_real_time_apps_in_django ...
django channels
一直都是用HTTP请求糊里糊涂的实现了一次请求,一次响应.最近尝试用Django这种框架实现websocket,用的是Django channels,结合官网给的例子,实现了日志动态展示到页面.源码地 ...
Django channels摄像头实时视频传输
Django channels摄像头实时视频传输(视屏能传别的当然也能传拉) 前言不想看我瞎扯可以直接跳到这服务端步骤解释发送端接收端运行前言 (网上绝大多数博客都是发送端或者接收端同 ...
Django使用WebSocket
使用WebSocket可以实现实时通信,而不用频繁刷新页面才能实现数据更新. 配置 websocket的第三方模块首先,安装channels,注意版本: pip install channels== ...
django channels
django channels 是django支持websocket的一个模块. 1. 安装 1 pip3 install channels 2. 快速上手 2.1 在settings中添加配置 IN ...
Django Channels配置
channels 4.0之后默认不带Daphne服务器了.解决方案可以有两种: 1.指定channels的版本为3.x: 2.安装时使用pip3 install -U channels["d ...
Django Channels 原理
Django Channels 是一个为 Django 提供异步扩展的库,通常主要用来提供 WebSocket 支持和后台任务. 原理它的原理是将 Django 分为 2 种进程类型: 一个用于处理 ...
基于Django channels 与 YOLO v8 搭建＜实时跟踪与统计系统＞
文章大纲 0. 简介系统Demo 前序文章 1. 系统架构:基于分层结构.组件解耦的<实时跟踪与统计系统> 2. YOLOv8 3. 目标跟踪与计数:SORT ( Simple Onli ...
Python+Django+Channels之Consumers（用户）
Consumers 因为建立Channels最底层的解释接口是ASGI,被用来连接复杂应用的操作接口当然你也可以忽略consumers而使用其他Channels部分,像routing,session ...

django-channels(websocket)

官文:

安装:

配置

单向通讯(实操)

双向通道(多人聊天配置)

归档

django-channels(websocket)相关推荐

最新文章

热门文章