分布式链路追踪原理与实战

// 手动埋点示例
Tracer tracer = ...;
Span span = tracer.buildSpan("mysql-query").start();
try {
    span.logEvent("query-start");
    // 执行 SQL 查询
    span.tag("db.statement", "SELECT * FROM orders WHERE id = ?");
    span.tag("db.type", "mysql");
    span.logEvent("query-end");
} catch (Exception e) {
    span.log(ObjectWrapper.of(e));
    span.setTag("error", true);
} finally {
    span.finish();
}

// OpenTracing API 示例
Tracer tracer = ...;

// 创建根 Span
Span parentSpan = tracer.buildSpan("http-request")
    .withTag("http.method", "GET")
    .withTag("http.url", "/api/users")
    .start();

// 创建子 Span
Span childSpan = tracer.buildSpan("database-query")
    .asChildOf(parentSpan)
    .withTag("db.type", "mysql")
    .withTag("db.statement", "SELECT ...")
    .start();

try {
    // 业务逻辑
    childSpan.finish();
} catch (Exception e) {
    childSpan.log(ImmutableMap.of("event", "error", "error.object", e));
    parentSpan.setTag("error", true);
} finally {
    parentSpan.finish();
}

// 注入到 HTTP Header
Tracer tracer = ...;
Span span = ...;
SpanContext spanContext = span.context();

// 注入到 carrier（这里是 HttpHeaders）
HttpHeaders headers = new HttpHeaders();
tracer.inject(spanContext, Format.Builtin.HTTP_HEADERS, new HttpHeadersCarrier(headers));

// 在下游服务中提取
SpanContext extractedContext = tracer.extract(
    Format.Builtin.HTTP_HEADERS,
    new HttpHeadersCarrier(headers)
);
Span childSpan = tracer.buildSpan("downstream-call")
    .asChildOf(extractedContext)
    .start();

import org.apache.skywalking.apm.toolkit.trace.ActiveSpan;
import org.apache.skywalking.apm.toolkit.trace.Trace;

@Trace
@Tag(key = "orderId", value = "${args[0].getOrderId()}")
public Order createOrder(OrderRequest request) {
    // 业务逻辑
    
    // 在追踪中记录异常
    ActiveSpan.error("创建订单失败");
    ActiveSpan.log(e);
    
    return order;
}

@RestController
public class OrderController {

    @Autowired
    private Tracer tracer;

    @PostMapping("/orders")
    public ResponseEntity<Order> createOrder(@RequestBody OrderRequest request) {
        // 在 Controller 层创建 Span
        Span span = tracer.buildSpan("POST /orders")
            .withTag("http.method", "POST")
            .withTag("http.url", "/orders")
            .withTag("user.id", request.getUserId())
            .start();

        try (Scope scope = tracer.activateSpan(span)) {
            Order order = orderService.createOrder(request);
            span.tag("http.status_code", "200");
            return ResponseEntity.ok(order);
        } catch (Exception e) {
            span.setTag("error", true);
            span.log(ImmutableMap.of("event", "error", "error.kind", e.getClass().getName()));
            throw e;
        } finally {
            span.finish();
        }
    }
}

@Service
public class OrderServiceImpl {

    @Autowired
    private Tracer tracer;

    @Autowired
    private UserClient userClient;

    public Order createOrder(OrderRequest request) {
        Span span = tracer.buildSpan("call-user-service")
            .withTag("rpc.system", "grpc")
            .withTag("rpc.method", "getUser")
            .start();

        try (Scope scope = tracer.activateSpan(span)) {
            // 注入 SpanContext 到 RPC 请求 Header
            SpanContext spanContext = span.context();
            Metadata metadata = new Metadata();
            tracer.inject(spanContext, Format.Builtin.HTTP_HEADERS,
                new RequestHeaderCarrier(metadata));

            User user = userClient.getUser(request.getUserId(), metadata);
            span.tag("rpc.user.id", user.getId());
            return orderMapper.create(request);
        } finally {
            span.finish();
        }
    }
}

@Repository
public class OrderDao {

    @Autowired
    private Tracer tracer;

    @Autowired
    private JdbcTemplate jdbcTemplate;

    public List<Order> findByUserId(Long userId) {
        Span span = tracer.buildSpan("db:select:orders")
            .withTag("db.type", "mysql")
            .withTag("db.instance", "ecommerce")
            .withTag("db.statement", "SELECT * FROM orders WHERE user_id = ?")
            .start();

        try (Scope scope = tracer.activateSpan(span)) {
            List<Order> orders = jdbcTemplate.query(
                "SELECT * FROM orders WHERE user_id = ?",
                new Object[]{userId},
                new OrderRowMapper()
            );
            span.tag("db.rows", orders.size());
            return orders;
        } finally {
            span.finish();
        }
    }
}

@SpanName("create-order")
class OrderService {

    @NewSpan
    @SpanTag(key = "order.type")
    public void createOrder(@SpanTag("order.id") String orderId) {
        // 自动创建名为 "create-order" 的 Span
        // 自动记录 orderId 作为标签
    }
}

import com.twitter.service.snowflake.IdWorker;

long traceId = IdWorker.getInstance().nextId();

public static String generateTraceId() {
    // 格式：时间戳-随机字符（Base62 编码）
    long timestamp = System.currentTimeMillis() * 1000;
    Random random = new Random();
    String randomPart = Base62.encode(random.nextLong());
    return timestamp + "-" + randomPart;
}

// 发送方
Span span = tracer.activeSpan();
SpanContext spanContext = span.context();
Map<String, String> metadata = new HashMap<>();
tracer.inject(spanContext, Format.Builtin.TEXT_MAP, new TextMapAdapter(metadata));
rpcCall.setMetadata(metadata);

// 接收方
TextMapExtractAdapter carrier = new TextMapExtractAdapter(receivedMetadata);
SpanContext parentContext = tracer.extract(Format.Builtin.TEXT_MAP, carrier);
Span childSpan = tracer.buildSpan("rpc-handler")
    .asChildOf(parentContext)
    .start();

// 生产者：注入 TraceId 到消息 Header
ProducerRecord<String, String> record = new ProducerRecord<>("topic", "message");
Span span = tracer.activeSpan();
SpanContext spanContext = span.context();
record.headers().add("X-Trace-Id", spanContext.traceId().getBytes());
kafkaTemplate.send(record);

// 消费者：从消息 Header 提取 TraceId
ConsumerRecord<String, String> record = ...;
SpanContext spanContext = tracer.extract(
    Format.Builtin.TEXT_MAP,
    new RecordHeadersCarrier(record.headers())
);
Span span = tracer.buildSpan("kafka-consume")
    .asChildOf(spanContext)
    .start();

import org.springframework.cloud.sleuth.Span;
import org.springframework.cloud.sleuth.Tracer;

@Service
public class OrderServiceImpl {

    @Autowired
    private Tracer tracer;

    public Order processOrder(OrderRequest request) {
        Span span = tracer.nextSpan().name("processOrder").start();
        
        try (Tracer.SpanInScope scope = tracer.withSpan(span)) {
            span.tag("order.type", request.getType());
            span.tag("user.id", String.valueOf(request.getUserId()));
            
            // 业务逻辑
            Order order = createOrder(request);
            
            span.tag("order.id", order.getId());
            span.tag("order.amount", String.valueOf(order.getAmount()));
            
            return order;
        } catch (Exception e) {
            span.error(e);
            throw e;
        } finally {
            span.end();
        }
    }
}

// 使用 Brave 手动埋点
Tracing tracing = Tracing.newBuilder()
    .localServiceName("payment-service")
    .sampler(Sampler.alwaysTrue())
    .spanReporter(ZipkinSpanReporter.create("http://localhost:9411/api/v2/spans"))
    .build();

Tracer tracer = tracing.getTracer();

// 在方法中创建 Span
Span span = tracer.newTrace().name("processPayment").start();
try (Scope scope = tracer.withSpanInScope(span)) {
    span.tag("payment.method", "alipay");
    span.annotate("支付回调开始处理");
    
    // 业务逻辑
    paymentService.processCallback();
    
    span.annotate("支付回调处理完成");
} finally {
    span.finish();
}