Python进阶：用websocket构建实时日志跟踪器-技术圈

作者：鸡仔

来源：鸡仔说

前言

websocket 是一种网络传输协议。可在单个 TCP 连接上进行全双工通信。基于此，websocket 使得客户端与服务端的通信变得更加简便和高效。

什么是 websocket

websocket 是独立的、创建在 TCP 上的协议。该协议在 2008 年诞生，并在 2011 年成为国际标准。它的一个主要特点是——全双工，即一旦建立连接，服务端或客户端可以主动向对方推送消息。

在 websocket 出现之前，网站如果需要实现推送技术，都是采用轮询的方式，即浏览器每隔一段时间就向服务器发出请求。这种模式的缺点在于，浏览器需要不断向服务器发送请求，消耗很多的带宽资源。比较新的 Comet 技术虽然也可以实现双向通信，但依然需要反复发送请求，并且 Comet 中普遍采用的 HTTP 长连接也会消耗服务器资源。

基于以上的情况，HTML5 定义了 websocket 协议，能够更好的节省服务器和带宽资源。并且实现高效的实时通信。目前，所有的浏览器都支持它。

websocket 通信原理和机制

websocket 虽然是一种新的协议，但它不能够脱离 http 单独存在，当客户端构建一个 websocket 实例，并且向服务端连接时，会首先发起一个 http 报文请求。告诉服务端需要将通信协议切换至 websocket。

如果服务端支持 websocket 协议，那么它会将通信协议切换至 websocket 并且返回响应报文。此时的返回状态码是 101，表示同意协议转换请求，接下来便可以进行数据传输了。

websocket 之所以借助 HTTP 完成握手协议，是因为有良好的兼容性，默认端口是 80 和 443。握手阶段不容易被防火墙屏蔽。

websocket 的特点

开销小，服务器和客户端交换数据时，协议包头部包含较少的信息
实时性高，协议采用全双工，相对于 http 请求客户端发起请求，服务端才能响应的模式，延迟明显更低
与 HTTP 有良好的兼容性，默认端口是 80 和 443。握手采用 HTTP 协议，不容易被防火墙屏蔽
支持文本和二进制数据传输
支持自定义拓展，用户可以自己实现自定义的子协议
通过心跳机制保持服务端与客户端的长连接

构建实时日志跟踪的小例子

服务端开启一个监听日志脚本的服务，服务会限制允许访问的路径范围（防止黑客利用程序漏洞，扫描整个服务器）；服务器通过解析客户端的请求，返回日志的消息内容给客户端；服务器定时发送心跳检测给客户端，如果没有收到客户端的响应，则断开连接

服务端核心程序代码逻辑如下

with open(file_path) as f:
      # 首次读取指定行数（NUM_LINES）的日志文件，发送给客户端
            content = ''.join(deque(f, NUM_LINES))
            content = conv.convert(content, full=False)
            await websocket.send(content)
      
      # 如果发现客户端有 tail 请求，则进行 tail 日志追踪
            if tail:
        # 首先创建发起这次请求的心跳时间
                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)

                    # 检测这次请求距上一次发起请求，是不是已经超过了最长心跳检测时长，如果是，发起心跳检测
                    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()

客户端就非常简单了，监听服务端日志的文件，发现有新的日志产生则输出日志或者直接将日志实时展示在前端页面上。相应地，如果需要长期监听，那么当服务端发送心跳检测的信号过来，也需要回应响应的心跳反馈

客户端核心代码逻辑如下

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")