1. 异常嵌套日志的痛点解析
在Python项目开发中，异常嵌套场景几乎无处不在。当外层异常捕获内层异常时，传统的日志记录方式往往存在三个典型问题：
- 信息割裂：内层异常被外层捕获后，原始堆栈信息可能被覆盖
- 日志冗余：同一异常在不同层级被重复记录
- 上下文丢失：异常发生时的关键变量状态未被保存
try: try: 1/0 # 内层异常 except Exception as e: logger.error("内层错误") # 丢失原始堆栈 raise ProcessError("处理失败") except Exception as e: logger.error(e) # 只记录最外层异常2. 结构化日志解决方案
2.1 日志格式标准化
推荐使用logging.Formatter构建包含以下字段的JSON日志：
{ "timestamp": "%(asctime)s", "level": "%(levelname)s", "path": "%(pathname)s", "line": "%(lineno)d", "trace_id": "%(trace_id)s", # 自定义字段 "exception": { "type": exc_type.__name__, "message": str(exc_value), "stack": traceback.format_exc() }, "context": {} # 自定义上下文 }2.2 异常上下文捕获器
通过装饰器自动捕获上下文变量：
def capture_context(*args): def decorator(func): @functools.wraps(func) def wrapper(*args, **kwargs): frame = inspect.currentframe() try: locals_before = frame.f_back.f_locals result = func(*args, **kwargs) return result except Exception as e: context = {k: locals_before.get(k) for k in args} logger.error("Context captured", extra={"context": context}, exc_info=True) raise finally: del frame return wrapper return decorator3. 嵌套异常处理最佳实践
3.1 异常链式记录
使用raise from保留原始异常：
try: db_operation() except DBError as e: logger.error("Database operation failed", exc_info=True) raise ProcessError("Processing failed") from e # 保留异常链3.2 日志去重机制
通过异常指纹实现重复检测：
def get_exception_fingerprint(exc): tb = traceback.extract_tb(exc.__traceback__) return hash(frozenset( (frame.filename, frame.lineno, frame.line) for frame in tb ))4. 实战：分布式系统日志追踪
4.1 请求链路追踪
集成OpenTelemetry实现跨服务追踪：
from opentelemetry import trace tracer = trace.get_tracer(__name__) def process_request(request): with tracer.start_as_current_span("request_processing") as span: try: result = risky_operation() span.set_status(Status(StatusCode.OK)) return result except Exception as e: span.record_exception(e) span.set_status(Status(StatusCode.ERROR)) raise4.2 日志聚合分析
ELK Stack配置示例：
class ELKHandler(logging.Handler): def emit(self, record): log_entry = self.format(record) requests.post( "http://elk:9200/logs/_doc", json=json.loads(log_entry), timeout=1 )5. 性能优化技巧
5.1 延迟日志格式化
使用logging.debug()的惰性求值特性：
logger.debug("SQL executed: %s", query) # 仅当级别达标时格式化5.2 异步日志处理
使用ConcurrentLogHandler：
from cloghandler import ConcurrentRotatingFileHandler handler = ConcurrentRotatingFileHandler( "app.log", maxBytes=100e6, backupCount=5 )6. 调试辅助工具
6.1 交互式调试日志
集成IPython调试器：
def log_and_debug(msg): logger.error(msg) from IPython import embed; embed()6.2 可视化异常图谱
使用pyvis生成异常传播图：
def visualize_exception_chain(exc): net = Network() current = exc while current: net.add_node(id(current), label=type(current).__name__) if current.__cause__: net.add_edge(id(current), id(current.__cause__), label="cause") current = current.__cause__ net.show("exception.html")关键提示：生产环境应确保日志异步写入，避免阻塞主线程。建议日志量超过100条/秒时启用队列缓冲机制。
