python和websocket構建實時日志跟蹤器的步驟

更新時間：2021年04月12日 14:16:17 作者：雞仔說

這篇文章主要介紹了python和websocket構建實時日志跟蹤器的步驟，幫助大家更好的理解和學習使用python，感興趣的朋友可以了解下

前言

websocket 是一種網(wǎng)絡傳輸協(xié)議?？稍趩蝹€ TCP 連接上進行全雙工通信?；诖?，websocket 使得客戶端與服務端的通信變得更加簡便和高效。

什么是 websocket

websocket 是獨立的、創(chuàng)建在 TCP 上的協(xié)議。該協(xié)議在 2008 年誕生，并在 2011 年成為國際標準。它的一個主要特點是——全雙工，即一旦建立連接，服務端或客戶端可以主動向對方推送消息。

在 websocket 出現(xiàn)之前，網(wǎng)站如果需要實現(xiàn)推送技術，都是采用輪詢的方式，即瀏覽器每隔一段時間就向服務器發(fā)出請求。這種模式的缺點在于，瀏覽器需要不斷向服務器發(fā)送請求，消耗很多的帶寬資源。比較新的 Comet 技術雖然也可以實現(xiàn)雙向通信，但依然需要反復發(fā)送請求，并且 Comet 中普遍采用的 HTTP 長連接也會消耗服務器資源。

基于以上的情況，HTML5 定義了 websocket 協(xié)議，能夠更好的節(jié)省服務器和帶寬資源。并且實現(xiàn)高效的實時通信。目前，所有的瀏覽器都支持它。

websocket 通信原理和機制

websocket 雖然是一種新的協(xié)議，但它不能夠脫離 http 單獨存在，當客戶端構建一個 websocket 實例，并且向服務端連接時，會首先發(fā)起一個 http 報文請求。告訴服務端需要將通信協(xié)議切換至 websocket。

如果服務端支持 websocket 協(xié)議，那么它會將通信協(xié)議切換至 websocket 并且返回響應報文。此時的返回狀態(tài)碼是 101，表示同意協(xié)議轉換請求，接下來便可以進行數(shù)據(jù)傳輸了。

websocket 之所以借助 HTTP 完成握手協(xié)議，是因為有良好的兼容性，默認端口是 80 和 443。握手階段不容易被防火墻屏蔽。

websocket 的特點

開銷小，服務器和客戶端交換數(shù)據(jù)時，協(xié)議包頭部包含較少的信息
實時性高，協(xié)議采用全雙工，相對于 http 請求客戶端發(fā)起請求，服務端才能響應的模式，延遲明顯更低
與 HTTP 有良好的兼容性，默認端口是 80 和 443。握手采用 HTTP 協(xié)議，不容易被防火墻屏蔽
支持文本和二進制數(shù)據(jù)傳輸
支持自定義拓展，用戶可以自己實現(xiàn)自定義的子協(xié)議
通過心跳機制保持服務端與客戶端的長連接

構建實時日志跟蹤的小例子

服務端開啟一個監(jiān)聽日志腳本的服務，服務會限制允許訪問的路徑范圍（防止黑客利用程序漏洞，掃描整個服務器）；服務器通過解析客戶端的請求，返回日志的消息內容給客戶端；服務器定時發(fā)送心跳檢測給客戶端，如果沒有收到客戶端的響應，則斷開連接

服務端核心程序代碼邏輯如下

with open(file_path) as f:
      # 首次讀取指定行數(shù)（NUM_LINES）的日志文件，發(fā)送給客戶端
            content = ''.join(deque(f, NUM_LINES))
            content = conv.convert(content, full=False)
            await websocket.send(content)
      
      # 如果發(fā)現(xiàn)客戶端有 tail 請求，則進行 tail 日志追蹤
            if tail:
        # 首先創(chuàng)建發(fā)起這次請求的心跳時間
                last_heartbeat = time.time()
                while True:
          # 每次 tail 服務端最新的日志記錄，返回給客戶端
                    content = f.read()
                    if content:
                        content = conv.convert(content, full=False)
                        await websocket.send(content)
                    else:
                        await asyncio.sleep(1)

                    # 檢測這次請求距上一次發(fā)起請求，是不是已經超過了最長心跳檢測時長，如果是，發(fā)起心跳檢測
                    if time.time() - last_heartbeat > HEARTBEAT_INTERVAL:
                        try:
                            await websocket.send('ping')
                            pong = await asyncio.wait_for(websocket.recv(), 5)
                            logger.info(f"pong:{pong}")
                            if pong != 'pong':
                                raise Exception()
                        except Exception:
                            raise Exception('Ping error')
                        else:
                            last_heartbeat = time.time()
            else:
                await websocket.close()

客戶端就非常簡單了，監(jiān)聽服務端日志的文件，發(fā)現(xiàn)有新的日志產生則輸出日志或者直接將日志實時展示在前端頁面上。相應地，如果需要長期監(jiān)聽，那么當服務端發(fā)送心跳檢測的信號過來，也需要回應響應的心跳反饋

客戶端核心代碼邏輯如下

async def consumer_handler(websocket: WebSocketClientProtocol) -> None:
    async for message in websocket:
        log_message(message)
        if message == "ping":
            await websocket.send("pong")

async def cousume(hostname: str, port: int, log_file: str, tail:bool=True) -> None:
    websocket_resource_url = f"ws://{hostname}:{port}{log_file}"
    if tail:
        websocket_resource_url = f"{websocket_resource_url}?tail=1"
    async with websockets.connect(websocket_resource_url) as websocket:
        await consumer_handler(websocket)

def log_message(message: str) -> None:
    logger.info(f"Message: {message}")

這里模擬一個日志生產文件

代碼邏輯如下

import os

from loguru import logger

class LoggerExtend(object):
    # 存放目錄名稱
    folder = '../logs'

    def __init__(self, filename, folder=None):

        self.folder = folder or self.folder

        if not os.path.exists(self.folder):
            os.mkdir(self.folder)

        self.file = self.folder + '/' + filename

        logger.add(self.file, rotation="100 MB")

    @property
    def get_logger(self):
        return logger

if __name__ == '__main__':
    logger = LoggerExtend(os.path.basename(__file__).replace(".py", ".log")).get_logger
    import time
    while True:

        logger.info("你好aaa")

最后依次啟動日志生產程序→服務端程序→客戶端程序

日志生產文件啟動后，運行效果如下