非API队列Worker应用如何基于OpenTelemetry实现链路追踪？

完全适合用OpenTelemetry，它不止支持HTTP API，消息队列消费、后台任务这类无请求链路的场景同样能覆盖，而且能帮你实现消息处理的全链路追踪、性能统计。

具体实现步骤 #

1. 复用基础OpenTelemetry初始化逻辑 #

你之前HTTP场景的初始化代码可以直接沿用，要确保在应用启动最开始执行，这样后续线程里的Tracer能获取到正确的Provider：

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
from your_config_module import config  # 替换成你的配置模块

# 初始化TracerProvider和Exporter
otlp_exporter = OTLPSpanExporter(endpoint=config('OPTL_ENDPOINT'))
span_processor = BatchSpanProcessor(otlp_exporter)

trace.set_tracer_provider(
    TracerProvider(resource=Resource.create({"service.name": config('OPTL_SERVICE_NAME')}))
)
trace.get_tracer_provider().add_span_processor(span_processor)

2. 手动为消息消费流程埋点创建Span #

因为没有自动 instrumentation 工具，需要在start_consume方法里手动创建Span，覆盖消息接收、处理的全流程：

from opentelemetry import trace

class SQSConsumer:
    def __init__(self):
        self.tracer = trace.get_tracer("sqs-consumer-tracer")  # 自定义Tracer标识名称

    def start_consume(self):
        while True:
            # 创建消息消费根Span，标记整个消息处理周期
            with self.tracer.start_as_current_span("sqs-message-consume") as span:
                # 替换为你实际的消息接收逻辑
                message = self.receive_message()
                if not message:
                    continue

                # 为Span添加关键属性，便于后续分析：消息ID、接收时间戳等
                span.set_attribute("sqs.message_id", message.message_id)
                span.set_attribute("sqs.receive_timestamp", message.attributes.get('SentTimestamp'))

                # 创建子Span单独记录业务处理环节
                with self.tracer.start_as_current_span("message-business-process") as process_span:
                    try:
                        self.process_message(message)
                        process_span.set_attribute("process.status", "success")
                    except Exception as e:
                        process_span.set_attribute("process.status", "failed")
                        process_span.record_exception(e)  # 记录异常详情
                        # 这里可根据业务需求添加异常处理逻辑，比如消息重新入队

3. 用Metrics统计消息接收与处理数量 #

如果需要统计消息接收总数、成功/失败数，可以添加OpenTelemetry Metrics的初始化和埋点：

from opentelemetry import metrics
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter

# 初始化Metrics Provider，放在应用启动处
metric_reader = PeriodicExportingMetricReader(OTLPMetricExporter(endpoint=config('OPTL_ENDPOINT')))
metrics.set_meter_provider(MeterProvider(resource=Resource.create({"service.name": config('OPTL_SERVICE_NAME')})))
meter = metrics.get_meter("sqs-consumer-meter")

# 定义计数器：消息接收总数
message_counter = meter.create_counter(
    name="sqs.messages.received",
    description="Total number of SQS messages received",
    unit="1"
)
# 定义计数器：按状态统计处理完成的消息数
process_status_counter = meter.create_counter(
    name="sqs.messages.processed",
    description="Number of processed SQS messages by status",
    unit="1"
)

# 在消费逻辑中更新计数器
class SQSConsumer:
    def start_consume(self):
        while True:
            with self.tracer.start_as_current_span("sqs-message-consume") as span:
                message = self.receive_message()
                if not message:
                    continue
                # 统计消息接收量
                message_counter.add(1, attributes={"queue_name": "your-target-queue"})
                
                try:
                    self.process_message(message)
                    process_status_counter.add(1, attributes={"status": "success", "queue_name": "your-target-queue"})
                except Exception as e:
                    process_status_counter.add(1, attributes={"status": "failed", "queue_name": "your-target-queue"})
                    # 异常处理逻辑

4. 线程上下文注意事项 #

Python线程默认会继承父线程的OpenTelemetry上下文，所以只要主线程初始化了TracerProvider，子线程里的get_tracer就能正常工作，无需额外处理上下文传递。如果后续用到线程池或异步任务，再考虑手动传递上下文即可。

关键提示 #

• Span命名要贴合业务场景，根Span标记整体消费流程，子Span对应具体业务步骤，这样在链路追踪系统中能清晰看到单条消息的处理路径。
• 给Span添加有价值的属性（如消息ID、队列名、处理状态），方便后续排查问题和统计分析。
• Tracer和Metrics可以配合使用：Tracer看单条消息的处理细节，Metrics看整体吞吐量、成功率等宏观指标。

内容的提问来源于stack exchange，提问作者AlexPauloDev