1. 为什么需要专业的日志记录在开发Python应用时很多开发者习惯用print()来调试和记录信息。但当项目规模扩大后这种简单粗暴的方式会带来诸多问题日志信息混杂在标准输出中难以区分、无法区分不同级别的信息、缺乏时间戳导致问题定位困难、日志无法持久化存储等。我曾经维护过一个线上服务因为缺乏规范的日志记录当出现性能问题时花了整整两天才定位到内存泄漏的根源。Python内置的logging模块提供了工业级的日志解决方案。合理配置后可以实现分级记录DEBUG/INFO/WARNING/ERROR/CRITICAL多目的地输出控制台/文件/网络等灵活的格式定制线程安全性能优化如延迟写入2. 基础配置与核心概念2.1 最小化可用配置import logging # 基础配置 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, filenameapp.log ) logger logging.getLogger(__name__) # 使用示例 logger.debug(调试信息) # 不会输出因为级别设置为INFO logger.info(系统启动) logger.warning(磁盘空间不足80%)这个配置已经比print()强很多但实际项目中我们需要更精细的控制。常见问题包括所有日志混在一个文件难以维护缺乏日志轮转文件会无限增大多模块使用同一个logger导致日志来源不清晰2.2 核心组件解析Python logging系统的四个核心组件Logger应用程序直接调用的接口通过logging.getLogger(name)获取遵循层级命名如app.module是app的子logger可以设置独立的日志级别和处理器Handler决定日志的输出目的地常用类型StreamHandler控制台、FileHandler文件、 RotatingFileHandler轮转文件、SMTPHandler邮件发送一个logger可以添加多个handlerFilter提供更细粒度的日志过滤可以基于logger名称、日志内容等条件过滤Formatter控制日志的输出格式使用%(placeholder)s语法定制格式常用字段asctime、name、levelname、message、pathname等3. 生产环境最佳实践3.1 多模块日志组织# 主模块 app/main.py import logging from logging.handlers import RotatingFileHandler def setup_logging(): formatter logging.Formatter( %(asctime)s - %(name)s - %(levelname)s - %(message)s ) # 主logger配置 main_handler RotatingFileHandler( app.log, maxBytes10*1024*1024, backupCount5 ) main_handler.setFormatter(formatter) root_logger logging.getLogger() root_logger.setLevel(logging.INFO) root_logger.addHandler(main_handler) # 子模块 app/db/connector.py import logging logger logging.getLogger(app.db.connector) class Database: def query(self, sql): logger.debug(执行查询: %s, sql) # 使用字符串格式化而非f-string # ...关键技巧每个模块使用__name__获取logger自动形成层级避免在模块级别直接配置handler应由主程序统一配置使用RotatingFileHandler防止日志文件无限增长字符串格式化用%或{}而非f-string避免不必要的计算3.2 结构化日志记录现代日志分析系统如ELK、Splunk更适合处理结构化数据。我们可以使用python-json-loggerfrom pythonjsonlogger import jsonlogger formatter jsonlogger.JsonFormatter( %(asctime)s %(name)s %(levelname)s %(message)s, rename_fields{asctime: timestamp, levelname: level} ) handler logging.StreamHandler() handler.setFormatter(formatter) logger logging.getLogger(app) logger.addHandler(handler) logger.setLevel(logging.INFO) logger.info(用户登录, extra{ user_id: 123, ip: 192.168.1.1, tags: [auth, login] })输出示例{ timestamp: 2023-07-20T14:32:15Z, name: app, level: INFO, message: 用户登录, user_id: 123, ip: 192.168.1.1, tags: [auth, login] }3.3 性能优化技巧避免昂贵的日志预处理# 反例 - 即使日志级别高于DEBUG也会执行字符串格式化 logger.debug(f查询结果: {expensive_call()}) # 正例 - 使用延迟求值 logger.debug(查询结果: %s, expensive_call())使用QueueHandler实现异步日志from logging.handlers import QueueHandler, QueueListener import queue log_queue queue.Queue(-1) queue_handler QueueHandler(log_queue) # 主handler实际执行I/O操作 file_handler logging.FileHandler(app.log) # 使用单独的线程处理日志 listener QueueListener(log_queue, file_handler) listener.start() logger logging.getLogger(app) logger.addHandler(queue_handler)采样高频日志from logging import Filter class SampleFilter(Filter): def __init__(self, sample_rate0.1): self.sample_rate sample_rate def filter(self, record): import random return random.random() self.sample_rate debug_filter SampleFilter() logger.addFilter(debug_filter)4. 高级场景与疑难解决4.1 分布式系统日志追踪在微服务架构中我们需要在日志中注入追踪IDimport logging from contextvars import ContextVar import uuid request_id ContextVar(request_id, defaultNone) class RequestIdFilter(logging.Filter): def filter(self, record): record.request_id request_id.get() return True logger logging.getLogger(app) logger.addFilter(RequestIdFilter()) # 中间件示例 async def middleware(request, call_next): req_id str(uuid.uuid4()) token request_id.set(req_id) try: response await call_next(request) response.headers[X-Request-ID] req_id return response finally: request_id.reset(token)4.2 常见问题排查问题1日志不输出检查logger的有效级别注意logger层级继承确认handler是否添加到正确的logger验证handler的级别是否高于logger级别问题2日志重复输出检查是否多次添加了相同的handler避免在模块级别调用basicConfig()注意root logger可能已经配置了handler问题3日志文件不轮转确认RotatingFileHandler的maxBytes设置合理检查进程是否有文件写入权限确保不同进程不要共享同一个日志文件考虑使用ConcurrentLogHandler4.3 安全注意事项敏感信息过滤class SensitiveDataFilter(logging.Filter): PATTERNS { password: r(password\s*:\s*)([^]), token: r(Bearer\s)(\w) } def filter(self, record): import re msg record.getMessage() for name, pattern in self.PATTERNS.items(): msg re.sub(pattern, rf\1[REDACTED_{name}], msg) record.msg msg return True日志文件权限设置适当的文件权限如600避免在日志中记录完整的堆栈跟踪可能暴露内部实现定期审计日志内容5. 工具链与生态系统5.1 常用日志分析工具工具名称适用场景Python集成方式ELK Stack大规模日志收集分析通过FileBeat或直接HTTP发送Sentry错误监控与告警sentry-sdk库LokiGrafana生态日志系统promtail代理或直接APIDatadog全栈可观测性datadog-agent或HTTP API5.2 推荐库structlog提供更友好的结构化日志API支持处理器管道与标准logging模块兼容import structlog structlog.configure( processors[ structlog.processors.JSONRenderer() ] ) log structlog.get_logger() log.info(user_login, user_id123, actionlogin)loguru更简单的API设计内置丰富的处理器和格式化选项适合小型项目和快速原型from loguru import logger logger.add(file.log, rotation100 MB) logger.info(User {user} logged in, userAlice)Watchtower将日志发送到AWS CloudWatch支持boto3的完整功能5.3 性能监控集成将日志与指标系统结合from prometheus_client import Counter LOG_COUNTER Counter( app_log_messages_total, Total log messages by level, [level] ) class MetricsFilter(logging.Filter): def filter(self, record): LOG_COUNTER.labels(levelrecord.levelname).inc() return True logger.addFilter(MetricsFilter())在实际项目中我会根据团队的技术栈选择适当的工具组合。对于Kubernetes部署的应用LokiPrometheusGrafana的组合非常高效而对于传统的服务器部署ELK仍然是可靠的选择。无论哪种方案确保日志格式的一致性和可解析性都是最关键的前提。