文章目录

引言

在软件开发过程中，日志记录是不可或缺的重要组成部分。无论是调试程序、监控系统运行状态，还是分析用户行为，良好的日志系统都能提供极大的帮助。Python内置的logging模块提供了一个灵活而强大的日志记录系统，能够满足从简单脚本到复杂企业级应用的各种需求。

本文将深入探讨logging模块的各个方面，包括基本概念、核心组件、高级特性、最佳实践以及常见问题解决方案，帮助您全面掌握Python日志记录的艺术。

1. logging模块概述

1.1 为什么需要日志记录

在程序开发中，我们经常使用print语句来输出调试信息，但这种方法存在诸多局限性：

  • 无法区分不同重要程度的信息
  • 输出目标单一（通常只是控制台）
  • 缺乏灵活的格式控制
  • 难以在生产环境中管理和分析

logging模块解决了这些问题，提供了：

  • 多级别日志记录
  • 多种输出目标
  • 灵活的格式配置
  • 高效的日志过滤和管理

1.2 logging模块的核心组件

logging模块基于四个核心组件构建：

  1. Logger：记录器，应用程序直接使用的接口
  2. Handler：处理器，决定日志输出的位置
  3. Formatter：格式化器，控制日志的输出格式
  4. Filter：过滤器，提供更细粒度的日志控制

2. 快速入门

2.1 最简单的日志记录

让我们从一个最简单的例子开始：

import logging

# 记录一条日志
logging.warning('这是一个警告信息')
logging.info('这是一个信息消息')  # 这行不会输出，因为默认级别是WARNING

运行上述代码，输出结果如下：

WARNING:root:这是一个警告信息

2.2 基本配置

对于简单的应用，可以使用basicConfig进行快速配置：

import logging

# 配置日志系统
logging.basicConfig(
    level=logging.DEBUG,  # 设置日志级别为DEBUG
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',  # 设置日志格式
    datefmt='%Y-%m-%d %H:%M:%S'  # 设置时间格式
)

# 测试不同级别的日志
logging.debug('调试信息')
logging.info('普通信息')
logging.warning('警告信息')
logging.error('错误信息')
logging.critical('严重错误信息')

输出结果：

2023-11-01 10:30:45 - root - DEBUG - 调试信息
2023-11-01 10:30:45 - root - INFO - 普通信息
2023-11-01 10:30:45 - root - WARNING - 警告信息
2023-11-01 10:30:45 - root - ERROR - 错误信息
2023-11-01 10:30:45 - root - CRITICAL - 严重错误信息

3. 日志级别详解

3.1 标准日志级别

logging模块定义了6个标准日志级别，按严重程度从低到高排列：

级别 数值 描述
DEBUG 10 详细信息，通常仅在诊断问题时使用
INFO 20 确认程序按预期运行
WARNING 30 表示意外情况，或指示一些问题，但程序仍能运行
ERROR 40 由于更严重的问题，程序已无法执行某些功能
CRITICAL 50 严重错误，程序本身可能无法继续运行

3.2 级别使用示例

import logging

# 配置日志
logging.basicConfig(level=logging.DEBUG)

def divide_numbers(a, b):
    logging.debug(f"开始除法运算: {a} / {b}")
    
    if b == 0:
        logging.error("除数不能为零!")
        return None
    
    result = a / b
    logging.info(f"计算结果: {result}")
    return result

# 测试函数
print("正常情况:")
divide_numbers(10, 2)

print("\n异常情况:")
divide_numbers(10, 0)

输出结果：

正常情况:
DEBUG:root:开始除法运算: 10 / 2
INFO:root:计算结果: 5.0

异常情况:
DEBUG:root:开始除法运算: 10 / 0
ERROR:root:除数不能为零!

4. Logger记录器详解

4.1 创建和使用Logger

在实际应用中，我们通常不会直接使用根记录器，而是创建自己的记录器：

import logging

# 创建记录器
logger = logging.getLogger('my_app')
logger.setLevel(logging.DEBUG)

# 创建控制台处理器
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.DEBUG)

# 创建格式化器
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
console_handler.setFormatter(formatter)

# 将处理器添加到记录器
logger.addHandler(console_handler)

# 使用记录器
logger.info('应用程序启动')
logger.debug('调试信息')
logger.error('发生错误')

4.2 Logger的层次结构

Logger支持层次结构，通过点号分隔名称：

import logging

# 配置根记录器
logging.basicConfig(level=logging.WARNING)

# 创建不同层次的记录器
main_logger = logging.getLogger('my_app')
module_logger = logging.getLogger('my_app.database')
submodule_logger = logging.getLogger('my_app.database.queries')

# 设置模块级记录器的级别
module_logger.setLevel(logging.DEBUG)

# 创建处理器和格式化器
handler = logging.StreamHandler()
formatter = logging.Formatter('%(name)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
module_logger.addHandler(handler)

# 测试日志记录
print("根记录器级别:", logging.getLogger().level)
print("主记录器级别:", main_logger.level)
print("模块记录器级别:", module_logger.level)
print("子模块记录器级别:", submodule_logger.level)

print("\n日志输出:")
main_logger.warning('主记录器警告')
module_logger.info('模块记录器信息')  # 这会输出，因为模块记录器级别是DEBUG
submodule_logger.debug('子模块记录器调试')  # 这会输出，因为继承了模块记录器的DEBUG级别

5. Handler处理器详解

5.1 常用的处理器类型

logging模块提供了多种处理器，用于将日志发送到不同的目的地：

import logging
import sys
from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler

# 创建记录器
logger = logging.getLogger('multi_handler_app')
logger.setLevel(logging.DEBUG)

# 1. 控制台处理器
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(logging.INFO)

# 2. 文件处理器
file_handler = logging.FileHandler('app.log')
file_handler.setLevel(logging.DEBUG)

# 3. 回滚文件处理器（按大小）
rotating_handler = RotatingFileHandler(
    'rotating_app.log', 
    maxBytes=1024*1024,  # 1MB
    backupCount=5
)
rotating_handler.setLevel(logging.DEBUG)

# 4. 定时回滚文件处理器
timed_rotating_handler = TimedRotatingFileHandler(
    'timed_app.log',
    when='midnight',  # 每天午夜
    interval=1,
    backupCount=7
)
timed_rotating_handler.setLevel(logging.INFO)

# 创建格式化器
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')

# 为所有处理器设置格式化器
console_handler.setFormatter(formatter)
file_handler.setFormatter(formatter)
rotating_handler.setFormatter(formatter)
timed_rotating_handler.setFormatter(formatter)

# 将处理器添加到记录器
logger.addHandler(console_handler)
logger.addHandler(file_handler)
logger.addHandler(rotating_handler)
logger.addHandler(timed_rotating_handler)

# 测试日志记录
for i in range(100):
    logger.debug(f'调试信息 {i}')
    logger.info(f'普通信息 {i}')
    logger.error(f'错误信息 {i}')

5.2 自定义处理器

您也可以创建自定义处理器：

import logging
import requests

class WebhookHandler(logging.Handler):
    """自定义处理器，将日志发送到Webhook"""
    
    def __init__(self, webhook_url):
        super().__init__()
        self.webhook_url = webhook_url
    
    def emit(self, record):
        log_entry = self.format(record)
        payload = {
            'message': log_entry,
            'level': record.levelname,
            'timestamp': self.formatTime(record)
        }
        try:
            requests.post(self.webhook_url, json=payload, timeout=5)
        except Exception as e:
            # 避免递归错误，使用print
            print(f"无法发送日志到Webhook: {e}")

# 使用自定义处理器
logger = logging.getLogger('webhook_logger')
logger.setLevel(logging.ERROR)

webhook_handler = WebhookHandler('https://example.com/webhook')
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
webhook_handler.setFormatter(formatter)

logger.addHandler(webhook_handler)

# 测试
logger.error('这是一个测试错误，将被发送到Webhook')

6. Formatter格式化器详解

6.1 标准格式属性

Formatter支持多种属性来定制日志输出格式：

import logging
import os

# 创建记录器
logger = logging.getLogger('formatter_demo')
logger.setLevel(logging.DEBUG)

# 创建处理器
handler = logging.StreamHandler()

# 创建详细格式的格式化器
detailed_formatter = logging.Formatter(
    fmt='%(asctime)s.%(msecs)03d | %(levelname)-8s | %(name)s | %(filename)s:%(lineno)d | %(funcName)s | %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)

handler.setFormatter(detailed_formatter)
logger.addHandler(handler)

# 定义一个测试函数
def test_function():
    logger.info('这是在测试函数内部')
    try:
        result = 10 / 0
    except Exception as e:
        logger.error('发生除零错误', exc_info=True)

# 调用测试函数
test_function()

输出结果：

2023-11-01 10:30:45.123 | INFO     | formatter_demo | demo.py:25 | test_function | 这是在测试函数内部
2023-11-01 10:30:45.124 | ERROR    | formatter_demo | demo.py:29 | test_function | 发生除零错误
Traceback (most recent call last):
  File "demo.py", line 27, in test_function
    result = 10 / 0
ZeroDivisionError: division by zero

6.2 自定义格式化器

您可以创建自定义格式化器来满足特定需求：

import logging
import json
from datetime import datetime

class JSONFormatter(logging.Formatter):
    """将日志记录格式化为JSON"""
    
    def format(self, record):
        log_entry = {
            'timestamp': datetime.utcnow().isoformat() + 'Z',
            'level': record.levelname,
            'logger': record.name,
            'message': record.getMessage(),
            'module': record.module,
            'function': record.funcName,
            'line': record.lineno
        }
        
        # 如果有异常信息，添加到日志中
        if record.exc_info:
            log_entry['exception'] = self.formatException(record.exc_info)
        
        # 添加额外的字段
        if hasattr(record, 'custom_fields'):
            log_entry.update(record.custom_fields)
            
        return json.dumps(log_entry, ensure_ascii=False)

# 使用JSON格式化器
logger = logging.getLogger('json_logger')
logger.setLevel(logging.DEBUG)

handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logger.addHandler(handler)

# 添加自定义字段
extra_info = {'user_id': '12345', 'request_id': 'req-67890'}
logger.info('用户登录成功', extra={'custom_fields': extra_info})

# 记录错误
try:
    1 / 0
except Exception:
    logger.error('计算错误', exc_info=True)

7. Filter过滤器详解

7.1 使用过滤器

过滤器可以提供比日志级别更细粒度的控制：

import logging

class LevelRangeFilter(logging.Filter):
    """只允许特定级别范围内的日志通过"""
    
    def __init__(self, min_level, max_level):
        super().__init__()
        self.min_level = min_level
        self.max_level = max_level
    
    def filter(self, record):
        return self.min_level <= record.levelno <= self.max_level

class KeywordFilter(logging.Filter):
    """根据关键词过滤日志"""
    
    def __init__(self, keyword):
        super().__init__()
        self.keyword = keyword
    
    def filter(self, record):
        if self.keyword in record.getMessage():
            return False  # 包含关键词的日志被过滤掉
        return True

# 创建记录器和处理器
logger = logging.getLogger('filter_demo')
logger.setLevel(logging.DEBUG)

# 创建多个处理器，每个使用不同的过滤器
# 只处理INFO和WARNING级别的处理器
info_warning_handler = logging.StreamHandler()
info_warning_handler.setLevel(logging.INFO)
info_warning_handler.addFilter(LevelRangeFilter(logging.INFO, logging.WARNING))

# 只处理ERROR和CRITICAL级别的处理器
error_critical_handler = logging.StreamHandler()
error_critical_handler.setLevel(logging.ERROR)
error_critical_handler.addFilter(LevelRangeFilter(logging.ERROR, logging.CRITICAL))

# 过滤包含"密码"的日志
sensitive_filter = KeywordFilter('密码')

# 创建格式化器
formatter = logging.Formatter('%(levelname)s: %(message)s')
info_warning_handler.setFormatter(formatter)
error_critical_handler.setFormatter(formatter)

# 添加处理器到记录器
logger.addHandler(info_warning_handler)
logger.addHandler(error_critical_handler)

# 测试日志记录
logger.info('普通信息')
logger.warning('警告信息')
logger.error('错误信息')
logger.critical('严重错误信息')
logger.info('用户密码修改成功')  # 这条将被过滤掉

8. 配置logging的多种方式

8.1 代码配置

import logging
import logging.config

def setup_logging():
    """通过代码配置日志系统"""
    
    config = {
        'version': 1,
        'disable_existing_loggers': False,
        'formatters': {
            'standard': {
                'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
            },
            'detailed': {
                'format': '%(asctime)s.%(msecs)03d | %(levelname)-8s | %(name)s | %(filename)s:%(lineno)d | %(message)s',
                'datefmt': '%Y-%m-%d %H:%M:%S'
            },
        },
        'handlers': {
            'console': {
                'class': 'logging.StreamHandler',
                'level': 'INFO',
                'formatter': 'standard',
                'stream': 'ext://sys.stdout'
            },
            'file': {
                'class': 'logging.FileHandler',
                'level': 'DEBUG',
                'formatter': 'detailed',
                'filename': 'app.log',
                'mode': 'a'
            },
            'rotating_file': {
                'class': 'logging.handlers.RotatingFileHandler',
                'level': 'INFO',
                'formatter': 'detailed',
                'filename': 'rotating_app.log',
                'maxBytes': 1048576,  # 1MB
                'backupCount': 5
            }
        },
        'loggers': {
            '': {  # 根记录器
                'handlers': ['console'],
                'level': 'WARNING'
            },
            'my_app': {
                'handlers': ['console', 'file'],
                'level': 'DEBUG',
                'propagate': False
            },
            'my_app.database': {
                'handlers': ['rotating_file'],
                'level': 'INFO',
                'propagate': False
            }
        }
    }
    
    logging.config.dictConfig(config)

# 应用配置
setup_logging()

# 测试不同的记录器
root_logger = logging.getLogger()
app_logger = logging.getLogger('my_app')
db_logger = logging.getLogger('my_app.database')

root_logger.info('根记录器信息')  # 不会输出，因为级别是WARNING
app_logger.debug('应用调试信息')   # 会输出到文件和控制台
db_logger.info('数据库操作信息')   # 只会输出到回滚文件

8.2 配置文件方式

创建一个logging.conf文件：

[loggers]
keys=root,my_app,my_app_database

[handlers]
keys=consoleHandler,fileHandler,rotatingFileHandler

[formatters]
keys=standardFormatter,detailedFormatter

[logger_root]
level=WARNING
handlers=consoleHandler

[logger_my_app]
level=DEBUG
handlers=consoleHandler,fileHandler
qualname=my_app
propagate=0

[logger_my_app_database]
level=INFO
handlers=rotatingFileHandler
qualname=my_app.database
propagate=0

[handler_consoleHandler]
class=StreamHandler
level=INFO
formatter=standardFormatter
args=(sys.stdout,)

[handler_fileHandler]
class=FileHandler
level=DEBUG
formatter=detailedFormatter
args=('app.log', 'a')

[handler_rotatingFileHandler]
class=handlers.RotatingFileHandler
level=INFO
formatter=detailedFormatter
args=('rotating_app.log', 'a', 1048576, 5)

[formatter_standardFormatter]
format=%(asctime)s - %(name)s - %(levelname)s - %(message)s

[formatter_detailedFormatter]
format=%(asctime)s.%(msecs)03d | %(levelname)-8s | %(name)s | %(filename)s:%(lineno)d | %(message)s
datefmt=%Y-%m-%d %H:%M:%S

然后在代码中使用这个配置：

import logging
import logging.config

# 从配置文件加载配置
logging.config.fileConfig('logging.conf')

# 使用记录器
logger = logging.getLogger('my_app')
logger.info('应用程序启动')

8.3 环境变量配置

对于需要根据环境动态调整日志配置的情况，可以使用环境变量：

import logging
import logging.config
import os
import json

def setup_logging_from_env():
    """根据环境变量配置日志"""
    
    # 默认配置
    default_config = {
        'version': 1,
        'disable_existing_loggers': False,
        'handlers': {
            'console': {
                'class': 'logging.StreamHandler',
                'level': 'INFO',
                'formatter': 'standard'
            }
        },
        'formatters': {
            'standard': {
                'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
            }
        },
        'root': {
            'level': 'INFO',
            'handlers': ['console']
        }
    }
    
    # 检查环境变量
    env = os.getenv('APP_ENV', 'development')
    
    if env == 'production':
        default_config['handlers']['file'] = {
            'class': 'logging.handlers.RotatingFileHandler',
            'level': 'WARNING',
            'formatter': 'standard',
            'filename': '/var/log/app.log',
            'maxBytes': 10485760,  # 10MB
            'backupCount': 10
        }
        default_config['root']['handlers'] = ['file']
        default_config['root']['level'] = 'WARNING'
    
    elif env == 'development':
        default_config['root']['level'] = 'DEBUG'
    
    # 应用配置
    logging.config.dictConfig(default_config)

# 应用配置
setup_logging_from_env()

# 使用记录器
logger = logging.getLogger()
logger.info(f"当前环境: {os.getenv('APP_ENV', 'development')}")

9. 高级特性和最佳实践

9.1 上下文信息记录

在复杂的应用程序中，通常需要记录额外的上下文信息：

import logging
import threading
from functools import wraps

# 创建线程本地存储
thread_local = threading.local()

class ContextFilter(logging.Filter):
    """添加上下文信息到日志记录"""
    
    def filter(self, record):
        # 添加线程信息
        record.thread_id = threading.get_ident()
        record.thread_name = threading.current_thread().name
        
        # 添加自定义上下文
        if hasattr(thread_local, 'context'):
            for key, value in thread_local.context.items():
                setattr(record, key, value)
        
        return True

def with_logging_context(**context):
    """装饰器，为函数添加日志上下文"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 设置上下文
            if not hasattr(thread_local, 'context'):
                thread_local.context = {}
            
            old_context = thread_local.context.copy()
            thread_local.context.update(context)
            
            try:
                return func(*args, **kwargs)
            finally:
                # 恢复旧上下文
                thread_local.context = old_context
        return wrapper
    return decorator

# 配置日志
logger = logging.getLogger('context_logger')
logger.setLevel(logging.DEBUG)

handler = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s | %(thread_name)s | %(user_id)s | %(request_id)s | %(levelname)s | %(message)s')
handler.setFormatter(formatter)
handler.addFilter(ContextFilter())
logger.addHandler(handler)

# 使用上下文
@with_logging_context(user_id='anonymous', request_id='initial')
def process_request():
    logger.info('开始处理请求')
    
    @with_logging_context(user_id='john_doe', request_id='req_123')
    def authenticate_user():
        logger.info('用户认证成功')
        logger.warning('密码即将过期')
    
    authenticate_user()
    logger.info('请求处理完成')

# 测试
process_request()

9.2 性能考虑

日志记录可能会影响应用程序性能，特别是在高频率记录DEBUG级别日志时：

import logging
import time
from functools import wraps

# 优化前的慢速日志记录
def slow_function():
    logger = logging.getLogger('slow_app')
    
    # 不推荐的写法：即使日志级别高于DEBUG，字符串连接仍然会执行
    for i in range(1000):
        logger.debug('处理第 ' + str(i) + ' 个元素，数据: ' + get_complex_data())

def get_complex_data():
    """模拟获取复杂数据的函数"""
    time.sleep(0.001)  # 模拟耗时操作
    return "复杂数据"

# 优化后的快速日志记录
def fast_function():
    logger = logging.getLogger('fast_app')
    
    # 推荐的写法：使用%格式化或f-string，并检查日志级别
    for i in range(1000):
        if logger.isEnabledFor(logging.DEBUG):
            logger.debug('处理第 %d 个元素，数据: %s', i, get_complex_data())

# 性能对比装饰器
def timing_decorator(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.time()
        result = func(*args, **kwargs)
        end_time = time.time()
        print(f"{func.__name__} 执行时间: {end_time - start_time:.4f}秒")
        return result
    return wrapper

# 配置日志级别为INFO，这样DEBUG日志不会输出
logging.basicConfig(level=logging.INFO)

# 测试性能
@timing_decorator
def test_slow():
    slow_function()

@timing_decorator
def test_fast():
    fast_function()

print("优化前:")
test_slow()

print("\n优化后:")
test_fast()

9.3 结构化日志记录

对于需要日志分析的系统，结构化日志非常有用：

import logging
import json

class StructuredLogger:
    """结构化日志记录器"""
    
    def __init__(self, name):
        self.logger = logging.getLogger(name)
        self.context = {}
    
    def bind(self, **kwargs):
        """绑定上下文信息"""
        self.context.update(kwargs)
        return self
    
    def log(self, level, event, **kwargs):
        """记录结构化日志"""
        if self.logger.isEnabledFor(level):
            log_data = {
                'timestamp': logging.Formatter().formatTime(self.logger.makeRecord(
                    self.logger.name, level, '', 0, '', (), None
                )),
                'level': logging.getLevelName(level),
                'event': event,
                'logger': self.logger.name
            }
            
            # 添加上下文和额外字段
            log_data.update(self.context)
            log_data.update(kwargs)
            
            # 记录JSON格式的日志
            self.logger.log(level, json.dumps(log_data, ensure_ascii=False))

    def debug(self, event, **kwargs):
        self.log(logging.DEBUG, event, **kwargs)
    
    def info(self, event, **kwargs):
        self.log(logging.INFO, event, **kwargs)
    
    def warning(self, event, **kwargs):
        self.log(logging.WARNING, event, **kwargs)
    
    def error(self, event, **kwargs):
        self.log(logging.ERROR, event, **kwargs)

# 配置JSON日志处理器
logger = logging.getLogger('structured_app')
logger.setLevel(logging.DEBUG)

handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter('%(message)s'))  # 只输出原始消息（JSON）
logger.addHandler(handler)

# 使用结构化日志记录器
structured_logger = StructuredLogger('structured_app')

# 记录带有上下文的日志
structured_logger.bind(
    application="my_app",
    version="1.0.0",
    environment="production"
).info(
    "user_login",
    user_id="user123",
    ip_address="192.168.1.100",
    user_agent="Mozilla/5.0..."
)

structured_logger.bind(
    transaction_id="txn_456"
).error(
    "payment_failed",
    amount=100.50,
    reason="insufficient_funds",
    user_id="user123"
)

10. 常见问题与解决方案

10.1 重复日志问题

import logging

# 问题：重复日志
def duplicate_logs_problem():
    # 创建记录器
    logger = logging.getLogger('duplicate_example')
    logger.setLevel(logging.DEBUG)
    
    # 添加多个处理器（可能在不同地方）
    handler1 = logging.StreamHandler()
    handler1.setFormatter(logging.Formatter('Handler1: %(message)s'))
    logger.addHandler(handler1)
    
    handler2 = logging.StreamHandler()
    handler2.setFormatter(logging.Formatter('Handler2: %(message)s'))
    logger.addHandler(handler2)
    
    # 记录一条日志，但会输出两次
    logger.info('这条消息会重复输出')

# 解决方案：检查是否已存在处理器
def duplicate_logs_solution():
    logger = logging.getLogger('no_duplicate_example')
    logger.setLevel(logging.DEBUG)
    
    # 只有在没有处理器时才添加
    if not logger.handlers:
        handler = logging.StreamHandler()
        handler.setFormatter(logging.Formatter('%(name)s: %(message)s'))
        logger.addHandler(handler)
    
    # 或者禁用传播
    logger.propagate = False
    
    logger.info('这条消息不会重复输出')

print("问题演示:")
duplicate_logs_problem()

print("\n解决方案:")
duplicate_logs_solution()

10.2 日志记录性能问题

import logging
import time

# 配置日志
logging.basicConfig(level=logging.WARNING)  # 生产环境使用较高的级别

class PerformanceCriticalClass:
    """性能关键的类"""
    
    def __init__(self):
        self.logger = logging.getLogger(self.__class__.__name__)
        
        # 预创建日志记录方法，避免每次查找
        self._log_debug = self.logger.debug
        self._log_info = self.logger.info
        self._log_warning = self.logger.warning
        self._log_error = self.logger.error
    
    def process_data(self, data):
        """处理数据的方法"""
        
        # 不推荐的写法：即使不记录DEBUG，字符串格式化仍然执行
        # self.logger.debug(f"处理数据: {expensive_format(data)}")
        
        # 推荐的写法：先检查级别
        if self.logger.isEnabledFor(logging.DEBUG):
            self._log_debug("处理数据: %s", expensive_format(data))
        
        # 处理逻辑...
        result = len(data)
        
        self._log_info("数据处理完成，结果: %d", result)
        return result

def expensive_format(data):
    """模拟昂贵的格式化操作"""
    time.sleep(0.001)  # 模拟耗时操作
    return f"格式化后的数据: {data}"

# 测试性能
critical_class = PerformanceCriticalClass()

start_time = time.time()
for i in range(100):
    critical_class.process_data([1, 2, 3, 4, 5])
end_time = time.time()

print(f"处理100次数据耗时: {end_time - start_time:.4f}秒")

10.3 多线程和多进程日志记录

import logging
import threading
import multiprocessing
import time
from logging.handlers import QueueHandler, QueueListener

def setup_queue_logging():
    """设置基于队列的日志系统，支持多进程"""
    
    # 创建队列
    log_queue = multiprocessing.Queue(-1)
    
    # 配置最终处理器
    handlers = [
        logging.StreamHandler(),
        logging.FileHandler('multiprocess_app.log')
    ]
    
    # 创建队列监听器
    listener = QueueListener(log_queue, *handlers)
    listener.start()
    
    return log_queue, listener

def worker_process(log_queue, worker_id):
    """工作进程函数"""
    
    # 设置队列处理器
    handler = QueueHandler(log_queue)
    logger = logging.getLogger(f'worker_{worker_id}')
    logger.addHandler(handler)
    logger.setLevel(logging.DEBUG)
    
    # 记录日志
    for i in range(5):
        logger.info(f"工作进程 {worker_id} 执行任务 {i}")
        time.sleep(0.1)
    
    logger.info(f"工作进程 {worker_id} 完成")

def multi_thread_example():
    """多线程日志记录示例"""
    
    logger = logging.getLogger('multi_thread_app')
    logger.setLevel(logging.DEBUG)
    
    # 添加处理器
    handler = logging.StreamHandler()
    handler.setFormatter(logging.Formatter('%(asctime)s | %(threadName)s | %(levelname)s | %(message)s'))
    logger.addHandler(handler)
    
    def worker_thread(thread_id):
        for i in range(3):
            logger.info(f"线程 {thread_id} 执行任务 {i}")
            time.sleep(0.05)
    
    # 创建并启动多个线程
    threads = []
    for i in range(3):
        thread = threading.Thread(target=worker_thread, args=(i,), name=f'Worker-{i}')
        threads.append(thread)
        thread.start()
    
    # 等待所有线程完成
    for thread in threads:
        thread.join()

# 测试多线程
print("多线程日志记录示例:")
multi_thread_example()

# 测试多进程
print("\n多进程日志记录示例:")
if __name__ == '__main__':
    log_queue, listener = setup_queue_logging()
    
    processes = []
    for i in range(2):
        process = multiprocessing.Process(target=worker_process, args=(log_queue, i))
        processes.append(process)
        process.start()
    
    for process in processes:
        process.join()
    
    # 停止监听器
    listener.stop()

11. 实际应用案例

11.1 Web应用日志配置

import logging
import logging.config
from flask import Flask, request, g
import time

def setup_flask_logging():
    """配置Flask应用日志"""
    
    config = {
        'version': 1,
        'disable_existing_loggers': False,
        'formatters': {
            'standard': {
                'format': '%(asctime)s | %(levelname)-8s | %(name)s | %(message)s'
            },
            'access': {
                'format': '%(asctime)s | %(client_ip)s | %(method)s %(path)s | %(status_code)d | %(response_time).3fs'
            }
        },
        'handlers': {
            'console': {
                'class': 'logging.StreamHandler',
                'level': 'INFO',
                'formatter': 'standard'
            },
            'access_file': {
                'class': 'logging.handlers.RotatingFileHandler',
                'level': 'INFO',
                'formatter': 'access',
                'filename': 'access.log',
                'maxBytes': 10485760,  # 10MB
                'backupCount': 5
            },
            'error_file': {
                'class': 'logging.handlers.RotatingFileHandler',
                'level': 'ERROR',
                'formatter': 'standard',
                'filename': 'error.log',
                'maxBytes': 10485760,
                'backupCount': 5
            }
        },
        'loggers': {
            '': {  # 根记录器
                'level': 'INFO',
                'handlers': ['console']
            },
            'app': {
                'level': 'DEBUG',
                'handlers': ['console', 'error_file'],
                'propagate': False
            },
            'access': {
                'level': 'INFO',
                'handlers': ['access_file'],
                'propagate': False
            }
        }
    }
    
    logging.config.dictConfig(config)

app = Flask(__name__)
setup_flask_logging()

# 获取记录器
app_logger = logging.getLogger('app')
access_logger = logging.getLogger('access')

@app.before_request
def before_request():
    """在请求前记录开始时间"""
    g.start_time = time.time()

@app.after_request
def after_request(response):
    """在请求后记录访问日志"""
    if hasattr(g, 'start_time'):
        response_time = time.time() - g.start_time
        
        # 记录访问日志
        access_logger.info(
            '',  # 消息为空，因为所有信息都在格式化器中
            extra={
                'client_ip': request.remote_addr,
                'method': request.method,
                'path': request.path,
                'status_code': response.status_code,
                'response_time': response_time
            }
        )
    
    return response

@app.route('/')
def index():
    app_logger.info('访问首页')
    return 'Hello, World!'

@app.route('/api/data')
def get_data():
    app_logger.info('获取数据API被调用')
    return {'data': [1, 2, 3, 4, 5]}

@app.route('/error')
def cause_error():
    app_logger.error('故意触发的错误')
    raise ValueError('这是一个测试错误')

if __name__ == '__main__':
    app.run(debug=True)

11.2 数据分析管道日志

import logging
import pandas as pd
import numpy as np
from logging.handlers import SMTPHandler

class DataPipelineLogger:
    """数据分析管道日志记录器"""
    
    def __init__(self, pipeline_name):
        self.pipeline_name = pipeline_name
        self.logger = logging.getLogger(f'pipeline.{pipeline_name}')
        self.setup_logging()
    
    def setup_logging(self):
        """配置管道日志"""
        
        # 避免重复配置
        if self.logger.handlers:
            return
        
        self.logger.setLevel(logging.DEBUG)
        self.logger.propagate = False
        
        # 控制台处理器
        console_handler = logging.StreamHandler()
        console_handler.setLevel(logging.INFO)
        
        # 文件处理器
        file_handler = logging.FileHandler(f'{self.pipeline_name}.log')
        file_handler.setLevel(logging.DEBUG)
        
        # 邮件处理器（仅错误）
        # email_handler = SMTPHandler(
        #     mailhost=('smtp.example.com', 587),
        #     fromaddr='alerts@example.com',
        #     toaddrs=['admin@example.com'],
        #     subject=f'数据管道 {self.pipeline_name} 错误',
        #     credentials=('username', 'password'),
        #     secure=()
        # )
        # email_handler.setLevel(logging.ERROR)
        
        # 格式化器
        formatter = logging.Formatter(
            '%(asctime)s | %(levelname)-8s | %(name)s | %(step)s | %(message)s'
        )
        
        console_handler.setFormatter(formatter)
        file_handler.setFormatter(formatter)
        # email_handler.setFormatter(formatter)
        
        self.logger.addHandler(console_handler)
        self.logger.addHandler(file_handler)
        # self.logger.addHandler(email_handler)
    
    def log_step(self, step_name, message, level=logging.INFO, **kwargs):
        """记录管道步骤"""
        extra = {'step': step_name}
        extra.update(kwargs)
        self.logger.log(level, message, extra=extra)
    
    def log_dataframe_stats(self, step_name, df, message="数据统计"):
        """记录DataFrame统计信息"""
        stats = {
            'rows': len(df),
            'columns': len(df.columns),
            'memory_mb': df.memory_usage(deep=True).sum() / 1024**2,
            'null_count': df.isnull().sum().sum()
        }
        
        self.log_step(
            step_name, 
            f"{message}: {stats}",
            extra=stats
        )

# 使用示例
def run_data_pipeline():
    """运行数据管道示例"""
    
    logger = DataPipelineLogger('customer_etl')
    
    try:
        # 步骤1: 数据提取
        logger.log_step('extract', '开始数据提取')
        data = {
            'customer_id': range(1, 101),
            'age': np.random.randint(18, 70, 100),
            'spend': np.random.normal(100, 30, 100)
        }
        df = pd.DataFrame(data)
        logger.log_dataframe_stats('extract', df, '提取的数据')
        
        # 步骤2: 数据清洗
        logger.log_step('transform', '开始数据清洗')
        df_clean = df[df['age'] >= 18].copy()
        df_clean['spend_category'] = pd.cut(
            df_clean['spend'], 
            bins=[0, 50, 150, float('inf')],
            labels=['low', 'medium', 'high']
        )
        logger.log_dataframe_stats('transform', df_clean, '清洗后的数据')
        
        # 步骤3: 数据加载
        logger.log_step('load', '开始数据加载')
        output_file = 'customer_data_processed.csv'
        df_clean.to_csv(output_file, index=False)
        logger.log_step('load', f'数据已保存到 {output_file}')
        
        # 完成
        logger.log_step('complete', '管道执行成功', logging.INFO)
        
    except Exception as e:
        logger.log_step('error', f'管道执行失败: {str(e)}', logging.ERROR, exc_info=True)
        raise

# 运行管道
run_data_pipeline()

12. 总结

Python的logging模块是一个功能强大且灵活的日志记录系统，能够满足从简单脚本到复杂企业级应用的各种需求。通过本文的详细讲解，您应该已经掌握了：

  1. 基本概念：了解了Logger、Handler、Formatter和Filter四个核心组件
  2. 级别管理：学会了如何合理使用不同日志级别
  3. 配置方法：掌握了代码配置、文件配置和环境变量配置等多种方式
  4. 高级特性：学习了上下文记录、结构化日志、性能优化等高级用法
  5. 实际问题：了解了常见问题的解决方案和最佳实践

关键要点总结

  • 合理使用日志级别：根据环境调整日志级别，开发环境使用DEBUG，生产环境使用WARNING或ERROR
  • 避免性能问题：使用isEnabledFor检查级别，避免不必要的字符串格式化
  • 结构化日志：对于需要日志分析的系统，使用结构化日志格式
  • 适当配置：根据应用需求选择合适的处理器和格式化器
  • 错误处理：始终记录完整的异常信息，使用exc_info=True

进一步学习

要深入了解logging模块，建议查阅：

通过合理使用logging模块，您可以构建出高效、可维护的日志系统，大大提升应用程序的可观测性和可维护性。

参考文献

  1. Python Software Foundation. (2023). “logging — Logging facility for Python”. Python 3.11 Documentation.
  2. “Logging in Python”. Real Python.
  3. “The Ultimate Guide to Logging in Python”. Scout APM Blog.
  4. “Python Logging: From Basics to Advanced”. Medium.
# python # 开发语言 # 服务器 # 运维 # linux # ssh
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

企业AI智能测试落地实施方案

企业内部搭建一个智能体并不难，难的是让它理解企业自己的业务知识。执行结束后，团队看到的不只是“通过”或“失败”，还可以结合日志、截图、接口数据和历史问题，快速判断问题更可能出在产品、环境、数据还是脚本本身。测吧持续探索大模型、AI Agent、MCP、Skills、Harness等前沿技术，并将其与企业业务场景深度结合，在多个企业项目中不断实践、优化和沉淀，形成了一套成熟的AI实施方法论。我们不仅

avatar AI Agent技术社区
cover

【监控实战】Spring Boot 3.3 + AI Agent × Prometheus：让AI自动发现故障并告警，平均故障响应从30分钟压到2分钟

avatar AI Agent技术社区
cover

【年度总结】用了半年AI Agent，这10条经验让我从怀疑到离不开——附完整工具链

avatar AI Agent技术社区