深入理解系統延遲：從 Latency 到 Tail Latency 的優化之道

延遲的組成：
請求發出 → [網路傳輸] → [處理時間] → [網路傳輸] → 收到回應
         ↑                                            ↑
         |________________總延遲____________________|

1. 網路延遲（Network Latency）：
   - 傳播延遲：光速限制（約 200,000 km/s 在光纖中）
   - 傳輸延遲：資料大小 / 頻寬
   - 處理延遲：路由器/交換機處理
   - 排隊延遲：網路擁塞

2. 應用延遲（Application Latency）：
   - CPU 處理時間
   - I/O 等待（資料庫、檔案系統）
   - 記憶體存取
   - 垃圾回收（GC）

3. 儲存延遲（Storage Latency）：
   - SSD：~0.1ms
   - HDD：~10ms
   - 網路儲存：更高

L1 Cache：         0.5 ns
L2 Cache：         7 ns
RAM：              100 ns
SSD 讀取：         150,000 ns (0.15 ms)
HDD 讀取：         10,000,000 ns (10 ms)
同城網路 RTT：     500,000 ns (0.5 ms)
跨國網路 RTT：     150,000,000 ns (150 ms)

常見指標：
- P50（中位數）：50% 的請求延遲低於此值
- P90：90% 的請求延遲低於此值
- P95：95% 的請求延遲低於此值
- P99：99% 的請求延遲低於此值
- P99.9：99.9% 的請求延遲低於此值

# 延遲分布範例
import numpy as np
import matplotlib.pyplot as plt

# 模擬延遲數據（對數常態分布）
np.random.seed(42)
latencies = np.random.lognormal(3.0, 0.5, 10000)

# 計算統計值
mean_latency = np.mean(latencies)
p50 = np.percentile(latencies, 50)
p90 = np.percentile(latencies, 90)
p95 = np.percentile(latencies, 95)
p99 = np.percentile(latencies, 99)
p999 = np.percentile(latencies, 99.9)

print(f"平均延遲: {mean_latency:.2f} ms")
print(f"P50: {p50:.2f} ms")
print(f"P90: {p90:.2f} ms")
print(f"P95: {p95:.2f} ms")
print(f"P99: {p99:.2f} ms")
print(f"P99.9: {p999:.2f} ms")

# 輸出範例：
# 平均延遲: 22.31 ms
# P50: 20.09 ms
# P90: 33.28 ms
# P95: 39.44 ms
# P99: 56.25 ms
# P99.9: 79.88 ms

場景：一個請求需要呼叫 10 個微服務

如果每個服務的 P99 = 100ms：
- 最佳情況：所有服務都快 = 100ms
- 最差情況：至少一個服務慢的機率 = 1 - 0.99^10 = 9.6%
- 也就是說，約 10% 的請求會遇到尾延遲！

import time
import functools
from collections import defaultdict
import threading

class LatencyTracker:
    def __init__(self, window_size=10000):
        self.window_size = window_size
        self.latencies = defaultdict(list)
        self.lock = threading.Lock()
    
    def record(self, operation, latency_ms):
        """記錄延遲"""
        with self.lock:
            latencies = self.latencies[operation]
            latencies.append(latency_ms)
            
            # 保持窗口大小
            if len(latencies) > self.window_size:
                latencies.pop(0)
    
    def get_percentiles(self, operation):
        """計算百分位數"""
        with self.lock:
            data = self.latencies[operation]
            if not data:
                return {}
            
            sorted_data = sorted(data)
            return {
                'count': len(data),
                'mean': sum(data) / len(data),
                'p50': self._percentile(sorted_data, 50),
                'p90': self._percentile(sorted_data, 90),
                'p95': self._percentile(sorted_data, 95),
                'p99': self._percentile(sorted_data, 99),
                'p999': self._percentile(sorted_data, 99.9),
                'max': sorted_data[-1]
            }
    
    def _percentile(self, sorted_data, p):
        """計算百分位數"""
        index = int(len(sorted_data) * p / 100)
        return sorted_data[min(index, len(sorted_data) - 1)]

# 全域追蹤器
latency_tracker = LatencyTracker()

# 裝飾器
def track_latency(operation_name):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            start_time = time.time()
            try:
                result = func(*args, **kwargs)
                return result
            finally:
                latency_ms = (time.time() - start_time) * 1000
                latency_tracker.record(operation_name, latency_ms)
        return wrapper
    return decorator

# 使用範例
@track_latency('api_request')
def handle_request():
    # 模擬處理
    time.sleep(0.02 + np.random.exponential(0.01))
    return "OK"

# 定期報告
def report_latencies():
    for operation, _ in latency_tracker.latencies.items():
        stats = latency_tracker.get_percentiles(operation)
        print(f"\n{operation}:")
        print(f"  Count: {stats['count']}")
        print(f"  Mean: {stats['mean']:.2f} ms")
        print(f"  P50: {stats['p50']:.2f} ms")
        print(f"  P90: {stats['p90']:.2f} ms")
        print(f"  P95: {stats['p95']:.2f} ms")
        print(f"  P99: {stats['p99']:.2f} ms")
        print(f"  P99.9: {stats['p999']:.2f} ms")
        print(f"  Max: {stats['max']:.2f} ms")

# OpenTelemetry 範例
from opentelemetry import trace
from opentelemetry.exporter.jaeger import JaegerExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

# 設定追蹤
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)

# Jaeger 匯出器
jaeger_exporter = JaegerExporter(
    agent_host_name="localhost",
    agent_port=6831,
)

# 批次處理器
span_processor = BatchSpanProcessor(jaeger_exporter)
trace.get_tracer_provider().add_span_processor(span_processor)

# 使用追蹤
@app.route('/api/order')
def create_order():
    with tracer.start_as_current_span('create_order') as span:
        # 追蹤資料庫操作
        with tracer.start_as_current_span('db_insert'):
            order = insert_order_to_db()
        
        # 追蹤外部服務呼叫
        with tracer.start_as_current_span('payment_service'):
            payment_result = call_payment_service(order)
        
        # 追蹤快取操作
        with tracer.start_as_current_span('cache_update'):
            update_cache(order)
        
        return jsonify({'order_id': order.id})

# Prometheus 指標定義
histogram_latency = Histogram(
    'http_request_duration_seconds',
    'HTTP request latency',
    ['method', 'endpoint', 'status'],
    buckets=[0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5]
)

# Grafana 查詢範例
# P95 延遲
histogram_quantile(0.95, 
  sum(rate(http_request_duration_seconds_bucket[5m])) 
  by (le, endpoint)
)

# 各百分位數
histogram_quantile(0.5, ...) # P50
histogram_quantile(0.9, ...) # P90
histogram_quantile(0.99, ...) # P99

class MultiLevelCache:
    """多層快取策略"""
    
    def __init__(self):
        self.l1_cache = {}  # 進程內快取
        self.l2_cache = Redis()  # Redis 快取
        self.l3_cache = CDN()  # CDN 快取
    
    async def get(self, key):
        # L1: 進程內（~0.001ms）
        if key in self.l1_cache:
            return self.l1_cache[key]
        
        # L2: Redis（~1ms）
        value = await self.l2_cache.get(key)
        if value:
            self.l1_cache[key] = value
            return value
        
        # L3: CDN（~10ms）
        value = await self.l3_cache.get(key)
        if value:
            await self.l2_cache.set(key, value)
            self.l1_cache[key] = value
            return value
        
        # 原始資料源（~100ms）
        value = await self.fetch_from_source(key)
        await self.update_all_caches(key, value)
        return value

import asyncio
from concurrent.futures import ThreadPoolExecutor

class ParallelProcessor:
    """並行處理降低延遲"""
    
    def __init__(self):
        self.executor = ThreadPoolExecutor(max_workers=10)
    
    async def process_request(self, request):
        # 可以並行的操作
        tasks = [
            self.fetch_user_data(request.user_id),
            self.fetch_recommendations(request.user_id),
            self.fetch_inventory(request.product_ids),
            self.check_promotions(request)
        ]
        
        # 並行執行
        results = await asyncio.gather(*tasks)
        
        user_data, recommendations, inventory, promotions = results
        
        # 組合結果
        return self.build_response(
            user_data, 
            recommendations, 
            inventory, 
            promotions
        )
    
    async def fetch_user_data(self, user_id):
        # 降級策略
        try:
            return await asyncio.wait_for(
                self.user_service.get(user_id),
                timeout=0.1  # 100ms 超時
            )
        except asyncio.TimeoutError:
            # 返回快取或預設值
            return self.get_cached_user(user_id)

# 資料庫連線池
from sqlalchemy.pool import QueuePool

engine = create_engine(
    'postgresql://user:pass@localhost/db',
    poolclass=QueuePool,
    pool_size=20,          # 連線池大小
    max_overflow=40,       # 最大溢出
    pool_pre_ping=True,    # 連線健康檢查
    pool_recycle=3600      # 連線回收時間
)

# HTTP 連線池
import httpx

class HTTPClient:
    def __init__(self):
        self.client = httpx.AsyncClient(
            limits=httpx.Limits(
                max_keepalive_connections=100,
                max_connections=200,
                keepalive_expiry=30
            ),
            timeout=httpx.Timeout(
                connect=5.0,
                read=10.0,
                write=5.0,
                pool=1.0
            )
        )
    
    async def request(self, url, **kwargs):
        # 自動重試
        for attempt in range(3):
            try:
                return await self.client.get(url, **kwargs)
            except httpx.TimeoutException:
                if attempt == 2:
                    raise
                await asyncio.sleep(0.1 * (attempt + 1))

class PrecomputedCache:
    """預先計算降低即時延遲"""
    
    def __init__(self):
        self.redis = Redis()
    
    def precompute_recommendations(self):
        """批次預計算推薦"""
        for user_id in self.get_active_users():
            recommendations = self.compute_recommendations(user_id)
            
            # 儲存結果
            key = f"recommendations:{user_id}"
            self.redis.setex(
                key, 
                3600,  # 1小時過期
                json.dumps(recommendations)
            )
    
    async def get_recommendations(self, user_id):
        # 優先使用預計算結果
        key = f"recommendations:{user_id}"
        cached = await self.redis.get(key)
        
        if cached:
            return json.loads(cached)
        
        # 即時計算（較慢）
        return await self.compute_recommendations_async(user_id)

class HedgedRequests:
    """請求對冲策略"""
    
    async def hedged_request(self, servers, request, hedge_delay=0.01):
        """
        發送請求到多個伺服器，返回最快的回應
        """
        tasks = []
        
        # 主請求
        primary_task = asyncio.create_task(
            self.send_request(servers[0], request)
        )
        tasks.append(primary_task)
        
        # 等待一小段時間
        await asyncio.sleep(hedge_delay)
        
        # 如果主請求還沒完成，發送備份請求
        if not primary_task.done():
            for server in servers[1:3]:  # 最多3個請求
                task = asyncio.create_task(
                    self.send_request(server, request)
                )
                tasks.append(task)
        
        # 返回第一個完成的結果
        done, pending = await asyncio.wait(
            tasks, 
            return_when=asyncio.FIRST_COMPLETED
        )
        
        # 取消其他請求
        for task in pending:
            task.cancel()
        
        return done.pop().result()

class AdaptiveTimeout:
    """根據歷史延遲動態調整超時"""
    
    def __init__(self):
        self.latency_history = []
        self.window_size = 1000
    
    def record_latency(self, latency):
        self.latency_history.append(latency)
        if len(self.latency_history) > self.window_size:
            self.latency_history.pop(0)
    
    def calculate_timeout(self):
        if not self.latency_history:
            return 1.0  # 預設 1 秒
        
        # 使用 P99 作為基準
        sorted_latencies = sorted(self.latency_history)
        p99 = sorted_latencies[int(len(sorted_latencies) * 0.99)]
        
        # 給予一些緩衝
        return min(p99 * 1.5, 5.0)  # 最多 5 秒
    
    async def request_with_timeout(self, func, *args, **kwargs):
        timeout = self.calculate_timeout()
        
        start_time = time.time()
        try:
            result = await asyncio.wait_for(
                func(*args, **kwargs), 
                timeout=timeout
            )
            latency = time.time() - start_time
            self.record_latency(latency)
            return result
        except asyncio.TimeoutError:
            # 記錄超時
            self.record_latency(timeout)
            raise

import heapq
from dataclasses import dataclass
from typing import Any

@dataclass
class PriorityRequest:
    priority: int
    timestamp: float
    request: Any
    
    def __lt__(self, other):
        # 優先級高的先處理
        return self.priority > other.priority

class PriorityQueue:
    """優先處理重要請求，降低其尾延遲"""
    
    def __init__(self):
        self.queue = []
        self.lock = threading.Lock()
    
    def push(self, request, priority=0):
        with self.lock:
            heapq.heappush(
                self.queue, 
                PriorityRequest(priority, time.time(), request)
            )
    
    def pop(self):
        with self.lock:
            if self.queue:
                return heapq.heappop(self.queue).request
            return None
    
    def process_requests(self):
        """處理器"""
        while True:
            request = self.pop()
            if request:
                # VIP 請求分配更多資源
                if request.priority > 100:
                    self.process_vip_request(request)
                else:
                    self.process_normal_request(request)
            else:
                time.sleep(0.001)  # 避免 busy waiting

class CircuitBreaker:
    """熔斷器避免級聯延遲"""
    
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failure_count = 0
        self.last_failure_time = None
        self.state = 'CLOSED'  # CLOSED, OPEN, HALF_OPEN
    
    async def call(self, func, *args, **kwargs):
        if self.state == 'OPEN':
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = 'HALF_OPEN'
                self.failure_count = 0
            else:
                # 快速失敗，避免等待
                raise Exception("Circuit breaker is OPEN")
        
        try:
            result = await asyncio.wait_for(
                func(*args, **kwargs),
                timeout=1.0  # 1秒超時
            )
            
            if self.state == 'HALF_OPEN':
                self.state = 'CLOSED'
            
            return result
            
        except Exception as e:
            self.failure_count += 1
            self.last_failure_time = time.time()
            
            if self.failure_count >= self.failure_threshold:
                self.state = 'OPEN'
            
            raise

# Service Level Objectives (SLO)
api_slos:
  availability: 99.9%  # 三個九
  latency:
    p50: 50ms
    p90: 100ms
    p95: 200ms
    p99: 500ms
    p999: 1000ms

# 告警規則
alerts:
  - name: high_p99_latency
    condition: p99 > 500ms for 5 minutes
    severity: warning
    
  - name: very_high_p99_latency
    condition: p99 > 1000ms for 2 minutes
    severity: critical
    
  - name: latency_spike
    condition: p99 > 2 * p99_baseline
    severity: warning

總延遲預算: 200ms

分配:
  網路 RTT: 20ms (10%)
  負載平衡器: 5ms (2.5%)
  API Gateway: 10ms (5%)
  應用處理: 50ms (25%)
  資料庫查詢: 80ms (40%)
  快取查詢: 5ms (2.5%)
  序列化: 10ms (5%)
  其他: 20ms (10%)

class EcommerceOptimization:
    """電商網站延遲優化實例"""
    
    async def get_product_page(self, product_id):
        # 並行獲取所有需要的資料
        tasks = []
        
        # 基本資訊（快取優先）
        tasks.append(self.get_product_info_cached(product_id))
        
        # 庫存（即時但有備案）
        tasks.append(self.get_inventory_with_fallback(product_id))
        
        # 價格（可能有促銷，需要計算）
        tasks.append(self.calculate_price_async(product_id))
        
        # 評論（分頁載入）
        tasks.append(self.get_reviews_first_page(product_id))
        
        # 推薦（預計算 + 即時混合）
        tasks.append(self.get_recommendations_hybrid(product_id))
        
        # 設定總超時
        try:
            results = await asyncio.wait_for(
                asyncio.gather(*tasks, return_exceptions=True),
                timeout=0.2  # 200ms 總預算
            )
        except asyncio.TimeoutError:
            # 降級回應
            results = await self.get_degraded_response(product_id)
        
        return self.build_product_page(results)

class TradingSystemOptimization:
    """交易系統的極致延遲優化"""
    
    def __init__(self):
        # 預分配記憶體
        self.order_pool = [Order() for _ in range(10000)]
        self.pool_index = 0
        
        # CPU 親和性
        self.bind_to_cpu_core(core_id=3)
        
        # 關閉 GC
        gc.disable()
    
    def process_order(self, order_data):
        # 使用物件池避免記憶體分配
        order = self.order_pool[self.pool_index]
        self.pool_index = (self.pool_index + 1) % 10000
        
        # 零拷貝解析
        order.parse_from_buffer(order_data)
        
        # 無鎖資料結構
        if self.lock_free_queue.try_push(order):
            return True
        
        # 失敗快速返回
        return False