python异步编程-channels使用,创建websocket服务

2024-08-22 04:52

本文主要是介绍python异步编程-channels使用,创建websocket服务,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

目录

  • channels介绍
  • 准备工作
    • 创建python虚拟环境
    • 安装channels
    • 安装django
    • 安装daphne
    • 创建django项目
    • 创建chat应用
  • 配置instance项目
  • 简单聊天室页面
    • 在chat应用中创建模板
      • base.html
      • room.html
    • 添加视图
    • 添加路由
      • 添加urls.py
      • 项目路由添加chat转发路由
      • 启动同步服务器
  • 搭建websocket服务
    • 创建channels路由
      • 新增协议使用者
      • 新增websocket路由
      • 创建asgi的application
      • 配置asgi应用
      • 修改项目的asgi文件
  • 启动通道层
    • settings中配置
  • daphne部署
  • 验证
  • 总结

channels介绍

channels是一个用来构建实时web应用的强大工具,比如聊天室、在线游戏,多人协作、实时通知等。今天简单介绍一下,用channels搭建一个简易的websocket服务。

准备工作

创建python虚拟环境

mkdir instance
cd install
python -m venv venv# 激活虚拟环境
source venv/bin/activate # linux
./venv/bin/activate.bat # window

安装channels

pip install channels

安装django

pip install django

安装daphne

这个稍后会用到,用来启动异步服务器使用

pip install daphne

创建django项目

django-admin startproject instance 

创建chat应用

django-admin startapp chat

配置instance项目

在instance/instance目录下,修改settings.py文件

建议最好用pycharm打开

# instance/instance/settings.py ...
INSTALLED_APPS = ['daphne', # 新增'django.contrib.admin','django.contrib.auth','django.contrib.contenttypes','django.contrib.sessions','django.contrib.messages','django.contrib.staticfiles','django.contrib.sites','django.contrib.sitemaps','chat', # 新增'channels', # 新增
]
...
TEMPLATES = [{'BACKEND': 'django.template.backends.django.DjangoTemplates','DIRS': [BASE_DIR / 'templates'],'APP_DIRS': True,'OPTIONS': { # 此处options可以不配置,与我们的功能无关'context_processors': ['django.template.context_processors.debug','django.template.context_processors.request','django.contrib.auth.context_processors.auth','django.contrib.messages.context_processors.messages',],},},
]
...

简单聊天室页面

在chat应用中创建模板

需要创建两个模板文件,其中room.html 继承base.html

base.html

主要定义了页面的基本结构,具体内容在子模板中填充
instance/chat/template/chat/base.html

{% load static %}
<!DOCTYPE html>
<html>
<head><meta charset=""utf-8><title>{% block title %}Education{% endblock %}</title><link href="{% static "css/base.css" %}" rel="stylesheet">
</head>
<body>
<div id="header"><a href="/" class="logo">Education</a><ul class="menu">{% if request.user.is_authenticated %}<li><a href="{% url "logout" %}">Sign out</a> </li>{% else %}<li><a href="{% url "login" %}">Sign in</a> </li>{% endif %}</ul>
</div>
<div id="content">{% block content %}{% endblock %}
</div><script src="https://ajax.googleapis.com/ajax/libs/jquery/3.4.1/jquery.min.js"></script>
<script>$(document).ready(function(){{% block domready %}{% endblock %}});
</script>
</body>
</html>

room.html

主要实现了一个输入框一个发送按钮,以及可以展示消息的div元素,页面简陋,能看即可

除了页面结构,还加了websocket链接的js代码

{% extends "chat/base.html" %}{% block title %} Chat room for "{{ course.title }}" {% endblock %}{% block content %}
<div id="chat"></div>
<div id="chat-input"><input id="chat-message-input" type="text"><input id="chat-message-submit" type="submit" value="Send">
</div>
{% endblock %}{% block domready %}var url = 'ws://' + window.location.host + '/ws/chat/room/' + '{{ course }}' + '/';console.log(url);var chatSocket = new WebSocket(url);console.log(chatSocket);chatSocket.onmessage = function(e) {var data = JSON.parse(e.data);var message = data.message;var $chat = $('#chat');$chat.append('<div class="message">' + message + '</div>');$chat.scrollTop($chat[0].scrollHeight);};chatSocket.onclose = function (e) {console.error('chat socket close unexpectedly');};var $input = $('#chat-message-input');var $submit = $('#chat-message-submit');$submit.click(function(){var message = $input.val();if(message) {chatSocket.send(JSON.stringify({'message': message}));$input.val('');$input.focus();}});$input.focus();$input.keyup(function (e){if(e.which === 13) {$submit.click();}});
{% endblock %}

添加视图

编辑chat目录下views.py

from django.shortcuts import renderdef course_chat_room(request, course_id):return render(request, 'chat/room.html', {'course': 1})

添加路由

添加urls.py

chat目录下新增urls.py

from django.urls import pathfrom chat import viewsapp_name = 'chat'urlpatterns = [path('room/<int:course_id>/', views.course_chat_room, name='course_chat_room'),
]

项目路由添加chat转发路由

编辑instance/instance/urls.py

urlpatterns = [path('admin/', admin.site.urls),path('chat/', include('chat.urls', namespace='chat')), # 新增...,
]

启动同步服务器

到这一步就可以启动wsgi服务器,查看页面

python manage.py runserver

启动后,在浏览器中打开网站http://127.0.0.1:8000/chat/room/1/ 可以看到页面
我这里有样式设置,所以可能你们电脑上看到的样式不一样,但是基本结构时一样的
在这里插入图片描述

搭建websocket服务

我们前面已经将channels 应用添加到settings中的INSTALLED_APPS中

所以可以直接配置channels

创建channels路由

新增协议使用者

channels中一个很重要的概念就是消费者,也叫使用者,本质上就是协议请求接收之后的处理者

chat目录下新建consumers.py

这里使用了异步实现,将注释掉的代码恢复,就是同步代码

import jsonfrom channels.generic.websocket import WebsocketConsumer, AsyncWebsocketConsumer
from asgiref.sync import async_to_sync
from django.utils import timezone# class ChatConsumer(WebsocketConsumer):
class ChatConsumer(AsyncWebsocketConsumer):async def connect(self):self.user = self.scope['user']self.id = self.scope['url_route']['kwargs']['course_id']self.room_group_name = 'chat_%s' % self.id# async_to_sync(self.channel_layer.group_add)(self.room_group_name, self.channel_name)await self.channel_layer.group_add(self.room_group_name, self.channel_name)# accept connectionawait self.accept()async def disconnect(self, code):# async_to_sync(self.channel_layer.group_dicard)(self.room_group_name, self.channel_name)await self.channel_layer.group_dicard(self.room_group_name, self.channel_name)# receive message from WebSocketasync def receive(self, text_data=None, bytes_data=None):text_data_json = json.loads(text_data)message = text_data_json["message"]# send message to WebSocket# self.send(text_data=json.dumps({'message': message}))now = timezone.now()# send message to room groupawait self.channel_layer.group_send(self.room_group_name, {"type": "chat_message","message": message,"user": self.user.username,"datetime": now.isoformat(),})async def chat_message(self, event):await self.send(text_data=json.dumps(event))

新增websocket路由

chat下新增routing.py

from django.urls import re_path
from chat import consumerswebsocket_urlpatterns = [re_path(r'ws/chat/room/(?P<course_id>\d+)/$', consumers.ChatConsumer.as_asgi()),
]

创建asgi的application

在instance/instance下新建routing.py

这里我们使用的ProtocolTypeRouter 协议路由,区分不同的协议使用不同的路由方式
http一定要配置,否则正常的http请求无法处理,会报错
其次就是新增了websocket协议
如此创建之后,才能应用websocket协议

import osfrom channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from django.core.asgi import get_asgi_applicationimport chat.routingos.environ.setdefault('DJANGO_SETTINGS_MODULE', 'instance.settings')application = ProtocolTypeRouter({'http': get_asgi_application(),'websocket': AuthMiddlewareStack(URLRouter(chat.routing.websocket_urlpatterns))
})

配置asgi应用

instance/instance/settings.py中新增配置项

...
ASGI_APPLICATION = 'instance.routing.application'
...

修改项目的asgi文件

修改instance/instance/asgi.py

import os
import django
from django.core.asgi import get_asgi_application
from channels.routing import get_default_applicationos.environ.setdefault('DJANGO_SETTINGS_MODULE', 'instance.settings')
django.setup()
# application = get_asgi_application()
application = get_default_application()

OK,走到这一步,就可以在聊天室内发送消息了,只不过我们为了模拟而已,所以没有两个用户,只是把发送过来的消息再转发回client端,并显示在页面上

你是不是用了python manage.py runserver 启动,发现没有任何变化,然后打开F12 发现websocket链接失败

别急,往下看

启动通道层

通道层类似于传话筒,能够实现不同应用之间的通信,也可以支持使用者实例之间的通信

一般常用的通道层后台是redis
这里我们为了演示,就直接本地内存作为通道层缓存

settings中配置

配置下面的项,则自动启动通道层
不配置或者保留为空,则不启动

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

daphne部署

如果你使用了python manage.py runserver 发现没有效果,可以尝试daphne部署

django项目的测试服务器,默认启动的都是wsgi程序,不支持异步处理

而websocket的应用一般都是异步(同步的话就没有了意义)

所以我们要使用能够部署异步服务器的工具

就是daphne

daphne 支持asgi应用,基于我们上面写好的代码和创建的项目结构,启动方式如下

daphne instance.asgi:application
# 或者
daphne instance.routing:application

此时应该就可以了!

验证

启动之后,可以浏览器中多打开几个页面,因为我们没有做鉴权,所以每个页面都会创建一个websocket链接,相当于多个client

在其中一个页面中发送消息,那么其他页面也会同步刷新

之后优化一下页面结构,那就是一个简易聊天室了
在这里插入图片描述
在这里插入图片描述

总结

channels是一个能够处理实时协议的强大工具,这个协议可以是http,websocket,也可以是其他的需要实时通讯的情况。

daphne命令可以启动验证,也可以直接使用在生产环境,是安全的

这篇关于python异步编程-channels使用,创建websocket服务的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!


原文地址:
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.chinasem.cn/article/1095298

相关文章

postgresql使用UUID函数的方法

《postgresql使用UUID函数的方法》本文给大家介绍postgresql使用UUID函数的方法,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考下吧... 目录PostgreSQL有两种生成uuid的方法。可以先通过sql查看是否已安装扩展函数,和可以安装的扩展函数

Python实现终端清屏的几种方式详解

《Python实现终端清屏的几种方式详解》在使用Python进行终端交互式编程时,我们经常需要清空当前终端屏幕的内容,本文为大家整理了几种常见的实现方法,有需要的小伙伴可以参考下... 目录方法一:使用 `os` 模块调用系统命令方法二:使用 `subprocess` 模块执行命令方法三:打印多个换行符模拟

Python实现MQTT通信的示例代码

《Python实现MQTT通信的示例代码》本文主要介绍了Python实现MQTT通信的示例代码,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一... 目录1. 安装paho-mqtt库‌2. 搭建MQTT代理服务器(Broker)‌‌3. pytho

如何使用Lombok进行spring 注入

《如何使用Lombok进行spring注入》本文介绍如何用Lombok简化Spring注入,推荐优先使用setter注入,通过注解自动生成getter/setter及构造器,减少冗余代码,提升开发效... Lombok为了开发环境简化代码,好处不用多说。spring 注入方式为2种,构造器注入和setter

基于Python开发一个图像水印批量添加工具

《基于Python开发一个图像水印批量添加工具》在当今数字化内容爆炸式增长的时代,图像版权保护已成为创作者和企业的核心需求,本方案将详细介绍一个基于PythonPIL库的工业级图像水印解决方案,有需要... 目录一、系统架构设计1.1 整体处理流程1.2 类结构设计(扩展版本)二、核心算法深入解析2.1 自

MySQL中比较运算符的具体使用

《MySQL中比较运算符的具体使用》本文介绍了SQL中常用的符号类型和非符号类型运算符,符号类型运算符包括等于(=)、安全等于(=)、不等于(/!=)、大小比较(,=,,=)等,感兴趣的可以了解一下... 目录符号类型运算符1. 等于运算符=2. 安全等于运算符<=>3. 不等于运算符<>或!=4. 小于运

使用zip4j实现Java中的ZIP文件加密压缩的操作方法

《使用zip4j实现Java中的ZIP文件加密压缩的操作方法》本文介绍如何通过Maven集成zip4j1.3.2库创建带密码保护的ZIP文件,涵盖依赖配置、代码示例及加密原理,确保数据安全性,感兴趣的... 目录1. zip4j库介绍和版本1.1 zip4j库概述1.2 zip4j的版本演变1.3 zip4

RabbitMQ消息总线方式刷新配置服务全过程

《RabbitMQ消息总线方式刷新配置服务全过程》SpringCloudBus通过消息总线与MQ实现微服务配置统一刷新,结合GitWebhooks自动触发更新,避免手动重启,提升效率与可靠性,适用于配... 目录前言介绍环境准备代码示例测试验证总结前言介绍在微服务架构中,为了更方便的向微服务实例广播消息,

从入门到进阶讲解Python自动化Playwright实战指南

《从入门到进阶讲解Python自动化Playwright实战指南》Playwright是针对Python语言的纯自动化工具,它可以通过单个API自动执行Chromium,Firefox和WebKit... 目录Playwright 简介核心优势安装步骤观点与案例结合Playwright 核心功能从零开始学习

Python 字典 (Dictionary)使用详解

《Python字典(Dictionary)使用详解》字典是python中最重要,最常用的数据结构之一,它提供了高效的键值对存储和查找能力,:本文主要介绍Python字典(Dictionary)... 目录字典1.基本特性2.创建字典3.访问元素4.修改字典5.删除元素6.字典遍历7.字典的高级特性默认字典