Python logging.BufferingFormatter 怎么批量组织日志：缓冲区、格式化与输出边界

批量处理导入、定时任务或爬虫队列时，单条日志虽然能看懂，却很难看出一批任务从哪里开始、在哪里结束。Python 标准库的 logging.BufferingFormatter 正好提供了这个边界：它接收一组 LogRecord，统一加上批次头、逐条格式化，再补上批次尾。

先把“多条日志”变成一个有边界的批次

先约定一个具体场景：一次批处理会写入若干条记录，运维人员希望在日志中一眼看到批次编号、记录数量和结束状态。普通的 logging.Formatter 面向单条记录；批次格式化则需要拿到记录列表，这就是 BufferingFormatter 的职责。

它的接口很小。formatHeader(records) 和 formatFooter(records) 默认返回空字符串，format(records) 在非空时按“头部、逐条 line formatter、尾部”拼接。也就是说，批次的数量、标题和结束标记都应该从这组记录计算，而不是硬编码到每一条消息里。

用 BufferingFormatter 组织 header、记录和 footer

下面的格式化器给每个批次增加编号和记录数量。linefmt 仍然使用标准的单行格式器，因此时间、级别和消息字段不用重复实现。

import logging


class BatchFormatter(logging.BufferingFormatter):
    def formatHeader(self, records):
        batch_id = records[0].__dict__.get("batch_id", "unknown")
        return f"--- batch={batch_id} count={len(records)} ---\n"

    def formatFooter(self, records):
        return f"--- end batch count={len(records)} ---\n"


linefmt = logging.Formatter("%(levelname)s %(name)s %(message)s\n")
formatter = BatchFormatter(linefmt)

records = []
for index in range(2):
    record = logging.LogRecord(
        name="import", level=logging.INFO, pathname=__file__,
        lineno=10 + index, msg=f"row {index + 1} accepted", args=(), exc_info=None
    )
    record.batch_id = "orders-20260826"
    records.append(record)

print(formatter.format(records))

可见结果应该先出现一行 batch=orders-20260826 count=2，中间是两条带级别和 logger 名称的记录，最后出现同样数量的结束标记。传入空列表时，基类格式化器不会强行输出头尾；这正适合把空批次当成“没有可提交数据”处理。

把格式化器接到缓冲处理器

格式化器不会自动收集记录。真正持有缓冲区的是 logging.handlers.BufferingHandler：emit() 追加记录，随后调用 shouldFlush(record)；返回 True 后才进入 flush()。下面的处理器在容量到达时一次性写入一个文本文件。

from pathlib import Path
from logging.handlers import BufferingHandler


class BatchFileHandler(BufferingHandler):
    def __init__(self, capacity, path, formatter):
        super().__init__(capacity)
        self.path = Path(path)
        self.setFormatter(formatter)

    def flush(self):
        if not self.buffer:
            return
        payload = self.formatter.format(self.buffer)
        with self.path.open("a", encoding="utf-8") as stream:
            stream.write(payload)
        self.buffer.clear()


handler = BatchFileHandler(3, "import.log", formatter)
logger = logging.getLogger("import")
logger.setLevel(logging.INFO)
logger.addHandler(handler)

try:
    for row in range(3):
        logger.info("row %s accepted", row + 1, extra={"batch_id": "orders-20260826"})
finally:
    handler.close()

第三条记录写入后，容量边界触发一次刷新；close() 再负责处理不足容量的尾批次。这个关闭动作很关键：如果程序在第二条记录后直接退出，未满容量的记录不会凭空写入磁盘。

什么时候用 MemoryHandler，什么时候自己写 flush

MemoryHandler 是现成的选择。它把记录暂存到内存，在缓冲区满或记录级别达到 flushLevel 时，把记录交给 target handler；默认关闭时也会刷新。适合“平时聚合，遇到 ERROR 立刻把上下文交给文件或控制台”的场景。

如果批次需要自定义文件格式、批次 ID、事务提交或失败重试，直接继承 BufferingHandler 更清楚。不要把业务提交逻辑塞进 BufferingFormatter，否则“格式化失败”和“写入失败”会混成一个问题，调用方也无法决定是否重试。

• 只改变文本结构：继承或配置 BufferingFormatter。
• 只需要容量/级别触发并转发：优先使用 MemoryHandler。
• 需要落盘、提交、重试或指标：自定义 BufferingHandler.flush()，并让刷新动作可观测。

常见问题：批量日志最容易漏掉哪一步

BufferingFormatter 会自动保存日志吗？

不会。它只接收记录列表并返回字符串；保存、转发和清空缓冲区由 handler 负责。

为什么最后几条日志没有出现在文件里？

它们可能还没达到容量。程序退出前调用 handler 的 flush() 或 close()，并确认自定义 flush() 会清空已经成功写出的 buffer。

header 和 footer 适合放哪些信息？

适合放批次编号、记录数量、时间范围和结束状态；不要放只存在于单条记录中的易变字段，避免批次头与正文互相矛盾。

落地前的验收清单

先用两条记录验证 header、正文和 footer 的顺序，再用少于容量的记录验证关闭刷新，最后用一条 ERROR 记录验证是否需要立即转发。只要这三个边界都能在输出文件中找到，批次格式化和缓冲触发就已经分开验证，不容易把排版问题误判成日志丢失。