React集成OpenTelemetry：寻求Vert.x Event Bus通信追踪适配库

前端（React + @vertx/eventbus-bridge-client.js） #

官方没有专门适配该Event Bus客户端的OpenTelemetry插件，需要手动封装客户端结合OpenTelemetry Web SDK实现追踪，核心依赖库包括：

• @opentelemetry/sdk-trace-web：Web端Trace SDK
• @opentelemetry/api：OpenTelemetry核心API
• @opentelemetry/exporter-trace-otlp-http：可选，根据你的追踪后端选择对应Exporter

实现思路与代码示例 #

1. 创建Tracer实例，封装消息发送方法
2. 在发送消息前生成Span，并将W3C Trace Context注入消息头
3. 收到回复或报错后更新Span状态并结束

import { trace } from '@opentelemetry/api';
import { EventBus } from '@vertx/eventbus-bridge-client.js';

const eventBus = new EventBus('/eventbus');
const tracer = trace.getTracer('vertx-eventbus-frontend-tracer');

// 封装带追踪的消息发送方法
export function sendEventBusMessage(address, payload) {
  return tracer.startActiveSpan(`vertx-eventbus-send:${address}`, async (span) => {
    // 注入W3C Trace Context到消息头
    const traceContext = trace.getSpanContext(span.context());
    const messageHeaders = {
      traceparent: `00-${traceContext.traceId}-${traceContext.spanId}-${traceContext.traceFlags.toString(16)}`
    };

    return new Promise((resolve, reject) => {
      eventBus.send(address, { ...payload, headers: messageHeaders }, (err, reply) => {
        if (err) {
          span.setStatus({ code: trace.SpanStatusCode.ERROR, message: err.message });
          reject(err);
        } else {
          span.setStatus({ code: trace.SpanStatusCode.OK });
          resolve(reply);
        }
        span.end();
      });
    });
  });
}

后端（Quarkus + Vert.x Event Bus） #

Quarkus原生支持OpenTelemetry，结合Vert.x扩展可实现端到端追踪，核心依赖库：

• quarkus-opentelemetry：Quarkus OpenTelemetry核心依赖
• quarkus-vertx：Vert.x Event Bus支持

实现思路与代码示例 #

1. 从消息头提取前端传递的Trace Context
2. 创建关联的Span，处理业务逻辑后更新Span状态
3. 回复消息时可将当前Trace Context注入回复头

import io.opentelemetry.api.trace.Span;
import io.opentelemetry.api.trace.Tracer;
import io.opentelemetry.context.Context;
import io.opentelemetry.context.propagation.TextMapGetter;
import io.quarkus.vertx.ConsumeEvent;
import jakarta.inject.Inject;
import io.vertx.core.eventbus.Message;

public class EventBusTraceHandler {

    @Inject
    Tracer tracer;

    @ConsumeEvent("user.service.query")
    public String handleUserQuery(Message<String> message) {
        // 自定义Trace Context提取器
        TextMapGetter<Message<?>> headerGetter = new TextMapGetter<>() {
            @Override
            public String get(Message<?> carrier, String key) {
                return carrier.headers().get(key);
            }

            @Override
            public Iterable<String> keys(Message<?> carrier) {
                return carrier.headers().names();
            }
        };

        // 提取并绑定Trace Context
        Context traceContext = tracer.getPropagators().getTextMapPropagator()
                .extract(Context.current(), message, headerGetter);

        return Context.with(traceContext).wrap(() -> {
            Span span = tracer.spanBuilder("vertx-eventbus-receive:user.service.query")
                    .setParent(traceContext)
                    .startSpan();

            try {
                // 业务逻辑处理
                String result = "User data for: " + message.body();
                span.setStatus(io.opentelemetry.api.trace.StatusCode.OK);
                return result;
            } catch (Exception e) {
                span.setStatus(io.opentelemetry.api.trace.StatusCode.ERROR, e.getMessage());
                throw e;
            } finally {
                span.end();
            }
        }).run();
    }
}

优化建议 #

• 可以通过Vert.x的MessageCodec或全局拦截器统一处理Trace Context的注入与提取，避免重复代码
• 严格遵循W3C Trace Context标准，确保跨服务追踪的兼容性

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