弹性治理

// gRPC:通过 context deadline 控制超时
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()

resp, err := client.GetOrder(ctx, req)
if status.Code(err) == codes.DeadlineExceeded {
    // 超时 → 降级处理
}

// 指数退避 + 抖动
func backoff(attempt int) time.Duration {
    base := time.Second * time.Duration(1<<attempt) // 1s, 2s, 4s, 8s...
    jitter := time.Duration(rand.Int63n(int64(base) / 2)) // 0 ~ base/2 的随机抖动
    return base + jitter
}

// gRPC service config 配置自动重试
serviceConfig := `{
    "methodConfig": [{
        "name": [{"service": "order.v1.OrderService"}],
        "retryPolicy": {
            "maxAttempts": 3,
            "initialBackoff": "0.1s",
            "maxBackoff": "1s",
            "backoffMultiplier": 2,
            "retryableStatusCodes": ["UNAVAILABLE"]
        }
    }]
}`

import "github.com/sony/gobreaker"

cb := gobreaker.NewCircuitBreaker(gobreaker.Settings{
    Name:        "user-service",
    MaxRequests: 3,                    // Half-Open 时放过的探测请求数
    Interval:    10 * time.Second,     // Closed 状态下错误计数窗口
    Timeout:     30 * time.Second,     // Open → Half-Open 的冷却时间
    ReadyToTrip: func(counts gobreaker.Counts) bool {
        // 10 次请求中失败率 > 60% → 熔断
        return counts.Requests > 10 &&
            float64(counts.TotalFailures)/float64(counts.Requests) > 0.6
    },
})

// 使用熔断器包裹调用
result, err := cb.Execute(func() (interface{}, error) {
    return client.GetUser(ctx, req)
})
if err == gobreaker.ErrOpenState {
    // 熔断中 → 返回降级结果
    return cachedUser, nil
}

import "golang.org/x/time/rate"

// 每秒 100 个请求,允许突发 20
limiter := rate.NewLimiter(rate.Limit(100), 20)

func rateLimitInterceptor(
    ctx context.Context,
    req interface{},
    info *grpc.UnaryServerInfo,
    handler grpc.UnaryHandler,
) (interface{}, error) {
    if !limiter.Allow() {
        return nil, status.Error(codes.ResourceExhausted, "rate limit exceeded")
    }
    return handler(ctx, req)
}

// Go 中用 channel 实现信号量隔离
type Bulkhead struct {
    sem chan struct{}
}

func NewBulkhead(maxConcurrency int) *Bulkhead {
    return &Bulkhead{sem: make(chan struct{}, maxConcurrency)}
}

func (b *Bulkhead) Execute(ctx context.Context, fn func() error) error {
    select {
    case b.sem <- struct{}{}:
        defer func() { <-b.sem }()
        return fn()
    case <-ctx.Done():
        return status.Error(codes.ResourceExhausted, "bulkhead full")
    }
}

// 为每个下游创建独立的隔离舱
var (
    userBulkhead    = NewBulkhead(50)  // 用户服务最多 50 并发
    paymentBulkhead = NewBulkhead(20)  // 支付服务最多 20 并发
)

func getRecommendations(ctx context.Context, userID string) ([]*Product, error) {
    products, err := recommendClient.GetForUser(ctx, userID)
    if err != nil {
        // 降级:返回热门商品缓存
        log.Warn("recommend service degraded", "err", err)
        return getHotProductsFromCache()
    }
    return products, nil
}