這是Django Channels系列文章的第二篇，以web端實現tailf的案例講解Channels的具體使用以及跟Celery的結合

通過上一篇《Django使用Channels實現WebSocket--上篇》的學習應該對Channels的各種概念有了清晰的認知，可以順利的將Channels框架整合到自己的Django專案中實現WebSocket了，本篇文章將以一個Channels+Celery實現web端tailf功能的例子更加深入的介紹Channels

先說下我們要實現的目標：所有登入的使用者可以檢視tailf日誌頁面，在頁面上能夠選擇日誌檔案進行監聽，多個頁面終端同時監聽任何日誌都互不影響，頁面同時提供終止監聽的按鈕能夠終止前端的輸出以及後臺對日誌檔案的讀取

最終實現的結果見下圖

接著我們來看下具體的實現過程

技術實現

所有程式碼均基於以下軟體版本：

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

celery4在windows下支援不完善，所以請在linux下執行測試

日誌資料定義

我們只希望使用者能夠查詢固定的幾個日誌檔案，就不是用資料庫僅藉助settings.py檔案裡寫全域性變數來實現資料儲存

在settings.py裡新增一個叫TAILF的變數，型別為字典，key標識檔案的編號，value標識檔案的路徑

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

基礎Web頁面搭建

假設你已經建立好了一個叫tailf的app，並新增到了settings.py的INSTALLED_APPS中，app的目錄結構大概如下

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

依然先構建一個標準的Django頁面，相關程式碼如下

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'),
]
 

因為我們規定只有通過登入的使用者才能檢視日誌，所以引入Django自帶的LoginView，logoutView幫助我們快速構建Login，Logout功能

指定了登入模板使用login.html，它就是一個標準的登入頁面，post傳入username和password兩個引數即可，不貼程式碼了

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裝飾器，來判斷使用者是否登入，未登入就給跳到/login登入頁面

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

template:

{% extends "base.html" %}

{% block content %}
<div class="col-sm-8">
  <select class="form-control" id="file">
    <option value="">選擇要監聽的日誌</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="開始監聽"/><br/>
</div>
<div class="col-sm-2">
  <input class="btn btn-warning btn-block" type="button" onclick="goclose()" value="終止監聽"/><br/>
</div>
<div class="col-sm-12">
  <textarea class="form-control" id="chat-log" disabled rows="20"></textarea>
</div>
{% endblock %}

前端拿到TAILF後通過迴圈的方式填充到select選擇框下，因為資料是字典格式，使用logDict.items的方式可以迴圈出字典的key和value

這樣一個日誌監聽頁面就完成了，但還無法實現日誌的監聽，繼續往下

整合Channels實現WebSocket

日誌監聽功能主要的設計思路就是頁面跟後端伺服器建立websocket長連線，後端通過celery非同步執行while迴圈不斷的讀取日誌檔案然後傳送到websocket的channel裡，實現頁面上的實時顯示

接著我們來整合channels

1. 先新增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),
        ])
    )
})

直接將路由資訊寫入到了URLRouter裡，注意路由資訊的外層多了一個list，區別於上一篇中介紹的寫路由檔案路徑的方式

頁面需要將監聽的日誌檔案傳遞給後端，我們使用routing正則P<id>\d+傳檔案ID給後端程式，後端程式拿到ID之後根據settings中指定的TAILF解析出日誌路徑

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

1. 新增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):
        # 中止執行中的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的單通道模式，每一個新連線都會啟用一個新的channel，彼此互不影響，可以隨意終止任何一個監聽日誌的請求

connect

我們知道self.scope類似於Django中的request，記錄了豐富的請求資訊，通過self.scope["url_route"]["kwargs"]["id"]取出routing中正則匹配的日誌ID

然後將id和channel_name傳遞給celery的任務函式tailf，tailf根據id取到日誌檔案的路徑，然後迴圈檔案，將新內容根據channel_name寫入對應channel

disconnect

當websocket連線斷開的時候我們需要終止Celery的Task執行，以清除celery的資源佔用

終止Celery任務使用到revoke指令，採用如下程式碼來實現

self.result.revoke(terminate=True)

注意self.result是一個result物件，而非id

引數terminate=True的意思是是否立即終止Task，為True時無論Task是否正在執行都立即終止，為False（預設）時需要等待Task執行結束之後才會終止，我們使用了While迴圈不設定為True就永遠不會終止了

終止Celery任務的另外一種方法是：

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

send_message

方便我們通過Django的view或者Celery的task呼叫給channel傳送訊息，官方也比較推薦這種方式

使用Celery非同步迴圈讀取日誌

上邊已經集成了Channels實現了WebSocket，但connect函式中的celery任務tailf還沒有實現，下邊來實現它

關於Celery的詳細內容可以看這篇文章：《Django配置Celery執行非同步任務和定時任務》，本文就不介紹整合使用以及細節原理，只講一下任務task

task實現程式碼如下：

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": "微信公眾號【運維咖啡吧】原創 版權所有 " + str(line)
                        }
                    )
                else:
                    time.sleep(0.5)
    except Exception as e:
        print(e)

這裡邊主要涉及到Channels中另一個非常重要的點：從Channels的外部發送訊息給Channel

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

async_to_sync(channel_layer.send)(
    channel_name,
    {
        "type": "send.message",
        "message": "微信公眾號【運維咖啡吧】原創 版權所有 " + str(line)
    }
)

channel_name 對應於傳遞給這個任務的channel_name，傳送訊息給這個名字的channel

type 對應於我們Channels的TailfConsumer類中的send_message方法，將方法中的_換成.即可

message 就是要傳送給這個channel的具體資訊

上邊是傳送給單Channel的情況，如果是需要傳送到Group的話需要使用如下程式碼

async_to_sync(channel_layer.group_send)(
    group_name,
    {
        'type': 'chat.message',
        'message': '歡迎關注公眾號【運維咖啡吧】'
    }
)

只需要將傳送單channel的send改為group_send，channel_name改為group_name即可

需要特別注意的是：使用了channel layer之後一定要通過async_to_sync來非同步執行

頁面新增WebSocket支援

後端功能都已經完成，我們最後需要新增前端頁面支援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);
        // 跳轉到頁面底部
        $('#chat-log').scrollTop($('#chat-log')[0].scrollHeight);
      };

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

      chatSocket.onclose = function(e) {
        toastr.error('websocket已關閉！')
      };
    } else {
      toastr.warning('請選擇要監聽的日誌檔案')
    }
  }

上一篇文章中有詳細介紹過websocket的訊息型別，這裡不多介紹了

至此我們一個日誌監聽頁面完成了，包含了完整的監聽功能，但還無法終止，接著看下面的內容

Web頁面主動斷開WebSocket

web頁面上“終止監聽”按鈕的主要邏輯就是觸發WebSocket的onclose方法，從而可以觸發Channels後端consumer的disconnect方法，進而終止Celery的迴圈讀取日誌任務

前端頁面通過.close()可以直接觸發WebSocket關閉，當然你如果直接關掉頁面的話也會觸發WebSocket的onclose訊息，所以不用擔心Celery任務無法結束的問題

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

    window.chatSocket.close();
    window.chatSocket.onclose = function(e) {
      toastr.success('已終止日誌監聽！')
    };
  }

至此我們包含完善功能的Tailf日誌監聽、終止頁面就全部完成了

寫在最後

兩篇文章結束不知道你是否對Channels有了更深一步的瞭解，能夠操刀上手將Channels用在自己的專案中，實現理想的功能。個人覺得Channels的重點和難點在於對channel layer的理解和運用，真正的理解了並能熟練運用，相信你一定能夠舉一反三完美實現更多需求。最後如果對本文的demo原始碼感興趣可以關注微信公眾號【運維咖啡吧】後臺回覆小二加我微信向我索取，一定有求必應

相關文章推薦閱讀：

• Django使用Channels實現WebSocket--上篇
• Django配置Celery執行非同步任務和定時任務