「产品矩阵平台」高并发与可用性架构

package ratelimiter

import (
    "context"
    "fmt"
    "sync"
    "time"
)

type TokenBucket struct {
    capacity   int
    tokens     int
    refillRate time.Duration // 多长时间补充 1 个 token
    lastRefill time.Time
    mu         sync.Mutex
}

func NewTokenBucket(capacity int, refillRate time.Duration) *TokenBucket {
    return &TokenBucket{
        capacity:   capacity,
        tokens:     capacity,
        refillRate: refillRate,
        lastRefill: time.Now(),
    }
}

func (tb *TokenBucket) Allow(n int) bool {
    tb.mu.Lock()
    defer tb.mu.Unlock()

    now := time.Now()
    elapsed := now.Sub(tb.lastRefill)
    addTokens := int(elapsed / tb.refillRate)
    if addTokens > 0 {
        tb.tokens = min(tb.capacity, tb.tokens+addTokens)
        tb.lastRefill = now
    }

    if tb.tokens >= n {
        tb.tokens -= n
        return true
    }
    return false
}

// DistributedLimiter 基于 Redis 的分布式限流
func DistributedLimiter(redis RedisClient, key string, capacity int, window time.Duration) bool {
    lua := `
        local key = KEYS[1]
        local capacity = tonumber(ARGV[1])
        local window = tonumber(ARGV[2])
        local current = redis.call('INCR', key)
        if current == 1 then
            redis.call('EXPIRE', key, window)
        end
        return current <= capacity
    `
    ok, err := redis.Eval(context.Background(), lua, []string{key}, capacity, int(window.Seconds())).Result()
    return err == nil && ok.(int64) == 1
}

// 多维度限流中间件
func RateLimitMiddleware(redis RedisClient) func(http.Handler) http.Handler {
    return func(next http.Handler) http.Handler {
        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
            ctx := r.Context()
            tenant := GetTenant(ctx)
            apiKey := r.URL.Path

            // 接口级限流
            if !DistributedLimiter(redis, "ratelimit:api:"+apiKey, 100, time.Minute) {
                w.WriteHeader(http.StatusTooManyRequests)
                return
            }
            // 租户级限流
            if !DistributedLimiter(redis, "ratelimit:tenant:"+tenant.TenantID, 1000, time.Minute) {
                w.WriteHeader(http.StatusTooManyRequests)
                return
            }
            next.ServeHTTP(w, r)
        })
    }
}

package circuitbreaker

import (
    "errors"
    "sync"
    "time"
)

type State int

const (
    StateClosed    State = iota // 正常
    StateOpen                   // 熔断开启
    StateHalfOpen              // 半开探测
)

// CircuitBreaker 保护外部依赖调用
type CircuitBreaker struct {
    failureThreshold int
    successThreshold int
    timeout          time.Duration
    state            State
    failures         int
    successes        int
    lastFailureTime  time.Time
    mu               sync.Mutex
}

func New(failureThreshold, successThreshold int, timeout time.Duration) *CircuitBreaker {
    return &CircuitBreaker{
        failureThreshold: failureThreshold,
        successThreshold: successThreshold,
        timeout:          timeout,
        state:            StateClosed,
    }
}

func (cb *CircuitBreaker) Call(fn func() error) error {
    cb.mu.Lock()
    switch cb.state {
    case StateOpen:
        if time.Since(cb.lastFailureTime) > cb.timeout {
            cb.state = StateHalfOpen
            cb.failures = 0
            cb.successes = 0
        } else {
            cb.mu.Unlock()
            return errors.New("circuit breaker is open")
        }
    case StateHalfOpen:
        // 继续执行下面的调用逻辑
    }
    cb.mu.Unlock()

    err := fn()

    cb.mu.Lock()
    defer cb.mu.Unlock()
    if err != nil {
        cb.failures++
        cb.lastFailureTime = time.Now()
        if cb.failures >= cb.failureThreshold {
            cb.state = StateOpen
        }
        return err
    }
    cb.successes++
    if cb.state == StateHalfOpen && cb.successes >= cb.successThreshold {
        cb.state = StateClosed
        cb.failures = 0
        cb.successes = 0
    }
    return nil
}

// 使用示例：保护 SMS 调用
func (svc *NotificationService) SendSMS(ctx context.Context, phone, message string) error {
    cb := cbStore.GetOrCreate("sms:" + svc.provider) // 按供应商共享 CB
    return cb.Call(func() error {
        return svc.smsClient.Send(ctx, phone, message)
    })
}

package health

import (
    "context"
    "database/sql"
    "net/http"
    "time"
)

type Checker struct {
    db      *sql.DB
    redis   RedisPinger
    kafka   KafkaPinger
}

type Status struct {
    Status    string                 `json:"status"`
    Timestamp time.Time              `json:"timestamp"`
    Checks    map[string]CheckResult `json:"checks"`
}

type CheckResult struct {
    Status string `json:"status"`
    Error  string `json:"error,omitempty"`
}

func (c *Checker) Liveness(w http.ResponseWriter, r *http.Request) {
    // 仅判断进程存活
    _ = json.NewEncoder(w).Encode(Status{Status: "ok", Timestamp: time.Now()})
}

func (c *Checker) Readiness(w http.ResponseWriter, r *http.Request) {
    ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
    defer cancel()

    checks := make(map[string]CheckResult)
    healthy := true

    // 数据库连通性
    if c.db != nil {
        start := time.Now()
        if err := c.db.PingContext(ctx); err != nil {
            checks["database"] = CheckResult{Status: "down", Error: err.Error()}
            healthy = false
        } else {
            checks["database"] = CheckResult{Status: "ok"}
        }
        checks["database_ping_ms"] = CheckResult{Status: time.Since(start).String()}
    }

    // Redis
    if c.redis != nil {
        if err := c.redis.Ping(ctx).Err(); err != nil {
            checks["redis"] = CheckResult{Status: "down", Error: err.Error()}
            healthy = false
        } else {
            checks["redis"] = CheckResult{Status: "ok"}
        }
    }

    // Kafka
    if c.kafka != nil {
        if err := c.kafka.Ping(); err != nil {
            checks["kafka"] = CheckResult{Status: "down", Error: err.Error()}
            healthy = false
        } else {
            checks["kafka"] = CheckResult{Status: "ok"}
        }
    }

    status := "ready"
    if !healthy {
        status = "not_ready"
        w.WriteHeader(http.StatusServiceUnavailable)
    }

    _ = json.NewEncoder(w).Encode(Status{
        Status:    status,
        Timestamp: time.Now(),
        Checks:    checks,
    })
}

package leader

import (
    "context"
    "fmt"
    "time"
)

type Election struct {
    redis      RedisClient
    key        string
    nodeID     string
    leaseTTL   time.Duration
    renewCh    chan struct{}
    isLeader   bool
}

func (e *Election) Campaign(ctx context.Context) error {
    // 尝试获取锁
    ok, err := e.redis.SetNX(ctx, e.key, e.nodeID, e.leaseTTL).Result()
    if err != nil {
        return err
    }
    if ok {
        e.isLeader = true
        go e.renewLoop(ctx)
        return nil
    }
    // 检查当前持有者是否是自己（防止重启后重入）
    val, _ := e.redis.Get(ctx, e.key).Result()
    if val == e.nodeID {
        e.isLeader = true
        go e.renewLoop(ctx)
        return nil
    }
    e.isLeader = false
    return fmt.Errorf("another node holds leadership")
}

func (e *Election) renewLoop(ctx context.Context) {
    ticker := time.NewTicker(e.leaseTTL / 3)
    defer ticker.Stop()
    for {
        select {
        case <-ctx.Done():
            e.redis.Del(ctx, e.key)
            return
        case <-ticker.C:
            e.redis.Expire(ctx, e.key, e.leaseTTL)
            // 可选：写带 TTL 的续租日志
        }
    }
}

func (e *Election) IsLeader() bool { return e.isLeader }

// 使用示例：定时归档任务仅由 Leader 执行
func (svc *ArchiveService) Schedule(ctx context.Context) {
    ticker := time.NewTicker(1 * time.Hour)
    defer ticker.Stop()
    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            if !svc.election.IsLeader() {
                continue
            }
            _ = svc.RunArchive(ctx)
        }
    }
}

package benchmark

import (
    "testing"
    "net/http"
    "net/http/httptest"
)

// 模块级基准测试
func BenchmarkOrderCreate(b *testing.B) {
    svc := setupTestOrderService()
    ctx := context.Background()
    cmd := CreateOrderCommand{
        UserID:    "u_001",
        ProductID: "p_001",
        Quantity:  1,
    }
    b.ResetTimer()
    for i := 0; i < b.N; i++ {
        _, _ = svc.CreateOrder(ctx, cmd)
    }
}

// 接口级负载测试（搭配 go test -bench=. 或 k6）
func BenchmarkOrderHandler(b *testing.B) {
    handler := setupTestHandler()
    body := `{"product_id":"p_001","quantity":1}`
    b.ResetTimer()
    for i := 0; i < b.N; i++ {
        req := httptest.NewRequest(http.MethodPost, "/api/v1/orders", strings.NewReader(body))
        req.Header.Set("Content-Type", "application/json")
        req.Header.Set("X-App-ID", "app_001")
        w := httptest.NewRecorder()
        handler.ServeHTTP(w, req)
    }
}