Python logging QueueHandler 停机时怎么保证日志不丢：队列排空、关闭顺序与异常兜底

Python 服务把日志从业务线程转进 QueueHandler 后，正常运行时看不到问题，真正容易丢日志的是停机的最后几百毫秒：进程先退出了，后台的 QueueListener 还没来得及把队列里的记录交给文件处理器。可靠的收尾顺序是先停止新的写入，再等待队列排空，最后停止监听器并关闭处理器；等待超过上限时要留下明确的告警，而不是假装全部成功。

先把停机现场拆成三条线

QueueHandler.emit() 的职责是把 LogRecord 放入队列，真正写文件的是 QueueListener 线程中的处理器。两者解耦后，应用线程返回“日志已提交”只代表记录进入了队列，并不代表文件已经落盘。

因此停机要处理三个状态：应用不再产生新记录；队列里的记录持续减少；监听器退出后，文件处理器仍被正确关闭。只调用 logging.shutdown()，通常无法替代这段应用级协调，因为它不了解你自建的监听线程是否已经消费完队列。

权限和配置先固定，再启动监听器

生产进程应让日志目录由部署用户预先创建，并让应用只拥有写入所需的权限。不要在运行时用高权限补目录，也不要把队列对象、监听器和计数器散落在全局变量里；收尾时很难确认谁还会继续写。

from __future__ import annotations

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

LOG_PATH = Path("var/app.log")
log_queue: queue.Queue[logging.LogRecord] = queue.Queue()
accepted = 0
accepted_lock = threading.Lock()

file_handler = logging.FileHandler(LOG_PATH, encoding="utf-8")
file_handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(message)s"))
queue_handler = logging.handlers.QueueHandler(log_queue)
listener = logging.handlers.QueueListener(log_queue, file_handler)

logger = logging.getLogger("service")
logger.setLevel(logging.INFO)
logger.handlers[:] = [queue_handler]
logger.propagate = False
listener.start()

示例里把文件处理器交给监听器管理，业务 logger 只保留一个队列处理器。真实项目还要在创建 LOG_PATH 前检查父目录、磁盘空间和当前用户权限；这些错误应该在启动阶段暴露。

停机顺序：先封口，再排空，最后关闭

“停止生产”最好由应用状态控制。收到 SIGTERM 后先把 accepting 设为 False，让新请求不再进入业务处理；正在执行的任务完成后再进入日志收尾。不要直接在线程之间调用 queue.join()，却仍允许生产者继续放入新记录，这会让排空时间不断被延长。

def shutdown(timeout: float = 5.0) -> bool:
    deadline = time.monotonic() + timeout

    # 1. 业务层先停止接收新任务；此处之后不再创建新的日志生产者。
    service_state.stop_accepting()

    # 2. 等待 QueueListener 对每条记录调用 task_done()。
    remaining = max(0.0, deadline - time.monotonic())
    drained = wait_queue_drained(log_queue, remaining)

    # 3. 只有排空或确认超时后，才停止后台监听线程。
    listener.stop()
    file_handler.close()
    logging.shutdown()

    if not drained:
        logger.warning("log queue did not drain before shutdown timeout")
    return drained

def wait_queue_drained(q: queue.Queue[logging.LogRecord], timeout: float) -> bool:
    deadline = time.monotonic() + timeout
    while q.unfinished_tasks:
        if time.monotonic() >= deadline:
            return False
        time.sleep(0.01)
    return True

这里的 unfinished_tasks 是排空判断，不是业务指标；它只适合在你确定所有入队路径都会经过同一个队列时使用。若项目封装了队列，应让封装层提供线程安全的 wait_drained()，不要让调用方到处读取内部字段。

超时不是静默成功，必须留下证据

队列排不空通常有三类原因：文件系统写入变慢、监听器线程抛出异常后停止消费，或者还有线程在关闭阶段继续产生日志。停机超时后继续强制退出是可以接受的运维决策，但返回值必须是失败，并把未完成数量、停机耗时和最后一条已写日志记录到标准错误或监控系统。

建议维护两个单调计数：入队成功数与文件处理完成数。两者在正常停机后应相等；若不相等，先保留现场，不要仅凭“进程退出码为 0”判断日志完整。文件末行也可以作为人工抽查，但不能替代计数器，因为日志消息可能相同。

三个容易踩中的关闭误区

把 listener.stop() 放在 queue.join() 前面

监听器一停，队列就没人继续消费，join() 只能一直等到超时。顺序必须是先让监听器工作到队列为空，再停止监听器。

在 handler 中吞掉所有异常

吞异常会让业务线程看起来正常，但监听器可能已经不再处理后续记录。至少要把处理异常计数、最后一次错误和处理线程存活状态送到 stderr 或监控端。

把队列排空当成无限等待

磁盘故障或网络文件系统卡住时，无限等待会拖住发布和重启。超时值应和进程管理器的停止宽限期一起设计，并在达到上限时保留未完成数。

上线前做一次可复查的验收

测试不要只验证文件存在。让生产者写入带序号的记录，例如 shutdown-check-0001 到 shutdown-check-0100，触发正常停机后读取文件，检查序号数量、最后一条记录和程序返回值。再故意把文件处理器替换成一个延迟处理器，验证超时会返回失败并留下告警。

expected = {f"shutdown-check-{i:04d}" for i in range(1, 101)}
written = set(Path("var/app.log").read_text(encoding="utf-8").splitlines())
missing = [item for item in sorted(expected) if not any(item in line for line in written)]
if missing:
    raise RuntimeError(f"missing log records: {missing[:5]}")

这类验收能发现“停止成功但还有记录在内存里”的问题，也能把变更后的停机行为固定成回归测试。真正需要多进程、多文件轮转或集中式日志时，再把同样的封口、排空、关闭边界延伸到对应组件。

相关问题

logging.shutdown() 还能不能调用？

可以调用，但它应放在自建监听器停止之后，作为标准 logging 处理器的最后清理动作，不能拿它替代队列排空。

队列排空超时后要不要重试？

进程已进入退出阶段时不建议无限重试。记录未完成数量和根因，按停机宽限期决定是否保留进程；下一次启动时检查日志文件与磁盘状态。

同步写文件是不是就不会丢？

同步写只减少了进程内队列这一层的不确定性，仍可能遇到缓冲区、磁盘或轮转失败。无论哪种方案，都需要可观测的错误和停机验收。

小结

QueueHandler 解决的是业务线程不被文件 I/O 拖慢，不是自动提供停机可靠性。把停止生产、等待排空、停止监听器和关闭处理器分开，给超时保留失败证据，再用带序号的记录做验收，才能知道这次重启到底有没有丢日志。