Python自定义带QueueHandler的logger如何跨类跨文件全局复用

核心原理 #

Python 的 logging 模块是全局层级单例设计，只要正确配置父级日志器，所有子日志器都会自动继承配置，不需要手动传递 logger 对象。你当前的 init 方法只给 custom_logging 模块对应的日志器（getLogger(__name__)）加了 handler，其他文件的日志器属于不同的命名节点，自然不会继承这套配置。

方案1：配置根日志器（全局生效，最简单） #

直接给根日志器加配置，所有模块的日志器都会默认继承这套规则，适合大部分业务场景。
修改你的 custom_logging.py 代码：

import logging
import queue
import json
import uuid
import requests
from logging.handlers import QueueHandler, QueueListener
from requests.adapters import HTTPAdapter, Retry
import atexit

# 全局持有QueueListener实例，避免被垃圾回收导致日志丢失
_listener = None

class CustomHttpHandler(logging.Handler):
    def __init__(self, url: str, token: str, silent: bool = True):
        self.url = url
        self.token = token
        self.silent = silent
        self.MAX_POOLSIZE = 100
        self.session = requests.Session()
        # 需要header配置的话取消下面注释即可
        # self.session.headers.update({
        #     "Content-Type": "application/json",
        #     "Authorization": "Bearer %s" % (self.token)
        # })
        self.session.mount("https://", HTTPAdapter(
            max_retries=Retry(
                total=5,
                backoff_factor=0.5,
                status_forcelist=[403, 500]
            ),
            pool_connections=self.MAX_POOLSIZE,
            pool_maxsize=self.MAX_POOLSIZE
        ))
        super().__init__()

    def emit(self, record):
        logEntry = self.format(record)
        logEntry=json.loads(logEntry)
        response = self.session.post(self.url, json=logEntry)
        if not self.silent:
            print(logEntry)
            print(response.content)

def init():
    global _listener
    # 避免重复初始化
    if _listener is not None:
        return

    # 取根日志器，所有子日志器都会继承配置
    root_logger = logging.getLogger()

    console_handler = logging.StreamHandler()
    console_formatter = logging.Formatter("%(levelname)s: %(message)s")
    console_handler.setFormatter(console_formatter)

    httpHandler = CustomHttpHandler(
        url="https://someAddress/log",
        token="1234",
        silent=True
    )
    MACAddr=uuid.getnode()
    http_formatter = logging.Formatter(json.dumps({
        "time": "%(asctime)s",
        "MAC Node": MACAddr,
        "line": "%(lineno)d",
        "module":"%(name)s",
        "logLevel": "%(levelname)s",
        "message": "%(message)s"
    }))
    httpHandler.setFormatter(http_formatter)

    log_queue = queue.Queue(-1)
    queue_handler = QueueHandler(log_queue)
    root_logger.addHandler(queue_handler)

    _listener = QueueListener(log_queue, console_handler, httpHandler)
    _listener.start()

    root_logger.setLevel(logging.DEBUG)

    # 注册程序退出钩子，自动停止listener，避免丢日志
    atexit.register(_listener.stop)

    root_logger.info("日志系统初始化完成")

使用方式 #

• 在 main.py 启动时先调用一次初始化：

import custom_logging
custom_logging.init()

• 其他任意文件/类中直接拿当前模块的日志器即可，自动继承配置：

import logging
# __name__会自动替换为当前模块名，日志的module字段会自动显示对应模块
logger = logging.getLogger(__name__)

logger.debug("调试日志")
logger.info("普通日志")

方案2：固定命名空间日志器（不污染全局根日志） #

如果不想影响第三方库的日志输出，可以配置指定命名空间的日志器，所有业务代码都从该命名空间取日志器。
在方案1的基础上修改init方法、新增获取日志器的封装方法：

def init():
    global _listener
    if _listener is not None:
        return
    # 配置固定命名的日志器，所有业务代码都取这个前缀的日志器
    app_logger = logging.getLogger("my_app")
    # 禁止向上传递到根日志器，避免重复输出
    app_logger.propagate = False

    console_handler = logging.StreamHandler()
    # 其余handler配置逻辑和方案1完全一致
    # ... 省略重复代码 ...
    app_logger.addHandler(queue_handler)
    app_logger.setLevel(logging.DEBUG)
    atexit.register(_listener.stop)

# 封装获取日志器的方法，统一入口
def get_logger(name: str = None):
    if name:
        return logging.getLogger(f"my_app.{name}")
    return logging.getLogger("my_app")

使用方式 #

• 初始化还是在main.py调用custom_logging.init()
• 其他文件中通过封装方法取日志器：

from custom_logging import get_logger
logger = get_logger(__name__)

logger.info("业务日志")

你之前想到的init返回logger的方式也可以用，但以上两种方案更符合logging模块的设计规范，不需要每个模块都手动导入特定logger对象，灵活性更高。

内容的提问来源于stack exchange，提问作者Martí