在 python 中使用 ZMQ 记录器

Question

我有一个多线程 python 程序，我已经实现了 zmq 记录器来记录。 我正在尝试使用 JSON 事件日志来解析后期处理中的日志，但我似乎无法让日志工作者以所需的格式登录，我只是得到了给函数的消息，而格式没有确实有效。

---

这是一个与我正在使用的代码非常相似的代码：

zmq_logger

import datetime
import logging
from pythonjsonlogger import jsonlogger
import os
import multiprocessing as mp
import pathlib
import time

import zmq
from zmq.log.handlers import PUBHandler

LOG_QUEUE = 'tcp://127.0.0.1:9142'


class LoggerListener(mp.Process):
    def __init__(self, logging_queue_string, logging_file_path):
        super().__init__()
        self.logging_queue_string = logging_queue_string
        self.logging_file_path = logging_file_path

    def run(self) -> None:
        listen(
            log_queue_string=self.logging_queue_string,
            logging_file_path=self.logging_file_path
        )


def listen(log_queue_string: str, logging_file_path: pathlib.Path):
    print('Starting logger listener')
    ctx = zmq.Context()
    sub = ctx.socket(zmq.SUB)
    sub.bind(log_queue_string)
    sub.setsockopt(zmq.SUBSCRIBE, b"tester")
    with logging_file_path.open('w') as f:
        while True:
            level, message = sub.recv_multipart()
            topic = level.decode('ascii').split('.')[0]
            message = message.decode('ascii')
            print(topic, message)
            if message.endswith('\n'):
                # trim trailing newline, which will get appended again
                message = message[:-1]
            f.write(message)
            f.flush()



def log_worker(log_queue_string: str, logger_topic: str):
    ctx = zmq.Context()
    pub = ctx.socket(zmq.PUB)
    pub.connect(log_queue_string)

    logger = logging.getLogger()
    # TODO: Pass to configs
    logger.setLevel(logging.DEBUG)
    handler = PUBHandler(pub)
    handler.setLevel(logging.DEBUG)
    jsformatter = jsonlogger.JsonFormatter('(level) (message)')
    handler.setFormatter(jsformatter)
    handler.root_topic = logger_topic.encode()
    logger.handlers.append(handler)

    return logger


if __name__ == '__main__':
    logger_process = LoggerListener(logging_queue_string=LOG_QUEUE, logging_file_path=pathlib.Path('./test_log.json'))
    logger_process.start()
    logger_process.join()

虚拟程序

from concurrent.futures import ThreadPoolExecutor
import multiprocessing as mp
from zmq_logger import log_worker, LOG_QUEUE


def dummy_function(logger):
    logger.debug('inside dummy function')


class JSONTester(mp.Process):
    @staticmethod
    def run():
        logger = log_worker(log_queue_string=LOG_QUEUE, logger_topic='tester')
        with ThreadPoolExecutor(max_workers=100) as executor:
            logger.debug('before execution')
            dummy_function(logger)


if __name__ == '__main__':
    jt = JSONTester()
    jt.start()
    jt.join()

我从 test_log.json 获取的日志记录：

before execution
inside dummy function

请注意：

1. 我尝试从 log_worker 接收 json 但解析失败
2. 我正在尝试使用 jsonlogger 来格式化 json 事件日志记录格式，但即使使用简单的日志记录格式，我也无法让它以正确的格式传递日志。

有什么想法吗？

谢谢， 霍德

Answer 1

快速浏览一下 zmq PUBHandler，我就明白了。 似乎 PUBHandler.setHandler() 没有实现，也没有覆盖类格式化程序。 所以在我的代码中：

def log_worker(log_queue_string: str, logger_topic: str):
    ctx = zmq.Context()
    pub = ctx.socket(zmq.PUB)
    pub.connect(log_queue_string)

    logger = logging.getLogger()
    # TODO: Pass to configs
    logger.setLevel(logging.DEBUG)
    handler = PUBHandler(pub)
    handler.setLevel(logging.DEBUG)
    jsformatter = jsonlogger.JsonFormatter('(level) (message)')
    handler.formatters[logging.DEBUG] = jsformatter
    handler.root_topic = logger_topic.encode()
    logger.handlers.append(handler)

    return logger

另一种解决方案是自己实现 setHandler 方法。