【系统架构设计百科】延迟分析：从 P50 到 P999 的全链路追踪

// percentile 计算给定百分位的延迟值
// latencies 必须已排序
func percentile(latencies []float64, p float64) float64 {
    if len(latencies) == 0 {
        return 0
    }
    rank := p / 100.0 * float64(len(latencies)-1)
    lower := int(rank)
    upper := lower + 1
    if upper >= len(latencies) {
        return latencies[len(latencies)-1]
    }
    weight := rank - float64(lower)
    return latencies[lower]*(1-weight) + latencies[upper]*weight
}

func main() {
    // 模拟 1000 个请求的延迟（毫秒）
    latencies := make([]float64, 1000)
    for i := 0; i < 999; i++ {
        latencies[i] = 1.0
    }
    latencies[999] = 10000.0 // 一个 10 秒的慢请求

    sort.Float64s(latencies)

    fmt.Printf("P50:  %.2f ms\n", percentile(latencies, 50))   // 1.00
    fmt.Printf("P99:  %.2f ms\n", percentile(latencies, 99))   // 1.00
    fmt.Printf("P999: %.2f ms\n", percentile(latencies, 99.9)) // 10000.00
    fmt.Printf("Avg:  %.2f ms\n", avg(latencies))              // 10.999
}

package main

import (
    "context"
    "fmt"
    "time"

    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/attribute"
    "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
    "go.opentelemetry.io/otel/sdk/resource"
    sdktrace "go.opentelemetry.io/otel/sdk/trace"
    semconv "go.opentelemetry.io/otel/semconv/v1.17.0"
    "go.opentelemetry.io/otel/trace"
)

func initTracer() (*sdktrace.TracerProvider, error) {
    exporter, err := otlptracegrpc.New(
        context.Background(),
        otlptracegrpc.WithEndpoint("otel-collector:4317"),
        otlptracegrpc.WithInsecure(),
    )
    if err != nil {
        return nil, fmt.Errorf("creating exporter: %w", err)
    }

    tp := sdktrace.NewTracerProvider(
        sdktrace.WithBatcher(exporter),
        sdktrace.WithResource(resource.NewWithAttributes(
            semconv.SchemaURL,
            semconv.ServiceName("search-service"),
            semconv.ServiceVersion("1.0.0"),
        )),
        // 生产环境通常不全量采样，按比例或按尾延迟采样
        sdktrace.WithSampler(sdktrace.ParentBased(
            sdktrace.TraceIDRatioBased(0.01), // 1% 采样
        )),
    )
    otel.SetTracerProvider(tp)
    return tp, nil
}

func handleSearch(ctx context.Context, query string) ([]Result, error) {
    tracer := otel.Tracer("search")
    ctx, span := tracer.Start(ctx, "handleSearch",
        trace.WithAttributes(attribute.String("query", query)),
    )
    defer span.End()

    // 阶段 1：解析查询
    ctx, parseSpan := tracer.Start(ctx, "parseQuery")
    parsed := parseQuery(query)
    parseSpan.End()

    // 阶段 2：并行查询分片
    ctx, fanoutSpan := tracer.Start(ctx, "fanoutShards",
        trace.WithAttributes(attribute.Int("shard.count", 30)),
    )
    shardResults := fanoutToShards(ctx, parsed, 30)
    fanoutSpan.End()

    // 阶段 3：合并排序
    ctx, mergeSpan := tracer.Start(ctx, "mergeResults")
    results := mergeAndRank(shardResults)
    mergeSpan.SetAttributes(attribute.Int("result.count", len(results)))
    mergeSpan.End()

    return results, nil
}

import io.opentelemetry.api.GlobalOpenTelemetry;
import io.opentelemetry.api.trace.Span;
import io.opentelemetry.api.trace.Tracer;
import io.opentelemetry.context.Scope;

public class OrderService {
    private static final Tracer tracer =
        GlobalOpenTelemetry.getTracer("order-service", "1.0.0");

    public OrderResult placeOrder(OrderRequest request) {
        Span span = tracer.spanBuilder("placeOrder")
            .setAttribute("order.item_count", request.getItems().size())
            .startSpan();

        try (Scope scope = span.makeCurrent()) {
            // 库存检查
            Span inventorySpan = tracer.spanBuilder("checkInventory").startSpan();
            try (Scope s = inventorySpan.makeCurrent()) {
                checkInventory(request.getItems());
            } finally {
                inventorySpan.end();
            }

            // 支付处理
            Span paymentSpan = tracer.spanBuilder("processPayment").startSpan();
            try (Scope s = paymentSpan.makeCurrent()) {
                processPayment(request.getPaymentInfo());
            } finally {
                paymentSpan.end();
            }

            // 订单持久化
            Span persistSpan = tracer.spanBuilder("persistOrder").startSpan();
            try (Scope s = persistSpan.makeCurrent()) {
                return persistOrder(request);
            } finally {
                persistSpan.end();
            }
        } catch (Exception e) {
            span.recordException(e);
            throw e;
        } finally {
            span.end();
        }
    }
}

# wrk：会产生协调省略，尾延迟数据不可信
wrk -t4 -c100 -d30s http://localhost:8080/api/search

# wrk2：恒定速率发送，延迟数据准确
wrk2 -t4 -c100 -d30s -R10000 http://localhost:8080/api/search
#                     ^^^^^^^ 恒定速率：10000 req/s

public class ConstantRateLoadGenerator {
    private final long intervalNanos;
    private final HdrHistogram histogram;

    public ConstantRateLoadGenerator(double targetRps) {
        this.intervalNanos = (long) (1_000_000_000.0 / targetRps);
        this.histogram = new HdrHistogram(1, 60_000_000, 3); // 1us - 60s
    }

    public void run(Duration duration) {
        long startTime = System.nanoTime();
        long intendedSendTime = startTime;
        long endTime = startTime + duration.toNanos();

        while (intendedSendTime < endTime) {
            long actualSendTime = System.nanoTime();

            // 在 intendedSendTime 之前不发送
            while (System.nanoTime() < intendedSendTime) {
                Thread.onSpinWait();
            }

            long responseTime = sendRequest(); // 返回纳秒级延迟

            // 关键：延迟 = 响应返回时间 - 预期发送时间
            // 而不是 响应返回时间 - 实际发送时间
            long latency = (System.nanoTime() - intendedSendTime);
            histogram.recordValue(latency / 1000); // 转为微秒

            intendedSendTime += intervalNanos;
        }
    }
}

import org.HdrHistogram.Histogram;

public class LatencyRecorder {
    // 范围：1 微秒到 1 小时，精度 3 位有效数字
    private final Histogram histogram =
        new Histogram(1, 3_600_000_000L, 3);

    public void recordLatency(long latencyMicros) {
        histogram.recordValue(latencyMicros);
    }

    public void printReport() {
        System.out.printf("P50:   %d us%n", histogram.getValueAtPercentile(50));
        System.out.printf("P90:   %d us%n", histogram.getValueAtPercentile(90));
        System.out.printf("P99:   %d us%n", histogram.getValueAtPercentile(99));
        System.out.printf("P999:  %d us%n", histogram.getValueAtPercentile(99.9));
        System.out.printf("P9999: %d us%n", histogram.getValueAtPercentile(99.99));
        System.out.printf("Max:   %d us%n", histogram.getMaxValue());
        System.out.printf("Mean:  %.2f us%n", histogram.getMean());
        System.out.printf("Count: %d%n", histogram.getTotalCount());
    }

    public void mergeFrom(Histogram other) {
        histogram.add(other);
    }

    public void reset() {
        histogram.reset();
    }
}

package main

import (
    "fmt"
    "math/rand"
    "time"

    "github.com/HdrHistogram/hdrhistogram-go"
)

func main() {
    // 范围：1 微秒到 10 秒，精度 3 位有效数字
    hist := hdrhistogram.New(1, 10_000_000, 3)

    // 模拟记录 100 万个延迟样本
    rng := rand.New(rand.NewSource(time.Now().UnixNano()))
    for i := 0; i < 1_000_000; i++ {
        // 模拟：大部分请求 1-5ms，少量请求 50-500ms
        latencyUs := int64(1000 + rng.ExpFloat64()*2000)
        if rng.Float64() < 0.01 { // 1% 的请求慢 10 倍以上
            latencyUs *= 10
        }
        hist.RecordValue(latencyUs)
    }

    fmt.Printf("P50:   %d us\n", hist.ValueAtQuantile(50))
    fmt.Printf("P90:   %d us\n", hist.ValueAtQuantile(90))
    fmt.Printf("P99:   %d us\n", hist.ValueAtQuantile(99))
    fmt.Printf("P999:  %d us\n", hist.ValueAtQuantile(99.9))
    fmt.Printf("Max:   %d us\n", hist.Max())
    fmt.Printf("Mean:  %.2f us\n", hist.Mean())

    // HDR Histogram 支持导出为百分位分布表
    brackets := hist.CumulativeDistribution()
    for _, b := range brackets {
        if b.Quantile >= 90 {
            fmt.Printf("  %.4f%%: %d us\n", b.Quantile, b.ValueAt)
        }
    }
}

// hedgedRequest 在超时后向备用副本发送对冲请求
func hedgedRequest(
    ctx context.Context,
    primary string,
    replicas []string,
    hedgeDelay time.Duration,
) (*Response, error) {
    type result struct {
        resp *Response
        err  error
    }

    ctx, cancel := context.WithCancel(ctx)
    defer cancel()

    ch := make(chan result, len(replicas)+1)

    // 向主副本发送请求
    go func() {
        resp, err := sendRequest(ctx, primary)
        ch <- result{resp, err}
    }()

    // 设置对冲定时器
    hedgeTimer := time.NewTimer(hedgeDelay)
    defer hedgeTimer.Stop()

    outstanding := 1

    for {
        select {
        case r := <-ch:
            if r.err == nil {
                return r.resp, nil // 拿到第一个成功结果就返回
            }
            outstanding--
            if outstanding == 0 {
                return nil, r.err
            }

        case <-hedgeTimer.C:
            // 主副本超时，向一个备用副本发送对冲请求
            if len(replicas) > 0 {
                replica := replicas[0]
                replicas = replicas[1:]
                outstanding++
                go func() {
                    resp, err := sendRequest(ctx, replica)
                    ch <- result{resp, err}
                }()
            }

        case <-ctx.Done():
            return nil, ctx.Err()
        }
    }
}

// tiedRequest 同时发送请求到两个副本，先开始执行的通知另一个取消
func tiedRequest(
    ctx context.Context,
    replica1, replica2 string,
) (*Response, error) {
    ctx, cancel := context.WithCancel(ctx)
    defer cancel()

    // 生成唯一请求 ID，两个副本用同一个 ID
    requestID := generateRequestID()

    type result struct {
        resp *Response
        err  error
    }
    ch := make(chan result, 2)

    go func() {
        resp, err := sendTiedRequest(ctx, replica1, requestID)
        ch <- result{resp, err}
    }()
    go func() {
        resp, err := sendTiedRequest(ctx, replica2, requestID)
        ch <- result{resp, err}
    }()

    // 取先返回的结果
    r := <-ch
    if r.err == nil {
        return r.resp, nil
    }
    // 第一个失败了，等第二个
    r = <-ch
    return r.resp, r.err
}

use std::time::{Duration, Instant};

/// 延迟预算：表示一个请求还剩多少时间可用
#[derive(Clone, Debug)]
pub struct LatencyBudget {
    deadline: Instant,
}

impl LatencyBudget {
    /// 创建一个新的延迟预算
    pub fn new(total_budget: Duration) -> Self {
        Self {
            deadline: Instant::now() + total_budget,
        }
    }

    /// 从传入的截止时间创建（跨服务传播时使用）
    pub fn from_deadline(deadline: Instant) -> Self {
        Self { deadline }
    }

    /// 剩余预算
    pub fn remaining(&self) -> Duration {
        self.deadline.saturating_duration_since(Instant::now())
    }

    /// 预算是否已耗尽
    pub fn is_exhausted(&self) -> bool {
        Instant::now() >= self.deadline
    }

    /// 分配子预算（为下游服务留出时间）
    pub fn allocate(&self, amount: Duration) -> Option<LatencyBudget> {
        let remaining = self.remaining();
        if amount > remaining {
            return None; // 预算不足
        }
        Some(LatencyBudget {
            deadline: Instant::now() + amount,
        })
    }
}

/// 请求处理示例
async fn handle_product_detail(budget: LatencyBudget, product_id: &str) -> Result<ProductDetail, Error> {
    if budget.is_exhausted() {
        return Err(Error::BudgetExhausted);
    }

    // 串行调用：商品信息服务（分配 50ms）
    let product_budget = budget.allocate(Duration::from_millis(50))
        .ok_or(Error::BudgetExhausted)?;
    let product = fetch_product_info(product_budget, product_id).await?;

    if budget.is_exhausted() {
        return Err(Error::BudgetExhausted);
    }

    // 并行调用：价格、库存、推荐（剩余预算的 70%，留 30% 给聚合）
    let parallel_budget_ms = (budget.remaining().as_millis() as f64 * 0.7) as u64;
    let parallel_budget = budget.allocate(Duration::from_millis(parallel_budget_ms))
        .ok_or(Error::BudgetExhausted)?;

    let (price, inventory, recs) = tokio::try_join!(
        fetch_price(parallel_budget.clone(), product_id),
        fetch_inventory(parallel_budget.clone(), product_id),
        fetch_recommendations(parallel_budget.clone(), product_id),
    )?;

    Ok(ProductDetail { product, price, inventory, recommendations: recs })
}

func (s *ProductService) GetProductDetail(
    ctx context.Context,
    req *pb.ProductRequest,
) (*pb.ProductDetail, error) {

    // 检查上游传来的 deadline
    deadline, ok := ctx.Deadline()
    if ok {
        remaining := time.Until(deadline)
        if remaining < 10*time.Millisecond {
            return nil, status.Error(codes.DeadlineExceeded,
                "insufficient budget on entry")
        }
    }

    // 为下游调用设置更短的 deadline
    // 给自身处理预留 20ms
    childCtx, cancel := context.WithTimeout(ctx, time.Until(deadline)-20*time.Millisecond)
    defer cancel()

    // 并行调用下游服务
    g, gCtx := errgroup.WithContext(childCtx)

    var price *pb.Price
    g.Go(func() error {
        var err error
        price, err = s.priceClient.GetPrice(gCtx, &pb.PriceRequest{
            ProductId: req.ProductId,
        })
        return err
    })

    var inventory *pb.Inventory
    g.Go(func() error {
        var err error
        inventory, err = s.inventoryClient.CheckInventory(gCtx, &pb.InventoryRequest{
            ProductId: req.ProductId,
        })
        return err
    })

    if err := g.Wait(); err != nil {
        return nil, err
    }

    return &pb.ProductDetail{
        Price:     price,
        Inventory: inventory,
    }, nil
}

#include <stdlib.h>
#include <string.h>
#include <time.h>

/* 环形缓冲区：预分配内存，避免动态分配导致的延迟抖动 */
typedef struct {
    char   *buf;
    size_t  capacity;
    size_t  head;
    size_t  tail;
    size_t  count;
} ring_buffer_t;

ring_buffer_t *ring_buffer_create(size_t capacity) {
    ring_buffer_t *rb = malloc(sizeof(ring_buffer_t));
    if (!rb) return NULL;
    rb->buf = malloc(capacity);
    if (!rb->buf) { free(rb); return NULL; }
    rb->capacity = capacity;
    rb->head = 0;
    rb->tail = 0;
    rb->count = 0;
    return rb;
}

/* 写入数据，O(1) 时间复杂度，无内存分配 */
int ring_buffer_write(ring_buffer_t *rb, const char *data, size_t len) {
    if (len > rb->capacity - rb->count) {
        return -1; /* 空间不足 */
    }
    size_t first_chunk = rb->capacity - rb->head;
    if (len <= first_chunk) {
        memcpy(rb->buf + rb->head, data, len);
    } else {
        memcpy(rb->buf + rb->head, data, first_chunk);
        memcpy(rb->buf, data + first_chunk, len - first_chunk);
    }
    rb->head = (rb->head + len) % rb->capacity;
    rb->count += len;
    return 0;
}

/* 使用示例：延迟敏感路径上的日志记录 */
static ring_buffer_t *log_ring = NULL;

void init_log_ring(void) {
    log_ring = ring_buffer_create(4 * 1024 * 1024); /* 4MB 预分配 */
}

/* 在请求关键路径上记录日志，不触发 malloc */
void log_latency(const char *endpoint, long latency_us) {
    char entry[128];
    int len = snprintf(entry, sizeof(entry), "%ld,%s,%ld\n",
                       (long)time(NULL), endpoint, latency_us);
    if (len > 0 && log_ring) {
        ring_buffer_write(log_ring, entry, (size_t)len);
    }
    /* 实际刷盘由后台线程异步完成，不阻塞关键路径 */
}

use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Instant;

/// 无锁、无分配的延迟百分位近似跟踪器
/// 使用指数桶，覆盖 1us 到 ~100s
pub struct LockFreeLatencyTracker {
    buckets: [AtomicU64; 64],
    total_count: AtomicU64,
}

impl LockFreeLatencyTracker {
    pub const fn new() -> Self {
        const ZERO: AtomicU64 = AtomicU64::new(0);
        Self {
            buckets: [ZERO; 64],
            total_count: AtomicU64::new(0),
        }
    }

    /// 将微秒延迟映射到桶索引
    fn bucket_index(latency_us: u64) -> usize {
        if latency_us == 0 {
            return 0;
        }
        let idx = 64 - latency_us.leading_zeros() as usize;
        idx.min(63)
    }

    /// 记录一个延迟值（无锁，可从任意线程调用）
    pub fn record(&self, latency_us: u64) {
        let idx = Self::bucket_index(latency_us);
        self.buckets[idx].fetch_add(1, Ordering::Relaxed);
        self.total_count.fetch_add(1, Ordering::Relaxed);
    }

    /// 查询近似百分位值
    pub fn percentile(&self, p: f64) -> u64 {
        let total = self.total_count.load(Ordering::Relaxed);
        if total == 0 {
            return 0;
        }
        let target = (total as f64 * p / 100.0) as u64;
        let mut cumulative = 0u64;
        for i in 0..64 {
            cumulative += self.buckets[i].load(Ordering::Relaxed);
            if cumulative >= target {
                return 1u64 << i; // 桶的上界
            }
        }
        1u64 << 63
    }
}

/// RAII 计时守卫：作用域退出时自动记录延迟
pub struct LatencyGuard<'a> {
    tracker: &'a LockFreeLatencyTracker,
    start: Instant,
}

impl<'a> LatencyGuard<'a> {
    pub fn new(tracker: &'a LockFreeLatencyTracker) -> Self {
        Self {
            tracker,
            start: Instant::now(),
        }
    }
}

impl<'a> Drop for LatencyGuard<'a> {
    fn drop(&mut self) {
        let elapsed = self.start.elapsed().as_micros() as u64;
        self.tracker.record(elapsed);
    }
}

// 反模式：串行调用三个独立服务，总延迟 = sum(三个服务延迟)
func badHandler(ctx context.Context) (*Response, error) {
    user, err := userService.GetUser(ctx, userID)
    if err != nil { return nil, err }

    orders, err := orderService.GetOrders(ctx, userID)
    if err != nil { return nil, err }

    recs, err := recService.GetRecommendations(ctx, userID)
    if err != nil { return nil, err }

    return &Response{User: user, Orders: orders, Recs: recs}, nil
}

// 正确做法：并行调用，总延迟 = max(三个服务延迟)
func goodHandler(ctx context.Context) (*Response, error) {
    g, gCtx := errgroup.WithContext(ctx)

    var user *User
    g.Go(func() error {
        var err error
        user, err = userService.GetUser(gCtx, userID)
        return err
    })

    var orders []*Order
    g.Go(func() error {
        var err error
        orders, err = orderService.GetOrders(gCtx, userID)
        return err
    })

    var recs []*Recommendation
    g.Go(func() error {
        var err error
        recs, err = recService.GetRecommendations(gCtx, userID)
        return err
    })

    if err := g.Wait(); err != nil {
        return nil, err
    }
    return &Response{User: user, Orders: orders, Recs: recs}, nil
}

// 反模式：没有超时，下游故障可以无限拖住上游
resp, err := http.Get("http://downstream-service/api/data")

// 正确做法：始终设置超时
client := &http.Client{Timeout: 500 * time.Millisecond}
resp, err := client.Get("http://downstream-service/api/data")

// 更好的做法：使用 context 传递截止时间
ctx, cancel := context.WithTimeout(parentCtx, 500*time.Millisecond)
defer cancel()
req, _ := http.NewRequestWithContext(ctx, "GET",
    "http://downstream-service/api/data", nil)
resp, err := client.Do(req)

# 反模式：基于平均延迟告警
# 平均延迟可以在 P999 飙到 10 秒时仍然保持 20ms
- alert: HighLatency
  expr: avg(http_request_duration_seconds) > 0.1
  for: 5m

# 正确做法：基于百分位延迟告警
- alert: HighP99Latency
  expr: |
    histogram_quantile(0.99,
      sum(rate(http_request_duration_seconds_bucket[5m])) by (le)
    ) > 0.2
  for: 5m
  annotations:
    summary: "P99 延迟超过 200ms SLO"

// 反模式：在延迟关键路径上使用 JSON 序列化
// JSON 序列化/反序列化可以消耗 5-20ms
String json = objectMapper.writeValueAsString(largeObject);
MyObject obj = objectMapper.readValue(json, MyObject.class);

// 优化方案 1：使用 Protocol Buffers，二进制序列化，通常快 5-10 倍
byte[] bytes = myProtoObject.toByteArray();
MyProto obj = MyProto.parseFrom(bytes);

// 优化方案 2：使用 FlatBuffers，零拷贝反序列化
// 无需反序列化步骤，直接读取序列化后的缓冲区
MyTable obj = MyTable.getRootAsMyTable(byteBuffer);
int value = obj.someField(); // 直接从缓冲区读取，O(1)