微服务间通信异常处理机制：gRPC错误处理、重试策略与熔断器设计的完整解决方案

引言

在现代微服务架构中，服务间的通信是系统正常运行的核心环节。随着业务复杂度的增加，微服务之间的调用变得越来越频繁和复杂，通信异常成为影响系统稳定性的主要因素之一。gRPC作为Google开源的高性能RPC框架，在微服务通信中扮演着重要角色，但如何有效处理通信过程中的各种异常情况，确保系统的高可用性和稳定性，成为了每个微服务架构设计者必须面对的挑战。

本文将深入探讨微服务间通信中的异常处理机制，重点介绍gRPC错误处理机制、智能重试策略以及熔断器模式的设计与实现。通过理论分析与实践代码相结合的方式，为读者提供一套完整的异常处理解决方案，帮助构建更加健壮和可靠的微服务系统。

gRPC错误处理机制详解

1.1 gRPC错误模型概述

gRPC基于HTTP/2协议，提供了统一的错误处理机制。在gRPC中，错误通过状态码（Status Code）来表示，每个错误都包含一个标准的状态码和可选的错误信息。gRPC定义了以下几种主要的状态码：

• OK (0)：操作成功
• CANCELLED (1)：操作被取消
• UNKNOWN (2)：未知错误
• INVALID_ARGUMENT (3)：参数无效
• DEADLINE_EXCEEDED (4)：超时
• NOT_FOUND (5)：资源未找到
• ALREADY_EXISTS (6)：资源已存在
• PERMISSION_DENIED (7)：权限拒绝
• UNAUTHENTICATED (16)：未认证
• RESOURCE_EXHAUSTED (8)：资源耗尽
• FAILED_PRECONDITION (9)：前置条件失败
• ABORTED (10)：操作中止
• OUT_OF_RANGE (11)：超出范围
• UNIMPLEMENTED (12)：未实现
• INTERNAL (13)：内部错误
• UNAVAILABLE (14)：服务不可用
• DATA_LOSS (15)：数据丢失

1.2 gRPC错误处理最佳实践

在实际开发中，合理的错误处理策略能够显著提升系统的健壮性。以下是一些关键的最佳实践：

// Go语言中的gRPC错误处理示例
import (
    "google.golang.org/grpc/codes"
    "google.golang.org/grpc/status"
    "errors"
)

func handleServiceError(err error) error {
    if err == nil {
        return nil
    }
    
    // 获取gRPC状态码
    st, ok := status.FromError(err)
    if !ok {
        // 非gRPC错误，包装为内部错误
        return status.Error(codes.Internal, "internal server error")
    }
    
    switch st.Code() {
    case codes.NotFound:
        // 业务逻辑处理：资源未找到
        return status.Error(codes.NotFound, "requested resource not found")
    case codes.DeadlineExceeded:
        // 超时处理：记录日志并返回超时错误
        log.Printf("RPC call timeout: %v", st.Message())
        return status.Error(codes.DeadlineExceeded, "request timeout")
    case codes.Unavailable:
        // 服务不可用：可能需要重试
        log.Printf("Service unavailable: %v", st.Message())
        return status.Error(codes.Unavailable, "service temporarily unavailable")
    default:
        // 其他错误统一处理为内部错误
        log.Printf("Unexpected error: %v", st.Message())
        return status.Error(codes.Internal, "unexpected error occurred")
    }
}

1.3 自定义错误类型设计

为了更好地支持业务逻辑，可以设计自定义的错误类型来扩展gRPC的标准错误处理：

// 定义自定义业务错误
type BusinessError struct {
    Code    int32
    Message string
    Details map[string]interface{}
}

func (e *BusinessError) Error() string {
    return fmt.Sprintf("business error %d: %s", e.Code, e.Message)
}

func (e *BusinessError) GRPCStatus() *status.Status {
    return status.New(codes.Code(e.Code), e.Message)
}

// 使用示例
func validateUser(user *User) error {
    if user == nil {
        return &BusinessError{
            Code:    int32(codes.InvalidArgument),
            Message: "user cannot be null",
            Details: map[string]interface{}{
                "field": "user",
                "value": nil,
            },
        }
    }
    
    if user.Email == "" {
        return &BusinessError{
            Code:    int32(codes.InvalidArgument),
            Message: "email is required",
            Details: map[string]interface{}{
                "field": "email",
                "value": "",
            },
        }
    }
    
    return nil
}

智能重试策略设计与实现

2.1 重试策略的重要性

在微服务架构中，网络抖动、服务临时不可用等场景是不可避免的。合理的重试机制能够在不增加系统负载的情况下提高服务的可用性。然而，盲目地进行重试可能会加剧问题，因此需要设计智能的重试策略。

2.2 基于指数退避的重试策略

指数退避是一种经典的重试策略，通过逐渐增加重试间隔来避免对服务造成过大的压力：

// 智能重试器实现
type RetryPolicy struct {
    MaxRetries      int
    InitialDelay    time.Duration
    MaxDelay        time.Duration
    BackoffMultiplier float64
    RetryableCodes  []codes.Code
}

type RetryInterceptor struct {
    policy *RetryPolicy
}

func NewRetryInterceptor(policy *RetryPolicy) *RetryInterceptor {
    return &RetryInterceptor{policy: policy}
}

func (ri *RetryInterceptor) UnaryClientInterceptor(
    ctx context.Context,
    method string,
    req, reply interface{},
    cc *grpc.ClientConn,
    invoker grpc.UnaryInvoker,
    opts ...grpc.CallOption,
) error {
    
    var lastErr error
    
    for attempt := 0; attempt <= ri.policy.MaxRetries; attempt++ {
        if attempt > 0 {
            // 计算退避延迟时间
            delay := ri.calculateDelay(attempt)
            log.Printf("Retry attempt %d after delay %v", attempt, delay)
            time.Sleep(delay)
        }
        
        lastErr = invoker(ctx, method, req, reply, cc, opts...)
        
        if lastErr == nil {
            return nil // 成功返回
        }
        
        // 检查是否应该重试
        if !ri.shouldRetry(lastErr) {
            break
        }
    }
    
    return lastErr
}

func (ri *RetryInterceptor) calculateDelay(attempt int) time.Duration {
    // 指数退避算法
    delay := ri.policy.InitialDelay * time.Duration(math.Pow(ri.policy.BackoffMultiplier, float64(attempt)))
    
    // 限制最大延迟时间
    if delay > ri.policy.MaxDelay {
        return ri.policy.MaxDelay
    }
    
    return delay
}

func (ri *RetryInterceptor) shouldRetry(err error) bool {
    if err == nil {
        return false
    }
    
    st, ok := status.FromError(err)
    if !ok {
        // 非gRPC错误，默认不重试
        return false
    }
    
    // 检查状态码是否可重试
    for _, code := range ri.policy.RetryableCodes {
        if st.Code() == code {
            return true
        }
    }
    
    return false
}

// 使用示例
func main() {
    retryPolicy := &RetryPolicy{
        MaxRetries:         3,
        InitialDelay:       time.Millisecond * 100,
        MaxDelay:           time.Second * 5,
        BackoffMultiplier:  2.0,
        RetryableCodes:     []codes.Code{codes.Unavailable, codes.DeadlineExceeded, codes.Internal},
    }
    
    retryInterceptor := NewRetryInterceptor(retryPolicy)
    
    conn, err := grpc.Dial(
        "localhost:50051",
        grpc.WithInsecure(),
        grpc.WithUnaryInterceptor(retryInterceptor.UnaryClientInterceptor),
    )
    
    if err != nil {
        log.Fatalf("Failed to connect: %v", err)
    }
    
    defer conn.Close()
}

2.3 基于服务健康状态的动态重试

除了固定的重试策略，还可以根据服务的实时健康状态来动态调整重试行为：

// 服务健康检查器
type HealthChecker struct {
    serviceHealth map[string]*ServiceHealth
    mu            sync.RWMutex
}

type ServiceHealth struct {
    Status       ServiceStatus
    LastCheck    time.Time
    FailureCount int32
    SuccessCount int32
}

type ServiceStatus int

const (
    StatusHealthy ServiceStatus = iota
    StatusDegraded
    StatusUnhealthy
)

func NewHealthChecker() *HealthChecker {
    return &HealthChecker{
        serviceHealth: make(map[string]*ServiceHealth),
    }
}

func (hc *HealthChecker) UpdateServiceStatus(serviceName string, success bool) {
    hc.mu.Lock()
    defer hc.mu.Unlock()
    
    health, exists := hc.serviceHealth[serviceName]
    if !exists {
        health = &ServiceHealth{}
        hc.serviceHealth[serviceName] = health
    }
    
    health.LastCheck = time.Now()
    
    if success {
        health.SuccessCount++
        health.FailureCount = 0
        if health.Status == StatusUnhealthy {
            // 恢复健康状态
            health.Status = StatusHealthy
        }
    } else {
        health.FailureCount++
        if health.FailureCount >= 3 {
            health.Status = StatusUnhealthy
        } else if health.FailureCount >= 1 {
            health.Status = StatusDegraded
        }
    }
}

func (hc *HealthChecker) ShouldRetry(serviceName string, err error) bool {
    hc.mu.RLock()
    defer hc.mu.RUnlock()
    
    health, exists := hc.serviceHealth[serviceName]
    if !exists {
        return true // 未知服务默认可重试
    }
    
    // 不健康的服务不建议重试
    if health.Status == StatusUnhealthy {
        return false
    }
    
    // 根据错误类型决定是否重试
    st, ok := status.FromError(err)
    if !ok {
        return true
    }
    
    // 只对特定错误类型进行重试
    switch st.Code() {
    case codes.Unavailable, codes.DeadlineExceeded, codes.Internal:
        return health.Status != StatusUnhealthy
    default:
        return false
    }
}

// 带健康检查的重试器
type HealthAwareRetryInterceptor struct {
    policy      *RetryPolicy
    healthCheck *HealthChecker
}

func NewHealthAwareRetryInterceptor(policy *RetryPolicy, healthCheck *HealthChecker) *HealthAwareRetryInterceptor {
    return &HealthAwareRetryInterceptor{
        policy:      policy,
        healthCheck: healthCheck,
    }
}

func (hri *HealthAwareRetryInterceptor) UnaryClientInterceptor(
    ctx context.Context,
    method string,
    req, reply interface{},
    cc *grpc.ClientConn,
    invoker grpc.UnaryInvoker,
    opts ...grpc.CallOption,
) error {
    
    var lastErr error
    serviceName := extractServiceName(method)
    
    for attempt := 0; attempt <= hri.policy.MaxRetries; attempt++ {
        if attempt > 0 {
            delay := hri.calculateDelay(attempt)
            time.Sleep(delay)
        }
        
        lastErr = invoker(ctx, method, req, reply, cc, opts...)
        
        if lastErr == nil {
            // 成功更新健康状态
            hri.healthCheck.UpdateServiceStatus(serviceName, true)
            return nil
        }
        
        // 更新健康状态
        hri.healthCheck.UpdateServiceStatus(serviceName, false)
        
        // 检查是否应该重试
        if !hri.shouldRetry(serviceName, lastErr) {
            break
        }
    }
    
    return lastErr
}

func (hri *HealthAwareRetryInterceptor) shouldRetry(serviceName string, err error) bool {
    // 先检查健康状态
    if !hri.healthCheck.ShouldRetry(serviceName, err) {
        return false
    }
    
    // 然后检查错误类型
    st, ok := status.FromError(err)
    if !ok {
        return true
    }
    
    for _, code := range hri.policy.RetryableCodes {
        if st.Code() == code {
            return true
        }
    }
    
    return false
}

func extractServiceName(method string) string {
    // 从方法名中提取服务名
    parts := strings.Split(method, "/")
    if len(parts) >= 2 {
        return parts[1]
    }
    return "unknown"
}

熔断器模式设计与实现

3.1 熔断器模式原理

熔断器模式是微服务架构中的重要容错机制，它通过监控服务的调用成功率来决定是否允许请求通过。当失败率达到阈值时，熔断器会打开，直接拒绝请求以避免级联故障。

3.2 熔断器核心组件设计

// 熔断器状态机
type CircuitBreaker struct {
    state           CircuitState
    failureCount    int32
    successCount    int32
    lastFailureTime time.Time
    mutex           sync.RWMutex
    
    // 配置参数
    failureThreshold   int32
    timeout            time.Duration
    successThreshold   int32
    rollingWindow      int32
    resetTimeout       time.Duration
}

type CircuitState int

const (
    StateClosed CircuitState = iota
    StateOpen
    StateHalfOpen
)

func NewCircuitBreaker(config *CircuitBreakerConfig) *CircuitBreaker {
    return &CircuitBreaker{
        state:            StateClosed,
        failureThreshold: config.FailureThreshold,
        timeout:          config.Timeout,
        successThreshold: config.SuccessThreshold,
        rollingWindow:    config.RollingWindow,
        resetTimeout:     config.ResetTimeout,
    }
}

// 熔断器配置
type CircuitBreakerConfig struct {
    FailureThreshold int32
    Timeout          time.Duration
    SuccessThreshold int32
    RollingWindow    int32
    ResetTimeout     time.Duration
}

// 执行请求并更新状态
func (cb *CircuitBreaker) Execute(fn func() error) error {
    cb.mutex.RLock()
    state := cb.state
    cb.mutex.RUnlock()
    
    switch state {
    case StateClosed:
        return cb.executeClosed(fn)
    case StateOpen:
        return cb.executeOpen(fn)
    case StateHalfOpen:
        return cb.executeHalfOpen(fn)
    default:
        return fmt.Errorf("unknown circuit breaker state")
    }
}

func (cb *CircuitBreaker) executeClosed(fn func() error) error {
    err := fn()
    if err != nil {
        cb.recordFailure()
        return err
    }
    
    cb.recordSuccess()
    return nil
}

func (cb *CircuitBreaker) executeOpen(fn func() error) error {
    // 检查是否应该尝试半开状态
    if time.Since(cb.lastFailureTime) > cb.resetTimeout {
        cb.mutex.Lock()
        cb.state = StateHalfOpen
        cb.mutex.Unlock()
        return cb.executeHalfOpen(fn)
    }
    
    return fmt.Errorf("circuit breaker is open")
}

func (cb *CircuitBreaker) executeHalfOpen(fn func() error) error {
    err := fn()
    if err != nil {
        // 半开状态失败，重新打开
        cb.mutex.Lock()
        cb.state = StateOpen
        cb.lastFailureTime = time.Now()
        cb.mutex.Unlock()
        return err
    }
    
    // 半开状态成功，关闭熔断器
    cb.mutex.Lock()
    cb.state = StateClosed
    cb.failureCount = 0
    cb.successCount = 0
    cb.mutex.Unlock()
    return nil
}

func (cb *CircuitBreaker) recordFailure() {
    cb.mutex.Lock()
    defer cb.mutex.Unlock()
    
    cb.failureCount++
    cb.lastFailureTime = time.Now()
    
    if cb.failureCount >= cb.failureThreshold {
        cb.state = StateOpen
    }
}

func (cb *CircuitBreaker) recordSuccess() {
    cb.mutex.Lock()
    defer cb.mutex.Unlock()
    
    cb.successCount++
    
    // 如果成功次数达到阈值，重置熔断器
    if cb.successCount >= cb.successThreshold {
        cb.state = StateClosed
        cb.failureCount = 0
        cb.successCount = 0
    }
}

3.3 高级熔断器实现

为了提供更丰富的功能，可以实现一个支持统计和监控的高级熔断器：

// 带统计信息的熔断器
type AdvancedCircuitBreaker struct {
    *CircuitBreaker
    stats *CircuitBreakerStats
}

type CircuitBreakerStats struct {
    TotalRequests   int64
    SuccessfulCalls int64
    FailedCalls     int64
    ShortCircuited  int64
    HalfOpenAttempts int64
    LastResetTime   time.Time
    FailureRate     float64
}

func NewAdvancedCircuitBreaker(config *CircuitBreakerConfig) *AdvancedCircuitBreaker {
    return &AdvancedCircuitBreaker{
        CircuitBreaker: NewCircuitBreaker(config),
        stats:          &CircuitBreakerStats{},
    }
}

func (acb *AdvancedCircuitBreaker) Execute(fn func() error) error {
    acb.stats.TotalRequests++
    
    cb := acb.CircuitBreaker
    cb.mutex.RLock()
    state := cb.state
    cb.mutex.RUnlock()
    
    switch state {
    case StateClosed:
        return acb.executeClosedWithStats(fn)
    case StateOpen:
        acb.stats.ShortCircuited++
        return fmt.Errorf("circuit breaker is open")
    case StateHalfOpen:
        acb.stats.HalfOpenAttempts++
        return acb.executeHalfOpenWithStats(fn)
    default:
        return fmt.Errorf("unknown circuit breaker state")
    }
}

func (acb *AdvancedCircuitBreaker) executeClosedWithStats(fn func() error) error {
    err := fn()
    
    if err != nil {
        acb.recordFailureWithStats()
        return err
    }
    
    acb.recordSuccessWithStats()
    return nil
}

func (acb *AdvancedCircuitBreaker) recordFailureWithStats() {
    cb := acb.CircuitBreaker
    cb.mutex.Lock()
    defer cb.mutex.Unlock()
    
    cb.failureCount++
    cb.lastFailureTime = time.Now()
    
    acb.stats.FailedCalls++
    
    if cb.failureCount >= cb.failureThreshold {
        cb.state = StateOpen
    }
    
    acb.updateFailureRate()
}

func (acb *AdvancedCircuitBreaker) recordSuccessWithStats() {
    cb := acb.CircuitBreaker
    cb.mutex.Lock()
    defer cb.mutex.Unlock()
    
    cb.successCount++
    acb.stats.SuccessfulCalls++
    
    if cb.successCount >= cb.successThreshold {
        cb.state = StateClosed
        cb.failureCount = 0
        cb.successCount = 0
    }
    
    acb.updateFailureRate()
}

func (acb *AdvancedCircuitBreaker) updateFailureRate() {
    total := acb.stats.TotalRequests
    if total > 0 {
        acb.stats.FailureRate = float64(acb.stats.FailedCalls) / float64(total)
    }
}

// 获取统计信息
func (acb *AdvancedCircuitBreaker) GetStats() *CircuitBreakerStats {
    return &CircuitBreakerStats{
        TotalRequests:   atomic.LoadInt64(&acb.stats.TotalRequests),
        SuccessfulCalls: atomic.LoadInt64(&acb.stats.SuccessfulCalls),
        FailedCalls:     atomic.LoadInt64(&acb.stats.FailedCalls),
        ShortCircuited:  atomic.LoadInt64(&acb.stats.ShortCircuited),
        HalfOpenAttempts: atomic.LoadInt64(&acb.stats.HalfOpenAttempts),
        LastResetTime:   acb.stats.LastResetTime,
        FailureRate:     acb.stats.FailureRate,
    }
}

// 熔断器监控接口
type CircuitBreakerMonitor struct {
    breakers map[string]*AdvancedCircuitBreaker
    mu       sync.RWMutex
}

func NewCircuitBreakerMonitor() *CircuitBreakerMonitor {
    return &CircuitBreakerMonitor{
        breakers: make(map[string]*AdvancedCircuitBreaker),
    }
}

func (cbm *CircuitBreakerMonitor) RegisterBreaker(name string, breaker *AdvancedCircuitBreaker) {
    cbm.mu.Lock()
    defer cbm.mu.Unlock()
    
    cbm.breakers[name] = breaker
}

func (cbm *CircuitBreakerMonitor) GetBreakerStats(name string) (*CircuitBreakerStats, error) {
    cbm.mu.RLock()
    defer cbm.mu.RUnlock()
    
    breaker, exists := cbm.breakers[name]
    if !exists {
        return nil, fmt.Errorf("breaker %s not found", name)
    }
    
    return breaker.GetStats(), nil
}

func (cbm *CircuitBreakerMonitor) GetAllStats() map[string]*CircuitBreakerStats {
    cbm.mu.RLock()
    defer cbm.mu.RUnlock()
    
    stats := make(map[string]*CircuitBreakerStats)
    for name, breaker := range cbm.breakers {
        stats[name] = breaker.GetStats()
    }
    
    return stats
}

完整的异常处理框架设计

4.1 综合异常处理架构

结合前面介绍的gRPC错误处理、智能重试和熔断器机制，我们可以构建一个完整的异常处理框架：

// 异常处理框架主结构
type ExceptionHandlingFramework struct {
    retryInterceptor      *HealthAwareRetryInterceptor
    circuitBreakerMonitor *CircuitBreakerMonitor
    errorLogger           *ErrorLogger
    metricsCollector      *MetricsCollector
}

type ServiceConfig struct {
    Name             string
    RetryPolicy      *RetryPolicy
    CircuitBreaker   *CircuitBreakerConfig
    Timeout          time.Duration
    MaxRetries       int32
}

func NewExceptionHandlingFramework() *ExceptionHandlingFramework {
    return &ExceptionHandlingFramework{
        circuitBreakerMonitor: NewCircuitBreakerMonitor(),
        errorLogger:           NewErrorLogger(),
        metricsCollector:      NewMetricsCollector(),
    }
}

// 配置服务
func (ehf *ExceptionHandlingFramework) ConfigureService(config *ServiceConfig) error {
    // 创建重试器
    retryPolicy := &RetryPolicy{
        MaxRetries:         int(config.MaxRetries),
        InitialDelay:       time.Millisecond * 100,
        MaxDelay:           time.Second * 5,
        BackoffMultiplier:  2.0,
        RetryableCodes:     []codes.Code{codes.Unavailable, codes.DeadlineExceeded, codes.Internal},
    }
    
    // 创建熔断器
    circuitBreaker := NewAdvancedCircuitBreaker(config.CircuitBreaker)
    
    // 注册到监控器
    ehf.circuitBreakerMonitor.RegisterBreaker(config.Name, circuitBreaker)
    
    // 创建重试拦截器
    retryInterceptor := NewHealthAwareRetryInterceptor(retryPolicy, 
        ehf.circuitBreakerMonitor)
    
    return nil
}

// 统一的gRPC客户端配置
func (ehf *ExceptionHandlingFramework) CreateGrpcClient(target string, serviceName string) (*grpc.ClientConn, error) {
    // 获取服务配置
    // 这里简化处理，实际应用中应该从配置中心获取
    
    return grpc.Dial(
        target,
        grpc.WithInsecure(),
        grpc.WithUnaryInterceptor(
            ehf.createServiceInterceptor(serviceName),
        ),
    )
}

func (ehf *ExceptionHandlingFramework) createServiceInterceptor(serviceName string) grpc.UnaryClientInterceptor {
    return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
        // 记录开始时间
        startTime := time.Now()
        
        // 执行熔断器检查
        circuitBreaker, err := ehf.getCircuitBreaker(serviceName)
        if err != nil {
            return err
        }
        
        // 使用熔断器执行请求
        err = circuitBreaker.Execute(func() error {
            // 记录调用开始
            ehf.metricsCollector.RecordCallStart(serviceName, method)
            
            // 执行实际的gRPC调用
            callErr := invoker(ctx, method, req, reply, cc, opts...)
            
            // 记录调用结束和结果
            duration := time.Since(startTime)
            if callErr != nil {
                ehf.metricsCollector.RecordCallError(serviceName, method, duration, callErr)
                ehf.errorLogger.LogError(serviceName, method, callErr)
            } else {
                ehf.metricsCollector.RecordCallSuccess(serviceName, method, duration)
            }
            
            return callErr
        })
        
        // 如果熔断器打开，记录错误
        if err != nil && strings.Contains(err.Error(), "circuit breaker is open") {
            ehf.errorLogger.LogCircuitBreakerOpen(serviceName, method)
        }
        
        return err
    }
}

func (ehf *ExceptionHandlingFramework) getCircuitBreaker(serviceName string) (*AdvancedCircuitBreaker, error) {
    ehf.circuitBreakerMonitor.mu.RLock()
    defer ehf.circuitBreakerMonitor.mu.RUnlock()
    
    breaker, exists := ehf.circuitBreakerMonitor.breakers[serviceName]
    if !exists {
        return nil, fmt.Errorf("no circuit breaker configured for service %s", serviceName)
    }
    
    return breaker, nil
}

4.2 实际应用示例

下面是一个完整的实际应用示例，展示如何使用上述框架：

// 服务消费者示例
type OrderServiceClient struct {
    conn *grpc.ClientConn
    client pb.OrderServiceClient
}

func NewOrderServiceClient(target string) (*OrderServiceClient, error) {
    // 创建异常处理框架
    framework := NewExceptionHandlingFramework()
    
    // 配置服务
    config := &ServiceConfig{
        Name: "order-service",
        RetryPolicy: &RetryPolicy{
            MaxRetries:         3,
            InitialDelay:       time.Millisecond * 100,
            MaxDelay:           time.Second * 5,
            BackoffMultiplier:  2.0,
            RetryableCodes:     []codes.Code{codes.Unavailable, codes.DeadlineExceeded, codes.Internal},
        },
        CircuitBreaker: &CircuitBreakerConfig{
            FailureThreshold: 5,
            Timeout:          time.Second * 30,
            SuccessThreshold: 3,
            RollingWindow:    100,
            ResetTimeout:     time.Minute * 5,
        },
        Timeout:    time.Second * 10,
        MaxRetries: 3,
    }
    
    framework.ConfigureService(config)
    
    // 创建gRPC连接
    conn