【系统架构设计百科】Uber 架构演进：从单体到领域导向微服务

# 早期 Dispatch 系统的简化结构（概念性代码）
class DispatchService:
    """Uber 早期单体调度系统的核心逻辑"""

    def __init__(self, db: PostgresConnection):
        self.db = db
        self.geo_index = InMemoryGeoIndex()

    def request_ride(self, rider_id: str, pickup: LatLng, dropoff: LatLng) -> Trip:
        # 定价计算
        price = self._calculate_price(pickup, dropoff)

        # 创建行程记录
        trip = self.db.insert("trips", {
            "rider_id": rider_id,
            "pickup_lat": pickup.lat,
            "pickup_lng": pickup.lng,
            "dropoff_lat": dropoff.lat,
            "dropoff_lng": dropoff.lng,
            "price": price,
            "status": "REQUESTING"
        })

        # 查找最近的可用司机
        nearby_drivers = self.geo_index.find_nearest(
            pickup, radius_km=5.0, limit=10
        )

        # 逐一推送给司机
        for driver in nearby_drivers:
            if self._offer_trip_to_driver(driver, trip):
                trip.status = "MATCHED"
                trip.driver_id = driver.id
                self.db.update("trips", trip)
                return trip

        trip.status = "NO_DRIVERS"
        self.db.update("trips", trip)
        return trip

    def _calculate_price(self, pickup: LatLng, dropoff: LatLng) -> Decimal:
        distance = haversine(pickup, dropoff)
        base_fare = Decimal("2.50")
        per_km = Decimal("1.75")
        return base_fare + per_km * Decimal(str(distance))

    def _offer_trip_to_driver(self, driver: Driver, trip: Trip) -> bool:
        # 通过长轮询推送给司机端 App
        response = self._push_to_driver(driver.id, {
            "type": "TRIP_OFFER",
            "trip_id": trip.id,
            "pickup": trip.pickup,
            "timeout_seconds": 15
        })
        return response.accepted

// TChannel 服务端注册示例（Go）
package main

import (
    "context"
    "log"

    "github.com/uber/tchannel-go"
    "github.com/uber/tchannel-go/thrift"
)

func main() {
    ch, err := tchannel.NewChannel("pricing-service", nil)
    if err != nil {
        log.Fatalf("创建 TChannel 失败: %v", err)
    }

    server := thrift.NewServer(ch)
    server.Register(newPricingHandler())

    if err := ch.ListenAndServe("0.0.0.0:7890"); err != nil {
        log.Fatalf("监听失败: %v", err)
    }

    log.Println("pricing-service 启动在端口 7890")
    select {} // 阻塞主协程
}

type pricingHandler struct{}

func newPricingHandler() *pricingHandler {
    return &pricingHandler{}
}

func (h *pricingHandler) CalculatePrice(
    ctx context.Context,
    request *PriceRequest,
) (*PriceResponse, error) {
    distance := haversine(
        request.PickupLat, request.PickupLng,
        request.DropoffLat, request.DropoffLng,
    )

    surgeMultiplier := getSurgeMultiplier(
        request.PickupLat, request.PickupLng,
    )

    baseFare := 2.50
    perKm := 1.75
    price := (baseFare + perKm*distance) * surgeMultiplier

    return &PriceResponse{
        PriceCents:      int64(price * 100),
        SurgeMultiplier: surgeMultiplier,
        Currency:        request.Currency,
    }, nil
}

// 出行领域网关的接口定义（Java + Protobuf）
// ride_domain_gateway.proto

syntax = "proto3";

package uber.ride.gateway;

option java_package = "com.uber.ride.gateway";
option java_outer_classname = "RideDomainGatewayProto";

// 出行领域对外暴露的标准接口
service RideDomainGateway {
    // 请求行程
    rpc RequestRide(RequestRideRequest) returns (RequestRideResponse);

    // 查询行程状态
    rpc GetRideStatus(GetRideStatusRequest) returns (GetRideStatusResponse);

    // 取消行程
    rpc CancelRide(CancelRideRequest) returns (CancelRideResponse);

    // 获取行程预估价格
    rpc EstimateRide(EstimateRideRequest) returns (EstimateRideResponse);
}

message RequestRideRequest {
    string rider_id = 1;
    Location pickup = 2;
    Location dropoff = 3;
    string product_id = 4;        // UberX, UberBlack 等
    PaymentMethod payment = 5;
}

message RequestRideResponse {
    string ride_id = 1;
    RideStatus status = 2;
    DriverInfo driver = 3;
    PriceEstimate price = 4;
    int32 eta_seconds = 5;
}

message Location {
    double latitude = 1;
    double longitude = 2;
    string address = 3;
}

enum RideStatus {
    RIDE_STATUS_UNKNOWN = 0;
    REQUESTING = 1;
    MATCHING = 2;
    MATCHED = 3;
    EN_ROUTE_TO_PICKUP = 4;
    ARRIVED_AT_PICKUP = 5;
    IN_TRIP = 6;
    COMPLETED = 7;
    CANCELLED = 8;
}

// 出行领域网关的实现（Java）
package com.uber.ride.gateway;

import com.uber.ride.dispatch.DispatchService;
import com.uber.ride.pricing.PricingService;
import com.uber.ride.matching.MatchingService;
import com.uber.ride.trip.TripService;

public class RideDomainGatewayImpl implements RideDomainGateway {

    private final DispatchService dispatchService;
    private final PricingService pricingService;
    private final MatchingService matchingService;
    private final TripService tripService;

    public RideDomainGatewayImpl(
            DispatchService dispatchService,
            PricingService pricingService,
            MatchingService matchingService,
            TripService tripService) {
        this.dispatchService = dispatchService;
        this.pricingService = pricingService;
        this.matchingService = matchingService;
        this.tripService = tripService;
    }

    @Override
    public RequestRideResponse requestRide(RequestRideRequest request) {
        // 1. 计算价格
        PriceEstimate price = pricingService.calculate(
            request.getPickup(),
            request.getDropoff(),
            request.getProductId()
        );

        // 2. 创建行程记录
        Trip trip = tripService.create(
            request.getRiderId(),
            request.getPickup(),
            request.getDropoff(),
            price
        );

        // 3. 发起司机匹配（异步）
        matchingService.findDriverAsync(trip.getId(), request.getPickup());

        // 4. 返回响应
        return RequestRideResponse.newBuilder()
            .setRideId(trip.getId())
            .setStatus(RideStatus.REQUESTING)
            .setPrice(price)
            .setEtaSeconds(dispatchService.estimateEta(request.getPickup()))
            .build();
    }

    @Override
    public GetRideStatusResponse getRideStatus(GetRideStatusRequest request) {
        Trip trip = tripService.get(request.getRideId());
        return GetRideStatusResponse.newBuilder()
            .setRideId(trip.getId())
            .setStatus(trip.getStatus())
            .setDriver(trip.getDriverInfo())
            .setCurrentLocation(trip.getCurrentLocation())
            .build();
    }

    @Override
    public CancelRideResponse cancelRide(CancelRideRequest request) {
        Trip trip = tripService.get(request.getRideId());

        CancellationResult result = tripService.cancel(
            trip.getId(),
            request.getReason()
        );

        if (result.hasFee()) {
            // 领域网关内部协调取消费的计算
            pricingService.applyCancellationFee(
                trip.getRiderId(),
                result.getCancellationFee()
            );
        }

        return CancelRideResponse.newBuilder()
            .setRideId(trip.getId())
            .setStatus(RideStatus.CANCELLED)
            .setCancellationFee(result.getCancellationFee())
            .build();
    }

    @Override
    public EstimateRideResponse estimateRide(EstimateRideRequest request) {
        PriceEstimate price = pricingService.calculate(
            request.getPickup(),
            request.getDropoff(),
            request.getProductId()
        );

        int etaSeconds = dispatchService.estimateEta(request.getPickup());

        return EstimateRideResponse.newBuilder()
            .setPrice(price)
            .setEtaSeconds(etaSeconds)
            .build();
    }
}

// Ringpop 核心使用示例（Go）
package main

import (
    "fmt"
    "log"

    "github.com/uber/ringpop-go"
    "github.com/uber/ringpop-go/swim"
    "github.com/uber/tchannel-go"
)

func main() {
    // 创建 TChannel 通道
    ch, err := tchannel.NewChannel("trip-state-service", nil)
    if err != nil {
        log.Fatal(err)
    }

    // 创建 Ringpop 实例
    rp, err := ringpop.New(
        "trip-state-ring",
        ringpop.Channel(ch),
        ringpop.Address("10.0.1.100:7000"),
    )
    if err != nil {
        log.Fatal(err)
    }

    // 启动监听
    if err := ch.ListenAndServe("0.0.0.0:7000"); err != nil {
        log.Fatal(err)
    }

    // 加入集群（通过引导节点列表）
    bootstrapNodes := []string{
        "10.0.1.100:7000",
        "10.0.1.101:7000",
        "10.0.1.102:7000",
    }
    opts := new(swim.BootstrapOptions)
    opts.Hosts = bootstrapNodes

    if _, err := rp.Bootstrap(opts); err != nil {
        log.Fatal(err)
    }

    // 根据行程 ID 查找负责的节点
    tripID := "trip_abc123"
    targetNode, err := rp.Lookup(tripID)
    if err != nil {
        log.Fatal(err)
    }

    fmt.Printf("行程 %s 由节点 %s 负责处理\n", tripID, targetNode)

    // 如果当前节点就是目标节点，直接处理
    if targetNode == rp.WhoAmI() {
        handleTripLocally(tripID)
    } else {
        forwardToNode(targetNode, tripID)
    }
}

func handleTripLocally(tripID string) {
    fmt.Printf("本地处理行程: %s\n", tripID)
}

func forwardToNode(node string, tripID string) {
    fmt.Printf("转发行程 %s 到节点 %s\n", tripID, node)
}

// 虚拟节点配置示例（Go）
package hashring

import (
    "crypto/md5"
    "encoding/binary"
    "fmt"
    "sort"
    "sync"
)

type HashRing struct {
    mu             sync.RWMutex
    nodes          map[string]bool
    virtualNodes   int
    ring           []uint32
    ringToNode     map[uint32]string
}

func NewHashRing(virtualNodes int) *HashRing {
    return &HashRing{
        nodes:        make(map[string]bool),
        virtualNodes: virtualNodes,
        ringToNode:   make(map[uint32]string),
    }
}

func (h *HashRing) AddNode(node string) {
    h.mu.Lock()
    defer h.mu.Unlock()

    h.nodes[node] = true

    for i := 0; i < h.virtualNodes; i++ {
        virtualKey := fmt.Sprintf("%s#%d", node, i)
        hashValue := h.hash(virtualKey)
        h.ring = append(h.ring, hashValue)
        h.ringToNode[hashValue] = node
    }

    sort.Slice(h.ring, func(i, j int) bool {
        return h.ring[i] < h.ring[j]
    })
}

func (h *HashRing) RemoveNode(node string) {
    h.mu.Lock()
    defer h.mu.Unlock()

    delete(h.nodes, node)

    newRing := make([]uint32, 0)
    for _, hashValue := range h.ring {
        if h.ringToNode[hashValue] != node {
            newRing = append(newRing, hashValue)
        } else {
            delete(h.ringToNode, hashValue)
        }
    }
    h.ring = newRing
}

func (h *HashRing) Lookup(key string) string {
    h.mu.RLock()
    defer h.mu.RUnlock()

    if len(h.ring) == 0 {
        return ""
    }

    hashValue := h.hash(key)

    // 二分查找第一个大于等于 hashValue 的位置
    idx := sort.Search(len(h.ring), func(i int) bool {
        return h.ring[i] >= hashValue
    })

    // 环绕
    if idx >= len(h.ring) {
        idx = 0
    }

    return h.ringToNode[h.ring[idx]]
}

func (h *HashRing) hash(key string) uint32 {
    sum := md5.Sum([]byte(key))
    return binary.BigEndian.Uint32(sum[:4])
}

-- Schemaless 底层的 MySQL 表结构
CREATE TABLE schemaless_cells (
    row_key    VARCHAR(255) NOT NULL,
    column_name VARCHAR(255) NOT NULL,
    ref_key    BIGINT NOT NULL,
    body       MEDIUMBLOB NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,

    PRIMARY KEY (row_key, column_name, ref_key),
    INDEX idx_created_at (created_at)
) ENGINE=InnoDB
  DEFAULT CHARSET=utf8mb4;

# Schemaless 客户端使用示例（Python）
import json
import time
from typing import Optional

class SchemalessClient:
    """Schemaless 追加写入存储客户端"""

    def __init__(self, mysql_pool, shard_count: int = 4096):
        self.mysql_pool = mysql_pool
        self.shard_count = shard_count

    def put_cell(
        self, row_key: str, column_name: str, body: dict
    ) -> int:
        """写入一个 cell（追加操作，不覆盖旧版本）"""
        shard_id = self._get_shard(row_key)
        conn = self.mysql_pool.get_connection(shard_id)

        ref_key = int(time.time() * 1_000_000)  # 微秒级时间戳
        body_bytes = json.dumps(body).encode("utf-8")

        conn.execute(
            """
            INSERT INTO schemaless_cells
                (row_key, column_name, ref_key, body)
            VALUES (%s, %s, %s, %s)
            """,
            (row_key, column_name, ref_key, body_bytes),
        )
        conn.commit()
        return ref_key

    def get_cell_latest(
        self, row_key: str, column_name: str
    ) -> Optional[dict]:
        """读取指定 row_key + column_name 的最新版本"""
        shard_id = self._get_shard(row_key)
        conn = self.mysql_pool.get_connection(shard_id)

        result = conn.execute(
            """
            SELECT body FROM schemaless_cells
            WHERE row_key = %s AND column_name = %s
            ORDER BY ref_key DESC
            LIMIT 1
            """,
            (row_key, column_name),
        )

        row = result.fetchone()
        if row is None:
            return None
        return json.loads(row[0])

    def get_cell_history(
        self, row_key: str, column_name: str, limit: int = 10
    ) -> list:
        """读取指定 cell 的历史版本"""
        shard_id = self._get_shard(row_key)
        conn = self.mysql_pool.get_connection(shard_id)

        result = conn.execute(
            """
            SELECT ref_key, body, created_at
            FROM schemaless_cells
            WHERE row_key = %s AND column_name = %s
            ORDER BY ref_key DESC
            LIMIT %s
            """,
            (row_key, column_name, limit),
        )

        return [
            {
                "ref_key": row[0],
                "body": json.loads(row[1]),
                "created_at": row[2],
            }
            for row in result.fetchall()
        ]

    def _get_shard(self, row_key: str) -> int:
        """基于行键计算分片 ID"""
        hash_value = hash(row_key)
        return hash_value % self.shard_count


# 使用示例：行程状态管理
client = SchemalessClient(mysql_pool)

# 创建行程
trip_id = "trip_2024_abc123"
client.put_cell(trip_id, "base", {
    "rider_id": "rider_001",
    "pickup": {"lat": 37.7749, "lng": -122.4194},
    "dropoff": {"lat": 37.3382, "lng": -121.8863},
    "product": "UberX",
    "created_at": "2024-01-15T10:30:00Z",
})

# 行程状态更新（追加新版本，不覆盖旧版本）
client.put_cell(trip_id, "status", {
    "state": "MATCHING",
    "updated_at": "2024-01-15T10:30:01Z",
})

client.put_cell(trip_id, "status", {
    "state": "MATCHED",
    "driver_id": "driver_042",
    "updated_at": "2024-01-15T10:30:15Z",
})

client.put_cell(trip_id, "status", {
    "state": "IN_TRIP",
    "updated_at": "2024-01-15T10:35:00Z",
})

# 查询最新状态
latest_status = client.get_cell_latest(trip_id, "status")
# 返回: {"state": "IN_TRIP", "updated_at": "2024-01-15T10:35:00Z"}

# 查询状态变更历史
history = client.get_cell_history(trip_id, "status")
# 返回最近 10 条状态变更记录

// 动态定价计算核心逻辑（Go）
package surge

import (
    "math"
    "sync"
    "time"
)

// H3Cell 代表一个 H3 六边形区域
type H3Cell struct {
    Index      uint64
    Resolution int
}

// SupplyDemand 记录一个区域的供需数据
type SupplyDemand struct {
    ActiveDrivers   int     // 可用司机数
    OpenRequests    int     // 未匹配的行程请求数
    RecentPickups   int     // 最近 5 分钟内的接单数
    AvgWaitSeconds  float64 // 平均等待时间
}

// SurgeCalculator 动态定价计算器
type SurgeCalculator struct {
    mu              sync.RWMutex
    cellData        map[uint64]*SupplyDemand
    surgeMultiplier map[uint64]float64
    config          SurgeConfig
}

type SurgeConfig struct {
    MinMultiplier     float64       // 最低倍率（通常为 1.0）
    MaxMultiplier     float64       // 最高倍率（通常为 5.0 或 8.0）
    CalculateInterval time.Duration // 计算周期
    SmoothingFactor   float64       // 平滑因子，避免倍率剧烈波动
    DemandThreshold   float64       // 触发加价的供需比阈值
}

func NewSurgeCalculator(config SurgeConfig) *SurgeCalculator {
    return &SurgeCalculator{
        cellData:        make(map[uint64]*SupplyDemand),
        surgeMultiplier: make(map[uint64]float64),
        config:          config,
    }
}

// UpdateSupplyDemand 更新某个区域的供需数据
func (sc *SurgeCalculator) UpdateSupplyDemand(
    cellIndex uint64, data *SupplyDemand,
) {
    sc.mu.Lock()
    defer sc.mu.Unlock()
    sc.cellData[cellIndex] = data
}

// CalculateSurge 计算所有区域的倍率
func (sc *SurgeCalculator) CalculateSurge() {
    sc.mu.Lock()
    defer sc.mu.Unlock()

    for cellIndex, data := range sc.cellData {
        newMultiplier := sc.computeMultiplier(data)
        oldMultiplier, exists := sc.surgeMultiplier[cellIndex]

        if !exists {
            oldMultiplier = 1.0
        }

        // 指数移动平均平滑，避免倍率剧烈波动
        smoothed := oldMultiplier +
            sc.config.SmoothingFactor*(newMultiplier-oldMultiplier)

        // 四舍五入到 0.1
        smoothed = math.Round(smoothed*10) / 10

        // 限制在合法范围内
        if smoothed < sc.config.MinMultiplier {
            smoothed = sc.config.MinMultiplier
        }
        if smoothed > sc.config.MaxMultiplier {
            smoothed = sc.config.MaxMultiplier
        }

        sc.surgeMultiplier[cellIndex] = smoothed
    }
}

func (sc *SurgeCalculator) computeMultiplier(
    data *SupplyDemand,
) float64 {
    if data.ActiveDrivers == 0 {
        return sc.config.MaxMultiplier
    }

    // 供需比 = 需求 / 供给
    demandSupplyRatio := float64(data.OpenRequests) /
        float64(data.ActiveDrivers)

    if demandSupplyRatio <= sc.config.DemandThreshold {
        return 1.0
    }

    // 超过阈值的部分线性映射到倍率
    excessRatio := demandSupplyRatio - sc.config.DemandThreshold
    multiplier := 1.0 + excessRatio*0.5

    return multiplier
}

// GetSurge 获取指定区域的当前倍率
func (sc *SurgeCalculator) GetSurge(cellIndex uint64) float64 {
    sc.mu.RLock()
    defer sc.mu.RUnlock()

    if m, ok := sc.surgeMultiplier[cellIndex]; ok {
        return m
    }
    return 1.0
}

// H3 索引使用示例（Go）
package geo

import (
    "github.com/uber/h3-go/v4"
)

// FindNearbyDrivers 查找某个位置附近的可用司机
func FindNearbyDrivers(
    lat, lng float64,
    radiusKm float64,
    driverIndex map[h3.Cell][]DriverInfo,
) []DriverInfo {
    // 将经纬度转换为 H3 索引（分辨率 9，约 0.1 km^2）
    center := h3.NewLatLng(lat, lng)
    centerCell := h3.LatLngToCell(center, 9)

    // 计算覆盖指定半径的 K 环大小
    // 分辨率 9 的六边形边长约 174 米
    kRings := int(radiusKm / 0.174)
    if kRings < 1 {
        kRings = 1
    }

    // 获取中心点及周围的所有六边形
    cells := h3.GridDisk(centerCell, kRings)

    var drivers []DriverInfo
    for _, cell := range cells {
        if cellDrivers, ok := driverIndex[cell]; ok {
            drivers = append(drivers, cellDrivers...)
        }
    }

    return drivers
}

type DriverInfo struct {
    DriverID  string
    Lat       float64
    Lng       float64
    ProductID string  // UberX, UberBlack 等
    Available bool
}

# Flink 作业配置示例（YAML）
job:
  name: surge-pricing-aggregator
  parallelism: 128
  checkpoint:
    interval: 30s
    mode: EXACTLY_ONCE
    storage: s3://uber-flink-checkpoints/surge/

sources:
  - name: driver-locations
    type: kafka
    topic: driver.location.updates
    group: surge-calculator
    deserializer: avro
    watermark:
      strategy: bounded-out-of-orderness
      max-delay: 5s

  - name: ride-requests
    type: kafka
    topic: ride.request.events
    group: surge-calculator
    deserializer: avro

sinks:
  - name: surge-output
    type: kafka
    topic: surge.multiplier.updates
    serializer: avro

windows:
  - name: supply-demand-window
    type: sliding
    size: 5m
    slide: 30s
    key: h3_cell_index
    aggregate:
      - field: active_drivers
        function: count_distinct(driver_id)
        filter: "status = 'AVAILABLE'"
      - field: open_requests
        function: count(request_id)
        filter: "status = 'UNMATCHED'"
      - field: avg_wait
        function: avg(wait_seconds)

// Jaeger 追踪集成示例（Go）
package main

import (
    "context"
    "fmt"
    "net/http"
    "time"

    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/attribute"
    "go.opentelemetry.io/otel/exporters/jaeger"
    "go.opentelemetry.io/otel/sdk/resource"
    sdktrace "go.opentelemetry.io/otel/sdk/trace"
    semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
    "go.opentelemetry.io/otel/trace"
)

func initTracer() (*sdktrace.TracerProvider, error) {
    exporter, err := jaeger.New(
        jaeger.WithAgentEndpoint(
            jaeger.WithAgentHost("jaeger-agent.uber.internal"),
            jaeger.WithAgentPort("6831"),
        ),
    )
    if err != nil {
        return nil, err
    }

    tp := sdktrace.NewTracerProvider(
        sdktrace.WithBatcher(exporter),
        sdktrace.WithResource(resource.NewWithAttributes(
            semconv.SchemaURL,
            semconv.ServiceNameKey.String("ride-pricing-service"),
            attribute.String("environment", "production"),
            attribute.String("domain", "ride"),
        )),
        sdktrace.WithSampler(
            sdktrace.ParentBased(sdktrace.TraceIDRatioBased(0.01)),
        ),
    )

    otel.SetTracerProvider(tp)
    return tp, nil
}

func handlePriceRequest(w http.ResponseWriter, r *http.Request) {
    ctx := r.Context()
    tracer := otel.Tracer("ride-pricing")

    ctx, span := tracer.Start(ctx, "CalculatePrice",
        trace.WithAttributes(
            attribute.String("product_id", r.URL.Query().Get("product")),
        ),
    )
    defer span.End()

    // 子 Span：路径计算
    routeCtx, routeSpan := tracer.Start(ctx, "CalculateRoute")
    route := calculateRoute(routeCtx)
    routeSpan.SetAttributes(
        attribute.Float64("distance_km", route.DistanceKm),
        attribute.Int("duration_minutes", route.DurationMinutes),
    )
    routeSpan.End()

    // 子 Span：获取动态定价倍率
    surgeCtx, surgeSpan := tracer.Start(ctx, "GetSurgeMultiplier")
    surge := getSurgeMultiplier(surgeCtx, route.PickupH3Cell)
    surgeSpan.SetAttributes(
        attribute.Float64("surge_multiplier", surge),
    )
    surgeSpan.End()

    // 子 Span：最终价格计算
    _, priceSpan := tracer.Start(ctx, "ComputeFinalPrice")
    price := computePrice(route, surge)
    priceSpan.SetAttributes(
        attribute.Int64("price_cents", price.Cents),
        attribute.String("currency", price.Currency),
    )
    priceSpan.End()

    span.SetAttributes(
        attribute.Int64("final_price_cents", price.Cents),
    )

    fmt.Fprintf(w, `{"price_cents": %d, "currency": "%s"}`,
        price.Cents, price.Currency)
}

func calculateRoute(ctx context.Context) *Route {
    time.Sleep(50 * time.Millisecond)
    return &Route{DistanceKm: 15.3, DurationMinutes: 22}
}

func getSurgeMultiplier(ctx context.Context, h3Cell uint64) float64 {
    time.Sleep(10 * time.Millisecond)
    return 1.5
}

func computePrice(route *Route, surge float64) *Price {
    baseFare := int64(250)
    perKm := int64(175)
    distanceFare := int64(route.DistanceKm * float64(perKm))
    total := int64(float64(baseFare+distanceFare) * surge)
    return &Price{Cents: total, Currency: "USD"}
}

type Route struct {
    DistanceKm      float64
    DurationMinutes int
    PickupH3Cell    uint64
}

type Price struct {
    Cents    int64
    Currency string
}

// 第一阶段：Gateway 作为代理层（Java）
public class PricingGatewayPhase1 implements PricingDomainGateway {

    private final FareEstimatorClient fareEstimator;
    private final SurgeServiceClient surgeService;
    private final PromotionServiceClient promotionService;

    @Override
    public PriceEstimate estimatePrice(EstimateRequest request) {
        // 第一阶段：直接代理到原有服务
        // 没有改变任何内部逻辑
        FareEstimate fare = fareEstimator.estimate(
            request.getPickup(),
            request.getDropoff(),
            request.getProductId()
        );

        double surgeMultiplier = surgeService.getSurge(
            request.getPickup().getH3Cell()
        );

        PromotionDiscount discount = promotionService.calculate(
            request.getRiderId(),
            fare.getBaseFare()
        );

        return PriceEstimate.builder()
            .baseFare(fare.getBaseFare())
            .surgeMultiplier(surgeMultiplier)
            .discount(discount.getAmount())
            .finalPrice(
                fare.getBaseFare() * surgeMultiplier
                - discount.getAmount()
            )
            .build();
    }
}

// 第三阶段：并行化与缓存优化后的定价核心（Java）
public class PricingCoreOptimized {

    private final SurgeCalculator surgeCalculator;
    private final PromotionEngine promotionEngine;
    private final TaxCalculator taxCalculator;
    private final PricingConfig pricingConfig;
    private final LoadingCache<String, ProductConfig> configCache;

    public PricingCoreOptimized(
            SurgeCalculator surgeCalculator,
            PromotionEngine promotionEngine,
            TaxCalculator taxCalculator,
            PricingConfig pricingConfig) {
        this.surgeCalculator = surgeCalculator;
        this.promotionEngine = promotionEngine;
        this.taxCalculator = taxCalculator;
        this.pricingConfig = pricingConfig;

        this.configCache = CacheBuilder.newBuilder()
            .maximumSize(1000)
            .expireAfterWrite(5, TimeUnit.MINUTES)
            .build(new CacheLoader<String, ProductConfig>() {
                @Override
                public ProductConfig load(String productId) {
                    return pricingConfig.getProductConfig(productId);
                }
            });
    }

    public PriceResult calculate(PriceRequest request) {
        // 从缓存获取产品配置
        ProductConfig config = configCache.getUnchecked(
            request.getProductId()
        );

        // 并行获取动态定价倍率和优惠信息
        CompletableFuture<Double> surgeFuture = CompletableFuture
            .supplyAsync(() -> surgeCalculator.getSurge(
                request.getPickupH3Cell()
            ));

        CompletableFuture<Discount> promoFuture = CompletableFuture
            .supplyAsync(() -> promotionEngine.calculate(
                request.getRiderId(),
                request.getProductId()
            ));

        // 等待并行结果
        double surgeMultiplier = surgeFuture.join();
        Discount discount = promoFuture.join();

        // 基础费用计算
        long baseFareCents = config.getBaseFareCents()
            + (long) (request.getDistanceKm()
                * config.getPerKmCents())
            + (long) (request.getDurationMinutes()
                * config.getPerMinuteCents());

        // 应用动态倍率
        long surgedFareCents = (long) (baseFareCents * surgeMultiplier);

        // 应用优惠
        long afterDiscountCents = Math.max(
            surgedFareCents - discount.getAmountCents(),
            config.getMinimumFareCents()
        );

        // 税费计算
        TaxResult tax = taxCalculator.calculate(
            afterDiscountCents,
            request.getPickupRegion()
        );

        long finalPriceCents = afterDiscountCents + tax.getTaxCents();

        return PriceResult.builder()
            .baseFareCents(baseFareCents)
            .surgeMultiplier(surgeMultiplier)
            .discountCents(discount.getAmountCents())
            .taxCents(tax.getTaxCents())
            .finalPriceCents(finalPriceCents)
            .currency(config.getCurrency())
            .breakdown(buildBreakdown(
                baseFareCents, surgeMultiplier,
                discount, tax, config
            ))
            .build();
    }

    private PriceBreakdown buildBreakdown(
            long baseFareCents,
            double surgeMultiplier,
            Discount discount,
            TaxResult tax,
            ProductConfig config) {
        return PriceBreakdown.builder()
            .baseFare(formatCents(baseFareCents, config.getCurrency()))
            .surgeAmount(formatCents(
                (long)(baseFareCents * (surgeMultiplier - 1)),
                config.getCurrency()
            ))
            .discount(formatCents(
                discount.getAmountCents(),
                config.getCurrency()
            ))
            .tax(formatCents(tax.getTaxCents(), config.getCurrency()))
            .build();
    }

    private String formatCents(long cents, String currency) {
        return String.format("%s %.2f", currency, cents / 100.0);
    }
}