這是Django Channels系列文章的第二篇，以web端實(shí)現(xiàn)tailf的案例講解Channels的具體使用以及跟Celery的結(jié)合

通過(guò)上一篇 《Django使用Channels實(shí)現(xiàn)WebSocket--上篇》 的學(xué)習(xí)應(yīng)該對(duì)Channels的各種概念有了清晰的認(rèn)知，可以順利的將Channels框架集成到自己的Django項(xiàng)目中實(shí)現(xiàn)WebSocket了，本篇文章將以一個(gè)Channels+Celery實(shí)現(xiàn)web端tailf功能的例子更加深入的介紹Channels

先說(shuō)下我們要實(shí)現(xiàn)的目標(biāo)：所有登錄的用戶可以查看tailf日志頁(yè)面，在頁(yè)面上能夠選擇日志文件進(jìn)行監(jiān)聽(tīng)，多個(gè)頁(yè)面終端同時(shí)監(jiān)聽(tīng)任何日志都互不影響，頁(yè)面同時(shí)提供終止監(jiān)聽(tīng)的按鈕能夠終止前端的輸出以及后臺(tái)對(duì)日志文件的讀取

最終實(shí)現(xiàn)的結(jié)果見(jiàn)下圖

接著我們來(lái)看下具體的實(shí)現(xiàn)過(guò)程

技術(shù)實(shí)現(xiàn)

所有代碼均基于以下軟件版本：

• python==3.6.3
• django==2.2
• channels==2.1.7
• celery==4.3.0

celery4在windows下支持不完善，所以請(qǐng) 在linux下運(yùn)行 測(cè)試

日志數(shù)據(jù)定義

我們只希望用戶能夠查詢固定的幾個(gè)日志文件，就不是用數(shù)據(jù)庫(kù)僅借助settings.py文件里寫(xiě)全局變量來(lái)實(shí)現(xiàn)數(shù)據(jù)存儲(chǔ)

在settings.py里添加一個(gè)叫 TAILF 的變量，類(lèi)型為字典，key標(biāo)識(shí)文件的編號(hào)，value標(biāo)識(shí)文件的路徑

TAILF = {
 1: '/ops/coffee/error.log',
 2: '/ops/coffee/access.log',
}

基礎(chǔ)Web頁(yè)面搭建

假設(shè)你已經(jīng)創(chuàng)建好了一個(gè)叫tailf的app，并添加到了settings.py的INSTALLED_APPS中，app的目錄結(jié)構(gòu)大概如下

tailf
 - migrations
 - __init__.py
 - __init__.py
 - admin.py
 - apps.py
 - models.py
 - tests.py
 - views.py

依然先構(gòu)建一個(gè)標(biāo)準(zhǔn)的Django頁(yè)面，相關(guān)代碼如下

url:

from django.urls import path
from django.contrib.auth.views import LoginView,LogoutView
from tailf.views import tailf
urlpatterns = [
 path('tailf', tailf, name='tailf-url'),
 path('login', LoginView.as_view(template_name='login.html'), name='login-url'),
 path('logout', LogoutView.as_view(template_name='login.html'), name='logout-url'),
]

因?yàn)槲覀円?guī)定只有通過(guò)登錄的用戶才能查看日志，所以引入Django自帶的LoginView，logoutView幫助我們快速構(gòu)建Login，Logout功能

指定了登錄模板使用 login.html ，它就是一個(gè)標(biāo)準(zhǔn)的登錄頁(yè)面，post傳入username和password兩個(gè)參數(shù)即可，不貼代碼了

view:

from django.conf import settings
from django.shortcuts import render
from django.contrib.auth.decorators import login_required

# Create your views here.
@login_required(login_url='/login')
def tailf(request):
 logDict = settings.TAILF
 return render(request, 'tailf/index.html', {"logDict": logDict})

引入了 login_required 裝飾器，來(lái)判斷用戶是否登錄，未登錄就給跳到 /login 登錄頁(yè)面

logDict去setting里取我們定義好的 TAILF 字典賦值，并傳遞給前端

template:

{% extends "base.html" %}

{% block content %}
<div class="col-sm-8">
 <select class="form-control" id="file">
 <option value="">選擇要監(jiān)聽(tīng)的日志</option>
 {% for k,v in logDict.items %}
 <option value="{{ k }}">{{ v }}</option>
 {% endfor %}
 </select>
</div>
<div class="col-sm-2">
 <input class="btn btn-success btn-block" type="button" onclick="connect()" value="開(kāi)始監(jiān)聽(tīng)"/><br/>
</div>
<div class="col-sm-2">
 <input class="btn btn-warning btn-block" type="button" onclick="goclose()" value="終止監(jiān)聽(tīng)"/><br/>
</div>
<div class="col-sm-12">
 <textarea class="form-control" id="chat-log" disabled rows="20"></textarea>
</div>
{% endblock %}

前端拿到 TAILF 后通過(guò)循環(huán)的方式填充到select選擇框下，因?yàn)閿?shù)據(jù)是字典格式，使用 logDict.items 的方式可以循環(huán)出字典的key和value

這樣一個(gè)日志監(jiān)聽(tīng)頁(yè)面就完成了，但還無(wú)法實(shí)現(xiàn)日志的監(jiān)聽(tīng)，繼續(xù)往下

集成Channels實(shí)現(xiàn)WebSocket

日志監(jiān)聽(tīng)功能主要的設(shè)計(jì)思路就是頁(yè)面跟后端服務(wù)器建立websocket長(zhǎng)連接，后端通過(guò)celery異步執(zhí)行while循環(huán)不斷的讀取日志文件然后發(fā)送到websocket的channel里，實(shí)現(xiàn)頁(yè)面上的實(shí)時(shí)顯示

接著我們來(lái)集成channels

先添加routing路由，直接修改 webapp/routing.py

from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from django.urls import path, re_path
from chat.consumers import ChatConsumer
from tailf.consumers import TailfConsumer
application = ProtocolTypeRouter({
 'websocket': AuthMiddlewareStack(
 URLRouter([
 path('ws/chat/', ChatConsumer),
 re_path(r'^ws/tailf/(?P<id>\d+)/$', TailfConsumer),
 ])
 )
})

直接將路由信息寫(xiě)入到了 URLRouter 里，注意路由信息的外層多了一個(gè)list，區(qū)別于上一篇中介紹的寫(xiě)路由文件路徑的方式

頁(yè)面需要將監(jiān)聽(tīng)的日志文件傳遞給后端，我們使用routing正則 P<id>\d+ 傳文件ID給后端程序，后端程序拿到ID之后根據(jù)settings中指定的 TAILF 解析出日志路徑

routing的寫(xiě)法跟Django中的url寫(xiě)法完全一致，使用 re_path 匹配正則routing路由

添加consumer在 tailf/consumers.py 文件中

import json
from channels.generic.websocket import WebsocketConsumer
from tailf.tasks import tailf
class TailfConsumer(WebsocketConsumer):
 def connect(self):
 self.file_id = self.scope["url_route"]["kwargs"]["id"]
 self.result = tailf.delay(self.file_id, self.channel_name)
 print('connect:', self.channel_name, self.result.id)
 self.accept()
 def disconnect(self, close_code):
 # 中止執(zhí)行中的Task
 self.result.revoke(terminate=True)
 print('disconnect:', self.file_id, self.channel_name)
 def send_message(self, event):
 self.send(text_data=json.dumps({
 "message": event["message"]
 }))

這里使用Channels的單通道模式，每一個(gè)新連接都會(huì)啟用一個(gè)新的channel，彼此互不影響，可以隨意終止任何一個(gè)監(jiān)聽(tīng)日志的請(qǐng)求

connect

我們知道 self.scope 類(lèi)似于Django中的request，記錄了豐富的請(qǐng)求信息，通過(guò) self.scope["url_route"]["kwargs"]["id"] 取出routing中正則匹配的日志ID

然后將 id 和 channel_name 傳遞給celery的任務(wù)函數(shù)tailf，tailf根據(jù) id 取到日志文件的路徑，然后循環(huán)文件，將新內(nèi)容根據(jù) channel_name 寫(xiě)入對(duì)應(yīng)channel

disconnect

當(dāng)websocket連接斷開(kāi)的時(shí)候我們需要終止Celery的Task執(zhí)行，以清除celery的資源占用

終止Celery任務(wù)使用到 revoke 指令，采用如下代碼來(lái)實(shí)現(xiàn)

self.result.revoke(terminate=True)


注意 self.result 是一個(gè)result對(duì)象，而非id

參數(shù) terminate=True 的意思是是否立即終止Task，為T(mén)rue時(shí)無(wú)論Task是否正在執(zhí)行都立即終止，為False（默認(rèn)）時(shí)需要等待Task運(yùn)行結(jié)束之后才會(huì)終止，我們使用了While循環(huán)不設(shè)置為T(mén)rue就永遠(yuǎn)不會(huì)終止了

終止Celery任務(wù)的另外一種方法是：

from webapp.celery import app
app.control.revoke(result.id, terminate=True)
send_message

方便我們通過(guò)Django的view或者Celery的task調(diào)用給channel發(fā)送消息，官方也比較推薦這種方式

使用Celery異步循環(huán)讀取日志

上邊已經(jīng)集成了Channels實(shí)現(xiàn)了WebSocket，但connect函數(shù)中的celery任務(wù) tailf 還沒(méi)有實(shí)現(xiàn)，下邊來(lái)實(shí)現(xiàn)它

關(guān)于Celery的詳細(xì)內(nèi)容可以看這篇文章： 《Django配置Celery執(zhí)行異步任務(wù)和定時(shí)任務(wù)》 ，本文就不介紹集成使用以及細(xì)節(jié)原理，只講一下任務(wù)task

task實(shí)現(xiàn)代碼如下：

from __future__ import absolute_import
from celery import shared_task
import time
from channels.layers import get_channel_layer
from asgiref.sync import async_to_sync
from django.conf import settings
@shared_task
def tailf(id, channel_name):
 channel_layer = get_channel_layer()
 filename = settings.TAILF[int(id)]
 try:
 with open(filename) as f:
 f.seek(0, 2)
 while True:
 line = f.readline()
 if line:
  print(channel_name, line)
  async_to_sync(channel_layer.send)(
  channel_name,
  {
  "type": "send.message",
  "message": "微信公眾號(hào)【運(yùn)維咖啡吧】原創(chuàng) 版權(quán)所有 " + str(line)
  }
  )
 else:
  time.sleep(0.5)
 except Exception as e:
 print(e)

這里邊主要涉及到Channels中另一個(gè)非常重要的點(diǎn)： 從Channels的外部發(fā)送消息給Channel

其實(shí) 上篇文章 中檢查通道層是否能夠正常工作的時(shí)候使用的方法就是從外部給Channel通道發(fā)消息的示例，本文的具體代碼如下

async_to_sync(channel_layer.send)(
 channel_name,
 {
 "type": "send.message",
 "message": "微信公眾號(hào)【運(yùn)維咖啡吧】原創(chuàng) 版權(quán)所有 " + str(line)
 }
)

channel_name對(duì)應(yīng)于傳遞給這個(gè)任務(wù)的channel_name，發(fā)送消息給這個(gè)名字的channel

type對(duì)應(yīng)于我們Channels的TailfConsumer類(lèi)中的 send_message 方法，將方法中的 _ 換成 . 即可

message就是要發(fā)送給這個(gè)channel的具體信息

上邊是發(fā)送給單Channel的情況，如果是需要發(fā)送到Group的話需要使用如下代碼

async_to_sync(channel_layer.group_send)(
 group_name,
 {
 'type': 'chat.message',
 'message': '歡迎關(guān)注公眾號(hào)【運(yùn)維咖啡吧】'
 }
)

只需要將發(fā)送單channel的 send 改為 group_send ， channel_name 改為 group_name 即可

需要特別注意的是： 使用了channel layer之后一定要通過(guò)async_to_sync來(lái)異步執(zhí)行

頁(yè)面添加WebSocket支持

后端功能都已經(jīng)完成，我們最后需要添加前端頁(yè)面支持WebSocket

function connect() {
 if ( $('#file').val() ) {
 window.chatSocket = new WebSocket(
 'ws://' + window.location.host + '/ws/tailf/' + $('#file').val() + '/');

 chatSocket.onmessage = function(e) {
 var data = JSON.parse(e.data);
 var message = data['message'];
 document.querySelector('#chat-log').value += (message);
 // 跳轉(zhuǎn)到頁(yè)面底部
 $('#chat-log').scrollTop($('#chat-log')[0].scrollHeight);
 };

 chatSocket.onerror = function(e) {
 toastr.error('服務(wù)端連接異常！')
 };

 chatSocket.onclose = function(e) {
 toastr.error('websocket已關(guān)閉！')
 };
 } else {
 toastr.warning('請(qǐng)選擇要監(jiān)聽(tīng)的日志文件')
 }
 }

上一篇文章 中有詳細(xì)介紹過(guò)websocket的消息類(lèi)型，這里不多介紹了

至此我們一個(gè)日志監(jiān)聽(tīng)頁(yè)面完成了，包含了完整的監(jiān)聽(tīng)功能，但還無(wú)法終止，接著看下面的內(nèi)容

Web頁(yè)面主動(dòng)斷開(kāi)WebSocket

web頁(yè)面上“終止監(jiān)聽(tīng)”按鈕的主要邏輯就是觸發(fā)WebSocket的onclose方法，從而可以觸發(fā)Channels后端consumer的 disconnect 方法，進(jìn)而終止Celery的循環(huán)讀取日志任務(wù)

前端頁(yè)面通過(guò) .close() 可以直接觸發(fā)WebSocket關(guān)閉，當(dāng)然你如果直接關(guān)掉頁(yè)面的話也會(huì)觸發(fā)WebSocket的onclose消息，所以不用擔(dān)心Celery任務(wù)無(wú)法結(jié)束的問(wèn)題

function goclose() {
 console.log(window.chatSocket);

 window.chatSocket.close();
 window.chatSocket.onclose = function(e) {
 toastr.success('已終止日志監(jiān)聽(tīng)！')
 };
 }

至此我們包含完善功能的Tailf日志監(jiān)聽(tīng)、終止頁(yè)面就全部完成了

總結(jié)

以上所述是小編給大家介紹的Django實(shí)現(xiàn)web端tailf日志文件功能及實(shí)例詳解,希望對(duì)大家有所幫助，如果大家有任何疑問(wèn)請(qǐng)給我留言，小編會(huì)及時(shí)回復(fù)大家的。在此也非常感謝大家對(duì)腳本之家網(wǎng)站的支持！
如果你覺(jué)得本文對(duì)你有幫助，歡迎轉(zhuǎn)載，煩請(qǐng)注明出處，謝謝！

最新評(píng)論