用Go实现gRPC加权随机负载均衡：从原理到etcd服务发现实战

1. 为什么需要自定义gRPC负载均衡策略

在分布式系统中，服务实例的性能往往存在差异。想象一下这样的场景：你有三台服务器，其中两台是高性能的云主机，另一台是性能较弱的旧机器。如果使用gRPC默认的轮询（round_robin）策略，每台服务器将平均分担请求量，这显然不是最优的资源利用方式。

加权随机负载均衡算法正是为解决这类问题而生。它允许我们：

• 根据服务器配置动态分配流量
• 高性能服务器承担更多请求
• 低性能服务器减少负载压力
• 实现资源利用率最大化

// 示例：服务器权重配置
servers := []Server{
    {Address: "10.0.0.1:50051", Weight: 5},  // 高性能服务器
    {Address: "10.0.0.2:50051", Weight: 3},  // 中等性能
    {Address: "10.0.0.3:50051", Weight: 1},  // 低性能服务器
}

2. gRPC负载均衡架构解析

gRPC的负载均衡机制主要包含两个核心接口：

2.1 V2PickerBuilder接口

type V2PickerBuilder interface {
    Build(info PickerBuildInfo) balancer.V2Picker
}

Build方法接收当前可用的子连接信息，返回一个V2Picker实例。这是我们实现自定义算法的关键切入点。

2.2 V2Picker接口

type V2Picker interface {
    Pick(info PickInfo) (PickResult, error)
}

Pick方法在每次RPC调用时被执行，决定使用哪个子连接发送请求。

权重存储技巧：gRPC的resolver.Address没有直接提供权重字段，但可以通过元数据(metadata)存储：

type AddrInfo struct {
    Weight int
}

func SetAddrInfo(addr resolver.Address, addrInfo AddrInfo) resolver.Address {
    addr.Attributes = attributes.New()
    addr.Attributes = addr.Attributes.WithValues(attributeKey{}, addrInfo)
    return addr
}

3. 加权随机算法实现

3.1 算法核心思想

加权随机法的核心是：权重越高的服务器，被选中的概率越大。我们采用"空间换时间"的策略：

1. 将权重转换为子连接数量（如权重3=3个子连接）
2. 随机选择一个子连接
3. 对应的服务器处理请求

// 示例：权重转换
权重5 → [s1, s1, s1, s1, s1]
权重3 → [s2, s2, s2]
权重1 → [s3]

3.2 完整实现代码

type weightedPicker struct {
    subConns []balancer.SubConn
    mu       sync.Mutex
}

func (p *weightedPicker) Pick(info balancer.PickInfo) (balancer.PickResult, error) {
    p.mu.Lock()
    defer p.mu.Unlock()
    
    if len(p.subConns) == 0 {
        return balancer.PickResult{}, balancer.ErrNoSubConnAvailable
    }
    
    index := rand.Intn(len(p.subConns))
    return balancer.PickResult{SubConn: p.subConns[index]}, nil
}

3.3 性能优化建议

1. 权重范围限制：建议权重设置在1-5之间，避免内存消耗过大
2. 随机数生成：使用crypto/rand替代math/rand提高随机性质量
3. 连接池管理：定期检查子连接健康状态，移除不可用连接

4. 集成etcd服务发现

4.1 服务注册实现

func (s *ServiceRegister) putKeyWithLease(lease int64) error {
    resp, err := s.cli.Grant(context.Background(), lease)
    if err != nil {
        return err
    }
    
    _, err = s.cli.Put(context.Background(), s.key, s.weight, 
        clientv3.WithLease(resp.ID))
    if err != nil {
        return err
    }
    
    // 保持租约活跃
    s.keepAliveChan, err = s.cli.KeepAlive(context.Background(), resp.ID)
    return err
}

4.2 服务发现实现

func (s *ServiceDiscovery) SetServiceList(key, val string) {
    addr := resolver.Address{Addr: strings.TrimPrefix(key, s.prefix)}
    
    weight, err := strconv.Atoi(val)
    if err != nil {
        weight = 1 // 默认权重
    }
    
    // 存储权重到元数据
    addr = weight.SetAddrInfo(addr, weight.AddrInfo{Weight: weight})
    s.serverList[key] = addr
    s.cc.UpdateState(resolver.State{Addresses: s.getServices()})
}

4.3 动态权重调整策略

通过etcd watch机制实现动态权重调整：

func (s *ServiceDiscovery) watcher() {
    rch := s.cli.Watch(context.Background(), s.prefix, clientv3.WithPrefix())
    for wresp := range rch {
        for _, ev := range wresp.Events {
            switch ev.Type {
            case mvccpb.PUT:
                s.SetServiceList(string(ev.Kv.Key), string(ev.Kv.Value))
            case mvccpb.DELETE:
                s.DelServiceList(string(ev.Kv.Key))
            }
        }
    }
}

5. 实战：完整示例与性能测试

5.1 服务端配置

func main() {
    // 启动gRPC服务器
    server := grpc.NewServer()
    pb.RegisterYourServiceServer(server, &service{})
    
    // 注册到etcd
    ser, err := etcdv3.NewServiceRegister(
        []string{"localhost:2379"}, 
        "service_name/127.0.0.1:50051",
        "3", // 权重值
        5)   // 租约TTL
    if err != nil {
        log.Fatal(err)
    }
    defer ser.Close()
    
    // 启动服务
    lis, _ := net.Listen("tcp", ":50051")
    server.Serve(lis)
}

5.2 客户端配置

func main() {
    // 创建etcd服务发现
    r := etcdv3.NewServiceDiscovery([]string{"localhost:2379"})
    resolver.Register(r)
    
    // 使用weight负载均衡策略
    conn, err := grpc.Dial(
        fmt.Sprintf("%s:///%s", r.Scheme(), "service_name"),
        grpc.WithBalancerName("weight"),
        grpc.WithInsecure(),
    )
    if err != nil {
        log.Fatal(err)
    }
    defer conn.Close()
    
    // 创建客户端并发送请求
    client := pb.NewYourServiceClient(conn)
    for i := 0; i < 100; i++ {
        resp, _ := client.YourMethod(context.Background(), &pb.Request{})
        log.Printf("Response from %s", resp.ServerId)
        time.Sleep(time.Second)
    }
}

5.3 性能测试结果

服务器权重	请求量(100次)	占比
5	56	56%
3	33	33%
1	11	11%

动态调整测试：

1. 移除权重5的服务器 → 请求自动转移到剩余服务器
2. 重新加入权重5的服务器 → 请求按权重比例重新分配

6. 高级优化技巧

6.1 平滑权重调整

为避免权重突变导致流量震荡，实现平滑过渡：

func smoothWeightChange(old, new int) int {
    // 每次最大调整幅度为±1
    if new > old {
        return old + 1
    } else if new < old {
        return old - 1
    }
    return old
}

6.2 基于QPS的动态权重

根据实时QPS自动调整权重：

func adjustWeightByQPS(currentWeight, currentQPS, maxQPS int) int {
    utilization := float64(currentQPS) / float64(maxQPS)
    if utilization > 0.8 {
        return currentWeight - 1
    } else if utilization < 0.5 {
        return currentWeight + 1
    }
    return currentWeight
}

6.3 熔断与降级机制

集成熔断器保护过载服务器：

type CircuitBreaker struct {
    failures    int
    maxFailures int
    cooldown    time.Duration
    lastFailure time.Time
}

func (cb *CircuitBreaker) Allow() bool {
    if time.Since(cb.lastFailure) > cb.cooldown {
        cb.failures = 0
        return true
    }
    return cb.failures < cb.maxFailures
}

7. 常见问题排查

1. 权重不生效：

   • 检查resolver.Address的元数据是否正确设置
   • 确认客户端使用了weight策略

2. etcd连接问题：

   • 检查etcd服务是否正常运行
   • 验证租约是否定期续期

3. 性能瓶颈：

   • 监控子连接列表大小
   • 检查随机数生成性能

4. 动态调整延迟：

   • 确保etcd watch机制正常工作
   • 检查网络延迟

// 调试日志示例
grpclog.Infof("Picker built with subConns: %v", info.ReadySCs)