Python之FastAPI的入门到精通系列：如何给业务添加trace id, 标记和串联完整业务请求链路
在代码和分布式系统中，trace id（追踪 ID） 是一个全局唯一的标识符，用于标记和串联一次完整的用户请求（或事务）在分布式系统中经过的所有服务和操作。
把它想象成一次探险旅程的 “护照号”，无论这次旅程经过多少个国家（服务）、换乘多少次交通工具（API 调用），这个唯一的号码都能标识出这是同一次完整的旅程。
核心意义：Trace ID
Trace ID 作为分布式系统的"神经系统"，它的存在解决了微服务架构中最根本的问题:在请求跨越多个服务时,如何将碎片化的日志和调用关系串联起来,还原完整的执行路径。
问题定位和分析
在传统单体应用中,一个请求的执行路径清晰可见。但在分布式系统中,一次用户请求可能触发几十个服务的调用。当出现问题时,Trace ID 能够:
快速收敛问题范围: 通过 Trace ID 立即找到所有相关日志,避免在海量日志中盲目搜索
重建调用链路: 清晰展示请求经过哪些服务、在哪个环节出错、耗时分布如何
跨团队沟通和协作: 不同团队维护不同服务时,Trace ID 成为统一的沟通语言
例如:用户报告支付失败,通过 Trace ID 可以追踪到是订单服务超时导致,进而发现是库存服务数据库慢查询引起,最终定位到某个未优化的 SQL。
性能分析和优化
Trace ID 配合时间戳能够构建完整的性能画像:
识别性能瓶颈: 发现哪个服务、哪个方法消耗了大部分时间
分析依赖关系: 理解服务间的调用依赖和并行/串行模式
优化决策支持: 基于真实数据决定是否需要缓存、异步化或服务降级
例如:发现某个接口 P99 耗时 2 秒,通过 Trace ID 分析发现其中 1.5 秒耗费在调用第三方服务,可以考虑引入缓存或超时控制。
全链路监控和可观测性
Trace ID 是实现系统可观测性的三大支柱(日志、指标、追踪)中追踪的基础。
聚合Trace数据，可以容量规划与成本优化，合理评估服务负载和峰值调度。
在混沌实验和压力测试等场景，可以为测试模型构建提供可靠架构信息。
Trace ID标准格式定义
在OpenTelemetry中，trace_id有严格的标准格式定义，以下是核心要点：
from opentelemetry.trace import TraceId
# 从十六进制字符串转换为二进制
trace_id_bytes = TraceId.from_hex("0af7651916cd43dd8448eb211c80319c")
# 从二进制转换为十六进制字符串
trace_id_hex = trace_id_bytes.to_hex()
在FastAPI添加trace_id实现可观测链路追踪
- 创建 TracerProvider：这是 OpenTelemetry 的核心组件，
负责管理所有的追踪数据和 span（跨度） - 设置服务标识：通过 Resource.create() 创建资源对象，将当前服务的名称设置为 settings.PROJECT_NAME（即配置中的项目名）
- 全局配置：将这个追踪提供者设置为全局默认，所有后续的追踪操作都会使用这个提供者
trace.set_tracer_provider(
TracerProvider(
resource=Resource.create(
attributes={"service.name": settings.PROJECT_NAME},
)
)
)
添加Middleware
FastAPI/Starlette 的中间件遵循后进先出的执行原则：
- 最后添加的中间件，在请求处理时会最先执行
- 最先添加的中间件，在请求处理时会最后执行
TraceMiddleware 的作用：生成根 Trace ID、初始化上下文，需要在所有其他处理逻辑（包括路由处理、其他中间件、OpenTelemetry 自动插桩）之前执行，以确保整个请求链路都能获取到正确的 Trace 上下文。
# trace_middleware 最后添加，最先执行
app.add_middleware(TraceMiddleware)
FastAPIInstrumentor.instrument_app(app, trace.get_tracer_provider())
app.include_router(api_router, prefix=settings.API_V1_STR)
链路插桩
FastAPIInstrumentor.instrument_app(app, trace.get_tracer_provider()) 的作用是 自动为 FastAPI 应用进行 OpenTelemetry 分布式追踪插桩。
FastAPIInstrumentor.instrument_app的作用是自动为路由函数、依赖注入等添加 Span 追踪，它本质上是通过修改路由处理函数的包装逻辑来实现的。
# 自动捕获 HTTP 请求/响应：无需手动添加追踪代码
# 自动创建 Span：为每个 API 端点自动创建追踪跨度
# 自动收集性能指标：如请求持续时间、状态码等
# FastAPIInstrumentor：负责自动创建和管理 Spans
# TraceMiddleware：负责传播 trace 上下文到响应头
FastAPIInstrumentor.instrument_app(app, trace.get_tracer_provider())
定义Middleware
在fastapi里，中间件是异步执行的! inject(response.headers)，注入 trace 上下文到 header，方便下游服务获取trace_id。
"""
分布式链路追踪中间件
此模块实现了 OpenTelemetry 分布式追踪功能，用于在 FastAPI 应用中
自动传播链路上下文信息，支持微服务架构下的全链路监控。
"""
from app.utility.trace import ctxvar_trace_id
class TraceMiddleware(BaseHTTPMiddleware):
"""
分布式链路追踪中间件
继承自 BaseHTTPMiddleware，用于在 HTTP 请求处理过程中自动注入OpenTelemetry trace 上下文信息到响应头中，实现链路追踪功能。
主要作用：
- 拦截所有进入的 HTTP 请求
- 在响应返回前注入 trace 上下文到响应头
- 支持分布式系统的链路追踪和问题排查
"""
def __init__(self, app):
"""
初始化 TraceMiddleware
Args:
app: ASGI 应用实例（FastAPI 或 Starlette 应用）
"""
super().__init__(app)
async def dispatch(self, request: Request, call_next):
"""
处理 HTTP 请求并注入 trace 上下文
这是中间件的核心分发方法，负责：
1. 接收并处理 HTTP 请求
2. 执行实际的业务逻辑处理
3. 在响应中注入 trace 上下文信息
4. 返回带有追踪信息的响应
Args:
request: Starlette Request 对象，包含 HTTP 请求信息
call_next: 下一个处理程序的调用函数
Returns:
response: 带有 trace 上下文信息的响应对象
"""
# 获取当前活跃的 span（追踪单元）
# span 是 OpenTelemetry 中的基本追踪单元，代表一个操作或请求
current_span = trace.get_current_span()
# 从当前 span 中提取 span 上下文信息
# 包含 trace_id（调用链唯一标识）和 span_id（当前单元标识）
ctx = current_span.get_span_context()
# 将二进制格式的 trace_id 转换为十六进制字符串格式
trace_id = format_trace_id(ctx.trace_id)
# 记录当前的 trace_id，便于调试和监控
# 使用 debug 级别避免对生产环境性能造成影响
logger.debug(f"TraceMiddleware: {ctx.trace_id}")
# 将 trace_id 设置到上下文变量中，供后续业务逻辑使用
# 这样在处理请求的任意位置都可以通过 ctxvar_trace_id.get() 获取 trace_id
token_trace_id = ctxvar_trace_id.set(trace_id)
# 调用下一个中间件或路由处理器执行实际的业务逻辑
# 这里会处理 HTTP 请求并生成响应
response = await call_next(request)
# 将 OpenTelemetry trace 上下文信息注入到 HTTP 响应头中
# 确保下游服务或客户端能够从响应头中提取追踪信息，继续链路追踪
inject(response.headers) # 注入 trace 上下文到 header
return response
验证是否生效
2025-11-20 21:47:14.548 | DEBUG | app.api.api_v1.endpoint.user:read_users:28 - 验证trace_id | trace_id=a4d0473b2a96890c345bc1ef194b9b92
OpenTelemetry 分布式追踪装饰器
在函数调用的过程中，自动创建并添加trace_id
为函数调用添加追踪功能
from functools import wraps
from inspect import iscoroutinefunction
from typing import Callable, Optional
from opentelemetry import trace
ctxvar_trace_id = ContextVar("trace_id", default="")
def trace_decorator(span_name: Optional[str] = None):
"""
OpenTelemetry 分布式追踪装饰器，具体功能如下：
核心功能
- 分布式追踪：为函数调用添加追踪功能，记录函数的执行时间、参数和返回值
- 异步支持：自动区分同步函数和异步函数，提供不同的包装器
- 可配置追踪名称：可以自定义追踪 span 的名称
装饰器工厂模式：
trace_decorator(span_name: Optional[str] = None) 接受一个可选的 span 名称
返回实际的装饰器函数
动态包装器选择：
使用 iscoroutinefunction() 检查被装饰函数是否为异步函数
为异步函数使用 async_wrapper
为同步函数使用 sync_wrapper
追踪实现：
获取 OpenTelemetry tracer
使用传入的 span 名称或自动生成名称（格式：{module}.{function_name}）
在函数执行期间创建并管理 span
"""
def decorator(func: Callable):
@wraps(func)
async def async_wrapper(*args, **kwargs):
tracer = trace.get_tracer(__name__)
name = span_name or f"{func.__module__}.{func.__name__}"
with tracer.start_as_current_span(name):
return await func(*args, **kwargs)
@wraps(func)
def sync_wrapper(*args, **kwargs):
tracer = trace.get_tracer(__name__)
name = span_name or f"{func.__module__}.{func.__name__}"
with tracer.start_as_current_span(name):
return func(*args, **kwargs)
return async_wrapper if iscoroutinefunction(func) else sync_wrapper
return decorator
使用方法
# 自动使用函数模块和名称作为追踪名称
@trace_decorator()
async def my_async_function():
pass
# 自定义追踪名称
@trace_decorator("custom_operation_name")
def my_sync_function():
pass
⏰ 刚刷到的朋友注意啦！
点击【关注】锁定宝藏库，从此升职加薪不迷路 ✨
☁ 主机显示特惠：只要80元（3TB流量，1vcpu，50GB硬盘）
购买地址：https://my.racknerd.com/aff.php?aff=14942
📢 腾讯云资源限时福利
有云服务器、CDN、对象存储、网络防护等需求的朋友，欢迎联系下方腾讯云官方销售 👇
✔️ 内部专属折扣，价格更优
✔️ 量大可谈，支持定制方案
✔️ 技术咨询与售后无忧
评论区