背景：日志写盘为什么会拖慢主流程

在数据处理或 Web 服务的主循环里，一条 logger.info(...) 最终会走到 StreamHandler 或 FileHandler 的 emit()，其中包含格式化、加锁以及真正的 write() / flush()。当日志文件位于网络盘、容器可写层或磁盘繁忙时，这一次写操作就可能让主线程停顿。日志本身是旁路需求，不应该成为主流程的瓶颈。

logging.handlers.QueueHandler 与 QueueListener 提供了一种官方支持的解耦方式：业务线程只把 LogRecord 放入队列，后台线程负责取出并交给真正的 handler 落盘。写入队列是内存操作，通常不会因为磁盘抖动而阻塞主线程。

工作模型

1. 创建一个线程安全的 queue.Queue。
2. 业务侧 logger 只挂载 QueueHandler，它的 emit() 实际是 put_nowait()。
3. 在独立线程中启动 QueueListener，它绑定一个真实的 FileHandler。
4. 退出前调用 listener.stop()，让监听线程把队列里剩余的记录处理完再结束。

注意一个容易踩的细节：QueueHandler.prepare() 会先用自身 formatter 把消息格式化成字符串，再替换 record.msg，同时清空 exc_info（异常堆栈文本已经被格式化进消息）。因此不要给 QueueHandler 设置带时间戳的 formatter，业务格式化应交给 FileHandler，否则会出现重复的时间字段。

最小可运行示例

import logging
import logging.handlers
import queue
import time
from pathlib import Path

LOG_PATH = Path("app.log")


def build_logger(log_path: Path):
    # 无界队列：不会丢日志，但积压会占用内存
    log_queue = queue.Queue(-1)

    file_handler = logging.FileHandler(log_path, encoding="utf-8")
    file_handler.setLevel(logging.INFO)
    file_handler.setFormatter(
        logging.Formatter(
            "%(asctime)s %(levelname)s %(threadName)s %(name)s %(message)s"
        )
    )

    # QueueHandler 不要设置 formatter，保持默认即可
    queue_handler = logging.handlers.QueueHandler(log_queue)

    logger = logging.getLogger("demo")
    logger.setLevel(logging.INFO)
    logger.handlers.clear()          # 避免重复添加
    logger.addHandler(queue_handler)
    logger.propagate = False         # 防止再冒泡到 root 造成重复输出

    listener = logging.handlers.QueueListener(
        log_queue,
        file_handler,
        respect_handler_level=True,  # 让 handler 自己的 level 生效
    )
    listener.start()
    return logger, listener


def main() -> None:
    logger, listener = build_logger(LOG_PATH)
    try:
        for i in range(1_000):
            logger.info("处理第 %s 条任务", i)
            if i % 200 == 0:
                logger.warning("当前进度 %s", i)
        logger.info("主流程结束")
    finally:
        # 会投递哨兵并等待监听线程收尾，队列中的记录仍会被写入
        listener.stop()


if __name__ == "__main__":
    main()

运行方法

把代码保存为 queue_logging_demo.py，在已安装 Python 3 的环境中执行：

python queue_logging_demo.py
cat app.log

日志文件按 时间 级别 线程名 logger名 消息 输出，QueueListener 线程负责写盘，主线程只做入队。

边界与限制

• 队列容量：queue.Queue(-1) 表示无界队列，不会丢日志，但日志产生速度长期大于落盘速度时，内存会持续增长。若改用有界队列，QueueHandler 使用 put_nowait()，队列满时会触发 logging.raiseExceptions 对应的错误处理路径。是否允许丢弃日志，需要在业务上明确。
• 单消费者串行落盘：QueueListener 只有一个线程，落盘仍是串行的。若单个文件写入成为瓶颈，应考虑换用更快的存储介质或做日志切分，而不是简单增加监听线程。
• 多进程场景：多进程需使用 multiprocessing.Queue，并且 QueueListener 要在父进程中启动；在 Windows 上注意进程启动方式带来的对象传递限制。
• 生命周期：listener.stop() 之后，新产生的日志不会被写入文件。应在业务收尾阶段先停止记录，再调用 stop()；异常退出时可通过 atexit 注册收尾函数作为兜底。
• 与框架集成：若使用 uvicorn、gunicorn 等框架，它们会自行配置 logger，直接 handlers.clear() 可能覆盖框架行为，建议只对自定义 logger 生效，或通过 dictConfig 统一声明。
• 格式化职责：异常堆栈会被 QueueHandler 提前格式化进消息体，因此 FileHandler 的 formatter 中保留 %(message)s 即可正确输出堆栈，不必再依赖 exc_info。

总结：QueueHandler 加 QueueListener 的价值在于把日志的“产出”和“落盘”分离，让主线程从磁盘 IO 中解耦出来。代价是多了一个线程和一段可能积压的内存队列，使用时需要根据日志量级明确队列是否有界，并在退出路径上可靠地执行 stop()。