Channels3.0.3

网上的教程是2.4.0，本文照着官方文档整合。

Installation — Channels 3.0.3 documentation
https://channels.readthedocs.io/en/stable/installation.html

版本：
Python 3.7
Django 3.2.2
Channels 3.0.3

python -m pip install -U channels

简单使用

新建项目，新建app，略。(本文新建的app名称为TestModel)

setting.py添加channels和ASGI_APPLICATION :

INSTALLED_APPS = ('django.contrib.auth','django.contrib.contenttypes','django.contrib.sessions','django.contrib.sites',...'channels',
)ASGI_APPLICATION = 'mysite.asgi.application'#mysite改为自己的项目名

asgi.py修改成：

import os
from channels.routing import ProtocolTypeRouter, URLRouter
from django.core.asgi import get_asgi_applicationfrom channels.auth import AuthMiddlewareStack
import TestModel.routingos.environ.setdefault('DJANGO_SETTINGS_MODULE', 'hat.settings')django_application = get_asgi_application()application = ProtocolTypeRouter({"http": get_asgi_application(),"websocket": AuthMiddlewareStack(URLRouter(TestModel.routing.websocket_urlpatterns)),
})

在新建的app创建TestModel/routing.py，跟urls.py一样用来跳转

from django.urls import re_path
from . import consumers
websocket_urlpatterns = [re_path(r'ws/chat/(?P<room_name>\w+)/$', consumers.ChatConsumer.as_asgi()),
]

app中新建TestModel/consumers.py，用来处理请求，.as_asgi()作用跟Django的as_view()一样，将类作为返回。

from channels.generic.websocket import WebsocketConsumer
class ChatConsumer(WebsocketConsumer):def connect(self):print("Connect")self.accept()def disconnect(self, close_code):passdef receive(self, text_data):print("收到消息:"+text_data)self.send(text_data=json.dumps({'message': text_data}))

运行服务器

python3 manage.py migrate
python3 manage.py runserver

使用测试工具或者浏览器打开控制台

在线websocket测试-在线工具-postjson
http://coolaf.com/tool/chattest

控制台依次输入（端口修改成自己的）：

ws = new WebSocket(‘ws://localhost:9300/ws/chat/test/’)
ws.onmessage = event => console.log(event.data)
ws.send(“ping”)

另一种写法

asgi.py

import os
from channels.routing import ProtocolTypeRouter, URLRouter
from django.core.asgi import get_asgi_application
from .websocket import websocket_applicationfrom channels.auth import AuthMiddlewareStack
import TestModel.routingos.environ.setdefault('DJANGO_SETTINGS_MODULE', 'hat.settings')django_application = get_asgi_application()async def application(scope, receive, send):if scope['type'] == 'http':# Let Django handle HTTP requestsawait django_application(scope, receive, send)elif scope['type'] == 'websocket':print("websocket")await websocket_application(scope, receive, send)else:raise NotImplementedError(f"Unknown scope type {scope['type']}")

同目录下新建websocket.py

async def websocket_application(scope, receive, send):while True:event = await receive()print(event['type'])if event['type'] == 'websocket.connect':await send({'type': 'websocket.accept'})if event['type'] == 'websocket.disconnect':breakif event['type'] == 'websocket.receive':if event['text'] == 'ping':await send({'type': 'websocket.send','text': 'pong!'})

多用户 Group

多个consumer同时连接的时候,需要通信就要使用到group

需要用到Synchronous
setting.py增加:

CHANNEL_LAYERS = {'default': {'BACKEND': 'channels_redis.core.RedisChannelLayer','CONFIG': {"hosts": [('127.0.0.1', 6379)],},},
}

上面这个需要redis pip install channels_redis
可以使用下面这个配置：

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

不要在生产环境使用此配置！

将原来的consumer.py改写为：

import json
from channels.generic.websocket import AsyncWebsocketConsumerclass 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 groupawait self.channel_layer.group_add(self.room_group_name,self.channel_name)await self.accept()async def disconnect(self, close_code):# Leave room groupawait self.channel_layer.group_discard(self.room_group_name,self.channel_name)# Receive message from WebSocketasync def receive(self, text_data):text_data_json = json.loads(text_data)message = text_data_json['message']# Send message to room groupawait self.channel_layer.group_send(self.room_group_name,{'type': 'chat_message','message': message})# Receive message from room groupasync def chat_message(self, event):message = event['message']# Send message to WebSocketawait self.send(text_data=json.dumps({'message': message}))

发送json格式数据：

{“message”: “hi”}

新开另一个网页，连接同一地址，可以收到别的用户发来的消息。

错误

错误:

Exception inside application: ERR unknown command ‘BZPOPMIN’

解决:升级redis5.0以上

Python Django使用websocket channels3.0.3相关推荐

基于python django框架的物联网平台即将开源，兼具IoT物联网平台和IBMS系统集成平台双重特性
IOTOS物联网中间件平台,前端使用antd.typescript,后端采用python django,rpc通信使用zeroc-ice,sdk驱动二开支持python.java.c++..net多种 ...
Python+django网页设计入门（20）：使用WebSocket创建多人在线聊天室
2019年3月8日-10日公众号送书活动:中小学生Python课应该学什么 ================ 前导课程: Python+django网页设计入门(19):创建新模型扩展自带用户表的字段 ...
Python+Django+channels实现websocket
Python+Django+channels实现websocket 前言公司需要实现一个长连接,用的Python的Django框架.研究了很长时间,发现Django+channels可以实现webs ...
【Azu Cloud云盘 V1.0】基于python django写的云盘全开源分享
Azu Cloud 欢迎使用Azu Cloud云盘 V1.0 新的改变如何使用注册教程左上角图标(登录后) 如何上传文件如何分享? 分享链接如何使用? 口令如何使用? 图像视频音频等如何在 ...
Django使用WebSocket
使用WebSocket可以实现实时通信,而不用频繁刷新页面才能实现数据更新. 配置 websocket的第三方模块首先,安装channels,注意版本: pip install channels== ...
python django面试题_Django面试问题
1. 对Django的认识? #1.Django是走大而全的方向,它最出名的是其全自动化的管理后台:只需要使用起ORM,做简单的对象定义,它就能自动生成数据库结构.以及全功能的管理后台. #2.Dja ...
python django web项目的构建步骤(一)
Django 一个开放源代码的Web应用框架,由Python写成.采用了MVC的软件设计模式,即模型M,视图V和控制器C.它最初是被开发来用于管理劳伦斯出版集团旗下的一些以新闻内容为主的网站的.并于2 ...
基于Python+Django实现药品管理系统
作者主页:编程指南针简介:Java领域优质创作者.CSDN博客专家 Java项目.简历模板.学习资料.面试题库.技术互助文末获取源码项目编号:BS-Python-001 技术架构:Python ...
Win10+Python+Django+Nginx+MySQL开发教程及实例（1）——开发环境搭建
Win10+Python+Django+Nginx+MySQL开发教程及实例 PaulTsao 本教程共有三篇内容: 第一篇:Win10+Python+Django+Nginx+MySQL 开发环境搭 ...

Python Django使用websocket channels3.0.3

Channels3.0.3

简单使用

另一种写法

多用户 Group

错误

Python Django使用websocket channels3.0.3相关推荐

最新文章

热门文章