先來看一下最終的效果吧
開始聊天,輸入消息並點擊發送消息就可以開始聊天了
點擊 “獲取後端數據”開啟實時推送
先來簡單瞭解一下 Django Channel
Channels是一個採用Django並將其功能擴展到HTTP以外的項目,以處理WebSocket,聊天協議,IoT協議等。它基於稱為ASGI的Python規範構建。
它以Django的核心為基礎,並在其下面分層了一個完全異步的層,以同步模式運行Django本身,但異步處理了連接和套接字,並提供了以兩種方式編寫的選擇,從而實現了這一點。
詳情請參考官方文檔:https://channels.readthedocs.io/en/latest/introduction.html
再簡單說下ASGI是什麼吧
ASGI 由 Django 團隊提出,為了解決在一個網絡框架裡(如 Django)同時處理 HTTP、HTTP2、WebSocket 協議。為此,Django 團隊開發了 Django Channels 插件,為 Django 帶來了 ASGI 能力。
在 ASGI 中,將一個網絡請求劃分成三個處理層面,最前面的一層,interface server(協議處理服務器),負責對請求協議進行解析,並將不同的協議分發到不同的 Channel(頻道);頻道屬於第二層,通常可以是一個隊列系統。頻道綁定了第三層的 Consumer(消費者)。
下面來說一下具體的實現步驟
一、安裝channel
<code>pip3 install channels pip3 install channels_redis/<code>
二、新建Django項目
1.新建項目
<code>django-admin startproject mysite/<code>
2.新建應用
<code>python3 manage.py startapp chat/<code>
3.編輯 mysite/settings.py文件
<code>#註冊應用 INSTALLED_APPS = [ .... 'chat.apps.ChatConfig', "channels", ] # 在文件尾部新增如下配置 #將ASGI_APPLICATION設置設置為指向該路由對象作為您的根應用程序: ASGI_APPLICATION = 'mysite.routing.application' #配置Redis CHANNEL_LAYERS = { 'default': { 'BACKEND': 'channels_redis.core.RedisChannelLayer', 'CONFIG': { "hosts": [('10.0.6.29', 6379)], }, }, }/<code>
三、詳細代碼與配置
1. 添加索引視圖的模板
在chat目錄中創建一個templates目錄。在您剛剛創建的templates 目錄中,創建另一個名為的目錄 chat,並在其中創建一個名為的文件index.html以保存索引視圖的模板
將以下代碼放入chat/templates/chat/index.html
<code> Chat Rooms What chat room would you like to enter?
/<code>
2.創建聊天與消息推送模板
chat/templates/chat/room.html
<code> Chat Room
{{ room_name|json_script:"room-name" }} /<code>
3.創建房間的視圖
將以下代碼放入chat/views.py
<code># chat/views.py/<code>
<code>from django.shortcuts import render from django.http import JsonResponse from channels.layers import get_channel_layer from asgiref.sync import async_to_sync def index(request): return render(request, "chat/index.html") def room(request, room_name): return render(request, "chat/room.html", {"room_name": room_name}) def pushRedis(request): room = request.GET.get("room") print(room) def push(msg): channel_layer = get_channel_layer() async_to_sync(channel_layer.group_send)( room, {"type": "push.message", "message": msg, "room_name": room} ) push("推送測試", ) return JsonResponse({"1": 1})/<code>
4. 創建項目二級路由
在chat目錄下創建一個名為的文件urls.py
<code># mysite/chat/urls.py from django.urls import path from . import views urlpatterns = [ path('', views.index, name='index'), path('/', views.room, name='room'),/<code>
<code>]/<code>
5. 修改根路由
<code># mysite/urls.py from django.contrib import admin from django.urls import path, include from chat.views import pushRedis urlpatterns = [ path('admin/', admin.site.urls), path("chat/", include("chat.urls")), path("push", pushRedis, name="push"), ]/<code>
6.創建一個消費者
文件chat/consumers.py
當Django接受HTTP請求時,它會查詢根URLconf來查找視圖函數,然後調用該視圖函數來處理該請求。同樣,當Channels接受WebSocket連接時,它會查詢根路由配置以查找使用者,然後在使用者上調用各種功能來處理來自連接的事件。
<code>import time import json from channels.generic.websocket import WebsocketConsumer, AsyncWebsocketConsumer from asgiref.sync import async_to_sync import redis pool = redis.ConnectionPool( host="10.0.6.29", port=6379, max_connections=10, decode_response=True, ) conn = redis.Redis(connection_pool=pool, decode_responses=True) 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 await self.channel_layer.group_add( self.room_group_name, self.channel_name, ) await self.accept() async def disconnect(self, close_code): print("close_code: ", close_code) await self.channel_layer.group_discard( self.room_group_name, self.channel_name ) async def receive(self, text_data=None, bytes_data=None): text_data_json = json.loads(text_data) message = text_data_json["message"] print("receive_message:", message) await self.channel_layer.group_send( self.room_group_name, { "type": "chat_message", "message": message } ) async def chat_message(self, event): receive_message = event["message"] response_message = "You message is :" + receive_message await self.send(text_data=json.dumps({ "message": response_message })) class PushMessage(WebsocketConsumer): def connect(self): self.room_group_name = self.scope["url_route"]["kwargs"]["room_name"] async_to_sync(self.channel_layer.group_add)( self.room_group_name, self.channel_name ) self.accept() def disconnect(self, code): async_to_sync(self.channel_layer.group_discard)( self.room_group_name, self.channel_name ) def push_message(self, event): """ 主動推送 :param event: :return: """ print(event, type(event)) while True: time.sleep(2) msg = time.strftime("%Y-%m-%d %H:%M:%S") + "--- room_name: %s" % event["room_name"] self.send(text_data=json.dumps( {"message": msg} ))/<code>
7.為項目添加websocket的路由配置
在chat目錄下創建一個名為的文件routing.py
<code># mysite/chat/routing.py from django.urls import re_path, path from . import consumers websocket_urlpatterns = [ re_path(r"ws/chat/(?P\w+)/$", consumers.ChatConsumer), path("ws/push/", consumers.PushMessage), ]/<code>
8.配置websocket根路由
與setting同級目錄新建ws根路由文件 routing.py
<code>from channels.routing import ProtocolTypeRouter, URLRouter from channels.auth import AuthMiddlewareStack import chat.routing application = ProtocolTypeRouter({ "websocket": AuthMiddlewareStack( URLRouter( chat.routing.websocket_urlpatterns ) ), })/<code>
9.最終的文件關係如下圖
10.啟動服務
<code>python3 manage.py runserver 10.0.6.2:80/<code>
注意看,這和django是不一樣的
還有另一種更穩健的啟動方式
和setting同級新增文件 asgi.py
<code>import os import django from channels.routing import get_default_application os.environ.setdefault("DJANGO_SETTINGS_MODULE", "mysite.settings") django.setup() application = get_default_application()/<code>
啟動方式為:
<code>daphne -b 10.0.6.2 -p 80 mysite.asgi:application/<code>
<code>daphne 在安裝channel時已經自動安裝好了/<code>
最後
小編是一名python開發工程師,這裡有我自己整理了一套最新的python系統學習教程,包括從基礎的python腳本到web開發、爬蟲、數據分析、數據可視化、機器學習等。想要這些資料的可以關注小編,並在後臺私信小編:“01”即可領取