实时竞价广告系统架构设计实战

// BidContext 竞价上下文，在 goroutine 间传递
type BidContext struct {
    RequestID  string
    DeviceID   string
    IP         string
    Geo        string
    UserTags   []string
    Candidates []*AdCandidate
    wg         sync.WaitGroup
    mu         sync.Mutex
}

func (e *BidEngine) ProcessBid(ctx context.Context, req *BidRequest) (*BidResponse, error) {
    bidCtx := &BidContext{
        RequestID: generateRequestID(),
        DeviceID:  req.Device.DeviceID,
        IP:        req.Device.IP,
        Geo:       req.Device.Geo,
        UserTags:  req.UserTags,
    }

    errChan := make(chan error, 3)

    // 三个子任务并发执行
    bidCtx.wg.Add(3)
    go func() { defer bidCtx.wg.Done(); errChan <- e.fraudFilter.Filter(ctx, bidCtx) }()
    go func() {
        defer bidCtx.wg.Done()
        bidCtx.Candidates, errChan <- e.adRetriever.Recall(ctx, bidCtx)
    }()
    go func() { defer bidCtx.wg.Done(); errChan <- e.freqBudget.PreCheck(ctx, bidCtx) }()

    bidCtx.wg.Wait()
    close(errChan)
    for err := range errChan { if err != nil { return nil, err } }

    // 串行出价计算
    bids := make([]*BidResult, 0, len(bidCtx.Candidates))
    for _, cand := range bidCtx.Candidates {
        price, err := e.bidder.Calculate(ctx, bidCtx, cand)
        if err != nil || price <= 0 { continue }
        bids = append(bids, &BidResult{AdID: cand.AdID, Price: price})
    }

    // 第二价格拍卖
    sort.Slice(bids, func(i, j int) bool { return bids[i].Price > bids[j].Price })
    if len(bids) == 0 { return noBidResponse(), nil }

    winner := bids[0]
    finalPrice := winner.Price
    if len(bids) > 1 { finalPrice = bids[1].Price }

    // 异步扣减预算和频次
    go e.freqBudget.Deduct(context.Background(), bidCtx, winner)

    return buildBidResponse(winner, finalPrice), nil
}

type ConsistentHash struct {
    nodes     map[uint32]string
    ring      []uint32
    mu        sync.RWMutex
    vfCount   int // 每节点虚拟节点数
    singleflight.Group
}

func (h *ConsistentHash) Add(node string) {
    h.mu.Lock()
    defer h.mu.Unlock()
    for i := 0; i < h.vfCount; i++ {
        hash := crc32.ChecksumIEEE([]byte(fmt.Sprintf("%s-%d", node, i)))
        h.ring = append(h.ring, hash)
        h.nodes[hash] = node
    }
    sort.Sort(uint32Slice(h.ring))
}

func (h *ConsistentHash) Get(key string) (string, bool) {
    h.mu.RLock()
    defer h.mu.RUnlock()
    if len(h.ring) == 0 { return "", false }
    hash := crc32.ChecksumIEEE([]byte(key))
    idx := sort.Search(len(h.ring), func(i int) bool { return h.ring[i] >= hash })
    if idx == len(h.ring) { idx = 0 }
    return h.nodes[h.ring[idx]], true
}

producerConfig := &kafka.ConfigMap{
    "bootstrap.servers":  "kafka-cluster:9092",
    "acks":               "all",
    "retries":            3,
    "batch.size":          524288,   // 512KB batch
    "linger.ms":           5,        // 5ms 等待聚合
    "compression.type":    "lz4",
    "max.in.flight":       5,
}

topic := "rtb-bid-log"
key := strconv.FormatInt(req.AdvertiserID, 10) // 广告主ID作为分区key

err := producer.Produce(&kafka.Message{
    TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: -1},
    Key:            []byte(key),
    Value:          serializeProto(req), // Protobuf 序列化，体积比 JSON 小 60%
}, deliveryChan)

go func() {
    for e := range deliveryChan {
        m := e.Event.(*kafka.Message)
        if m.TopicPartition.Error != nil {
            metrics.BidLogFailedCounter.Inc()
        }
    }
}()

syntax = "proto3";
package rtb;
option go_package = "pb/rtb";

service Attribution {
    // Bidirectional streaming: 客户端发送曝光事件，服务端推送归因结果
    rpc StreamAttribution(stream ImpressionEvent) returns (stream AttributionResult);
    // 服务端推送更新（如归因结果变化）
    rpc SubscribeUpdates(SubscribeRequest) returns (stream AttributionUpdate);
}

message ImpressionEvent {
    string request_id = 1;
    int64 advertiser_id = 2;
    string user_id = 3;
    int64 ad_id = 4;
    int64 timestamp = 5;
    map<string, string> context = 6;
}

message AttributionResult {
    string request_id = 1;
    int64 advertiser_id = 2;
    string user_id = 3;
    string attribution_channel = 4; // 归因渠道：自然搜索/付费搜索/直接访问
    double attribution_weight = 5;  // 归因权重（多触点归因）
    int64 attribution_time = 6;
}

message AttributionUpdate {
    string event_id = 1;
    string update_type = 2;
    AttributionResult result = 3;
}

type CircuitBreaker struct {
    name       string
    threshold  float64   // 失败率阈值，如 0.6
    window     int       // 统计窗口大小（请求数）
    sleepTime  time.Duration
    state      atomic.Int32 // 0=closed, 1=open, 2=half-open
    failures   int64
    successes  int64
    total      int64
    lastSwitch time.Time
}

const (StateClosed = 0; StateOpen = 1; StateHalfOpen = 2)

func (cb *CircuitBreaker) Call(fn func() error) error {
    switch cb.state.Load() {
    case StateOpen:
        if time.Since(cb.lastSwitch) > cb.sleepTime {
            cb.state.Store(StateHalfOpen)
        } else {
            return ErrCircuitOpen
        }
    case StateHalfOpen:
        // 只允许一个请求通过探测
        if !atomic.CompareAndSwapInt32(&cb.state, StateHalfOpen, StateHalfOpen) {
            return ErrCircuitOpen
        }
    }

    err := fn()
    atomic.AddInt64(&cb.total, 1)

    if err != nil {
        failures := atomic.AddInt64(&cb.failures, 1)
        if float64(failures)/float64(cb.total.Load()) > cb.threshold {
            cb.state.Store(StateOpen)
            cb.lastSwitch = time.Now()
            atomic.StoreInt64(&cb.failures, 0)
            atomic.StoreInt64(&cb.total, 0)
        }
        return err
    }

    atomic.AddInt64(&cb.successes, 1)
    if cb.state.Load() == StateHalfOpen && cb.successes.Load() >= 3 {
        cb.state.Store(StateClosed)
        atomic.StoreInt64(&cb.failures, 0)
        atomic.StoreInt64(&cb.total, 0)
        atomic.StoreInt64(&cb.successes, 0)
    }
    return nil
}