Django使用Channels實現WebSocket--下篇

但願經過對這兩篇文章的學習，可以對Channels有更加深刻的瞭解，使用起來駕輕就熟遊刃有餘

經過上一篇《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執行異步任務和定時任務