4. Python 日志库实战：logging 模块高级用法、日志格式化、日志处理器（Handler）配置

在异步任务编排场景中，日志不再是简单的 print 输出，而是需要具备结构化、可路由、可持久化的能力。Python 内置的 logging 模块提供了强大的三级架构：Logger（记录器）、Handler（处理器）和 Formatter（格式化器）。本章将深入讲解如何利用这些组件构建生产级的日志采集系统。

4.1 logging 模块核心组件架构

logging 模块采用分层设计，每个组件各司其职。理解它们之间的关系是配置高级日志系统的前提。

4.2 日志格式化（Formatter）高级用法

Formatter 决定了日志记录的最终呈现形式。在异步任务编排中，我们需要记录任务 ID、执行耗时、上下文追踪 ID 等关键字段。

4.2.1 标准格式字段

4.2.2 自定义格式化器

通过继承 logging.Formatter 并重写 format() 方法，可以注入自定义上下文信息。

import logging
import threading

class TaskContextFormatter(logging.Formatter):
    """为日志注入任务上下文（task_id, trace_id）"""

    def format(self, record):
        # 从当前线程或协程上下文获取任务 ID
        task_id = getattr(threading.current_thread(), 'task_id', 'N/A')
        trace_id = getattr(threading.current_thread(), 'trace_id', 'N/A')
        record.task_id = task_id
        record.trace_id = trace_id
        return super().format(record)

# 使用自定义格式化器
fmt = TaskContextFormatter(
    '[%(asctime)s] [%(levelname)-8s] '
    '[task:%(task_id)s] [trace:%(trace_id)s] '
    '%(name)s:%(filename)s:%(lineno)d - %(message)s'
)

handler = logging.StreamHandler()
handler.setFormatter(fmt)
logger = logging.getLogger('async.worker')
logger.addHandler(handler)
logger.setLevel(logging.DEBUG)

4.3 日志处理器（Handler）配置实战

不同的输出目标需要不同的 Handler。以下是开发环境中最常用的三种 Handler 配置方案。

4.3.1 控制台输出：StreamHandler

适用于开发调试阶段，实时查看日志流。

import logging
import sys

console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(logging.DEBUG)
console_fmt = logging.Formatter(
    '%(asctime)s | %(levelname)-8s | %(message)s',
    datefmt='%H:%M:%S'
)
console_handler.setFormatter(console_fmt)

4.3.2 文件轮转：RotatingFileHandler

生产环境必备，避免单个日志文件无限增长。

from logging.handlers import RotatingFileHandler

file_handler = RotatingFileHandler(
    filename='logs/async_tasks.log',
    maxBytes=10 * 1024 * 1024,  # 10 MB
    backupCount=5,               # 保留 5 个备份
    encoding='utf-8'
)
file_handler.setLevel(logging.INFO)
file_fmt = logging.Formatter(
    '%(asctime)s [%(levelname)s] %(name)s - %(message)s'
)
file_handler.setFormatter(file_fmt)

4.3.3 按时间轮转：TimedRotatingFileHandler

适合按天或按小时切分日志，便于归档分析。

from logging.handlers import TimedRotatingFileHandler

time_handler = TimedRotatingFileHandler(
    filename='logs/daily_task.log',
    when='midnight',      # 每天午夜轮转
    interval=1,
    backupCount=30,       # 保留 30 天
    encoding='utf-8'
)
time_handler.setLevel(logging.WARNING)
time_handler.setFormatter(file_fmt)

4.4 多 Handler 组合配置

在异步任务编排系统中，通常需要同时使用多个 Handler，实现分级输出：控制台显示所有级别，文件只记录 WARNING 及以上，错误日志单独归档。

4.4.1 完整配置示例

import logging
from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler

def setup_logging():
    """配置异步任务编排系统的日志体系"""
    logger = logging.getLogger('async_tasks')
    logger.setLevel(logging.DEBUG)  # 全局最低级别

    # 1. 控制台 Handler（开发调试用）
    console = logging.StreamHandler()
    console.setLevel(logging.DEBUG)
    console_fmt = logging.Formatter(
        '[%(asctime)s] %(levelname)-8s %(message)s',
        datefmt='%H:%M:%S'
    )
    console.setFormatter(console_fmt)

    # 2. 轮转文件 Handler（常规日志）
    rotating = RotatingFileHandler(
        'logs/tasks.log', maxBytes=5*1024*1024, backupCount=3, encoding='utf-8'
    )
    rotating.setLevel(logging.INFO)
    rotating_fmt = logging.Formatter(
        '%(asctime)s | %(levelname)-8s | %(name)s | %(message)s'
    )
    rotating.setFormatter(rotating_fmt)

    # 3. 错误专用 Handler（异常追踪）
    error_handler = RotatingFileHandler(
        'logs/errors.log', maxBytes=10*1024*1024, backupCount=7, encoding='utf-8'
    )
    error_handler.setLevel(logging.ERROR)
    error_fmt = logging.Formatter(
        '%(asctime)s | %(levelname)-8s | %(name)s:%(filename)s:%(lineno)d\n'
        'Message: %(message)s\n'
        'Exception: %(exc_info)s\n'
        '---'
    )
    error_handler.setFormatter(error_fmt)

    # 添加到 Logger
    logger.addHandler(console)
    logger.addHandler(rotating)
    logger.addHandler(error_handler)

    return logger

# 使用
log = setup_logging()
log.info('Task orchestration started')
log.error('Task failed: timeout after 30s', exc_info=True)

4.5 日志级别与传播机制

理解日志级别和传播规则，可以避免日志重复或丢失。

4.6 异步任务中的日志最佳实践

• 使用 contextvars 传递上下文：在异步任务中，使用 contextvars.ContextVar 存储 task_id 和 trace_id，确保协程切换时上下文不丢失。
• 避免在热路径中格式化日志：使用 logger.debug("Task %s processed %d items", task_id, count) 的惰性格式化方式，避免字符串拼接开销。
• 统一日志时间格式：所有 Handler 使用 ISO 8601 格式（%Y-%m-%dT%H:%M:%S），便于日志聚合系统（如 ELK）解析。
• 错误日志包含完整堆栈：在 except 块中始终使用 logger.exception() 或 exc_info=True，记录完整的异常回溯。

import contextvars
import logging

task_id_var = contextvars.ContextVar('task_id', default='N/A')
trace_id_var = contextvars.ContextVar('trace_id', default='N/A')

class AsyncContextFormatter(logging.Formatter):
    def format(self, record):
        record.task_id = task_id_var.get()
        record.trace_id = trace_id_var.get()
        return super().format(record)

# 在异步任务入口设置上下文
async def run_task(task_id: str):
    token1 = task_id_var.set(task_id)
    token2 = trace_id_var.set(f'trace-{task_id}')
    try:
        logger.info('Task started')
        # ... 执行任务逻辑
    finally:
        task_id_var.reset(token1)
        trace_id_var.reset(token2)