grpc-resolver

Go go grpc resolver

发布时间 : 2025-03-13 10:27

字数:2.4k 阅读 :

1. resolver.Resolver 预留接口
1. 🔹 Resolver 关键方法
2. resolver.Builder 接口（解析器构造器）
1. 🔹 Builder 关键方法
3. resolver.ClientConn 预留接口
1. 🔹 ClientConn 关键方法
4. 解析器的 resolver.State 结构
5. 组合：完整的 gRPC 自定义解析器
1. 完整代码示例
6. 运行流程
7. 实验
9. 总结

在 gRPC-Go 中，预留的 resolver.Resolver 接口 定义了 自定义解析器（Resolver） 需要实现的基本方法。gRPC 使用 resolver.Resolver 来解析目标地址（如 custom:///service），并动态提供服务器地址列表。

1. `resolver.Resolver` 预留接口

gRPC 预留的 resolver.Resolver 接口定义如下（google.golang.org/grpc/resolver）：

package resolver

// Resolver 定义了解析 gRPC 目标地址的接口
type Resolver interface {
    ResolveNow(ResolveNowOptions) // 立即解析服务地址
    Close()                       // 关闭解析器，释放资源
}

// 立即解析的选项（目前为空）
type ResolveNowOptions struct{}

🔹 `Resolver` 关键方法

方法	作用
`ResolveNow(ResolveNowOptions)`	触发服务地址解析，用于监听变更
`Close()`	关闭解析器，释放资源

2. `resolver.Builder` 接口（解析器构造器）

在 gRPC 中，所有的解析器都需要先通过 resolver.Builder 进行注册：

package resolver

// Builder 是解析器的工厂接口
type Builder interface {
    Build(target Target, cc ClientConn, opts BuildOptions) (Resolver, error) // 创建解析器
    Scheme() string // 返回解析器的 scheme（如 "consul", "etcd"）
}

🔹 `Builder` 关键方法

方法	作用
`Build(target, cc, opts)`	创建解析器实例，用于解析 gRPC 地址
`Scheme()`	返回解析器的 scheme，如 `"consul"`、`"etcd"`、`"custom"`

3. `resolver.ClientConn` 预留接口

resolver.ClientConn 用于通知 gRPC 负载均衡器 解析后的服务地址：

package resolver

// ClientConn 是 Resolver 和 gRPC 负载均衡之间的通信接口
type ClientConn interface {
    UpdateState(State) error  // 更新 gRPC 服务器地址列表
    ReportError(error)        // 通知 gRPC 解析错误
    NewAddress([]Address)     // 过时方法（建议用 UpdateState）
    NewServiceConfig(string)  // 过时方法
}

🔹 `ClientConn` 关键方法

方法	作用
`UpdateState(State)`	更新解析出的服务器列表
`ReportError(error)`	解析失败时通知 gRPC

4. 解析器的 `resolver.State` 结构

当 resolver.Resolver 解析完 gRPC 目标地址后，需要将解析结果传递给 gRPC 负载均衡器，格式如下：

package resolver

// State 代表解析后的 gRPC 服务器状态
type State struct {
    Addresses []Address // 解析出的服务器地址
}

示例：

state := resolver.State{
    Addresses: []resolver.Address{
        {Addr: "127.0.0.1:50051"},
        {Addr: "127.0.0.1:50052"},
    },
}
cc.UpdateState(state) // 更新解析结果

5. 组合：完整的 gRPC 自定义解析器

根据 gRPC 预留的 resolver.Resolver 接口，我们可以实现一个 自定义的 Consul 解析器：

完整代码示例

package main

import (
    "fmt"
    "github.com/hashicorp/consul/api"
    "log"
    "net"
    "sync"

    "context"
    "google.golang.org/grpc"
    "google.golang.org/grpc/resolver"
    "test/proto"
)

// 自定义解析器的 scheme
const myScheme = "consul"

var consulAddress string = "127.0.0.1:8500"

// 构造器（Builder）实现 `resolver.Builder`
type myResolverBuilder struct{}

// Target url;
func (b *myResolverBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
    log.Printf("解析目标: %s", target.Endpoint())
    r := &myResolver{
        target: target,
        cc:     cc,
    }
    r.ResolveNow(resolver.ResolveNowOptions{}) // 立即解析
    return r, nil
}

func (b *myResolverBuilder) Scheme() string {
    return myScheme
}

// 解析器（Resolver）实现 `resolver.Resolver`
type myResolver struct {
    target resolver.Target
    cc     resolver.ClientConn
    mu     sync.Mutex
}

// 解析服务地址
func (r *myResolver) ResolveNow(opts resolver.ResolveNowOptions) {
    r.mu.Lock()
    defer r.mu.Unlock()

    // 假设这里从 Consul 获取服务地址
    addresses, err := queryConsul(consulAddress, r.target.Endpoint())
    if err != nil {
        panic(err)
    }

    // 更新 gRPC 服务器地址
    r.cc.UpdateState(resolver.State{Addresses: addresses})
    log.Println("解析器更新完成")
}

// 关闭解析器
func (r *myResolver) Close() {}

// 注册解析器
func init() {
    consulAddress, _ = GetLocalIP()
    consulAddress = fmt.Sprintf("%s:8500", consulAddress)
    resolver.Register(&myResolverBuilder{})
}

// 从 Consul 查询 `serviceName` 并返回可用的 gRPC 服务器列表
func queryConsul(consulAddress string, serviceName string) ([]resolver.Address, error) {
    // 创建 Consul 客户端
    consulConfig := api.DefaultConfig()
    consulConfig.Address = consulAddress
    client, err := api.NewClient(consulConfig)
    if err != nil {
        return nil, err
    }

    // 查询 Consul 中健康的 gRPC 服务
    services, _, err := client.Health().Service(serviceName, "", true, nil)
    if err != nil {
        return nil, err
    }

    // 转换为 gRPC `resolver.Address`
    var addresses []resolver.Address
    for _, service := range services {
        addr := service.Service.Address
        port := service.Service.Port
        addresses = append(addresses, resolver.Address{
            Addr:       addr + ":" + fmt.Sprintf("%d", port),
            ServerName: serviceName})
    }

    // 如果没有可用服务，返回错误
    if len(addresses) == 0 {
        return nil, fmt.Errorf("未找到任何健康的 %s 服务", serviceName)
    }

    for _, address := range addresses {
        fmt.Println(address)
    }

    return addresses, nil
}

// 获取本机 IPv4 地址（第一个非 127.0.0.1 的）
func GetLocalIP() (string, error) {
    interfaces, err := net.Interfaces()
    if err != nil {
        return "", err
    }

    for _, iface := range interfaces {
        // 忽略未启用的网卡
        if iface.Flags&net.FlagUp == 0 || iface.Flags&net.FlagLoopback != 0 {
            continue
        }

        addrs, err := iface.Addrs()
        if err != nil {
            continue
        }

        for _, addr := range addrs {
            ipNet, ok := addr.(*net.IPNet)
            if !ok || ipNet.IP.IsLoopback() || ipNet.IP.To4() == nil {
                continue
            }
            return ipNet.IP.String(), nil // 找到第一个 IPv4 地址直接返回
        }
    }

    return "", fmt.Errorf("no valid IPv4 address found")
}

func main() {
    // 使用自定义解析器
    conn, err := grpc.Dial(
        "consul:///mx_shop_users",
        grpc.WithInsecure(),
        grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"round_robin"}`),
    )
    if err != nil {
        log.Fatalf("连接 gRPC 服务器失败: %v", err)
    }
    defer conn.Close()

    log.Println("gRPC 连接成功")
    UserSrvClient := proto.NewUserClient(conn)
    rsp, err := UserSrvClient.GetUserList(context.Background(), &proto.PageInfo{
        Pn:    1,
        PSize: 2,
    })
    if err != nil {
        panic(err)
    }
    for index, data := range rsp.Data {
        fmt.Println(index, data)
    }
}

6. 运行流程

grpc.Dial("custom:///myservice") 触发 gRPC 解析器。
gRPC 调用 customResolverBuilder.Build() 创建解析器。
ResolveNow() 解析目标地址，返回 服务器列表（如 127.0.0.1:50051、127.0.0.1:50052）。
UpdateState() 更新服务器地址列表，gRPC 负载均衡器 开始均衡请求。
所有 gRPC 请求 都会自动负载均衡到解析出来的服务器。

7. 实验

user.proto

syntax = "proto3";

import "google/protobuf/empty.proto";

option go_package = ".;proto";

service User {
//  传递一些信息进来
  rpc GetUserList(PageInfo) returns (UserListResponse); // 用户的列表
  rpc GetUserByMobile(MobileRequest) returns (UserInfoResponse); // 通过mobile查询用户
  rpc GetUserById(IdRequest) returns (UserInfoResponse); // 通过id查询用户
  rpc CreateUser(CreateUserInfo) returns (UserInfoResponse); // 添加用户
  rpc UpdateUser(UpdateUserInfo) returns (google.protobuf.Empty); // 更新用户
  rpc CheckPassWord(PasswordCheckInfo) returns (CheckResponse);// 检查
}


message PasswordCheckInfo {
  string password = 1;
  string encryptedPassword = 2;
}
message CheckResponse {
  bool success = 1;
}


message PageInfo {
  uint32 pn = 1; // 页码
  uint32 pSize = 2; // 每一页的大小
}

message UserInfoResponse {
  int32 id = 1;
  string password = 2;
  string mobile = 3;
  string nickName = 4;
  uint64 birthDay = 5;
  string gender = 6;
  int32 role = 7;
}

message UserListResponse {
  int32 total = 1;
  repeated UserInfoResponse data = 2;
}

message MobileRequest {
  string mobile = 1;
}

message IdRequest {
  int32 id = 1;
}

message CreateUserInfo {
  string nickName = 1;
  string password = 2;
  string mobile = 3;
}

message UpdateUserInfo {
  int32 id = 1;
  string nickName = 2;
  string gender = 3;
  uint64 birthday = 4;
}

main.go

package main

import (
    "fmt"
    "github.com/hashicorp/consul/api"
    "log"
    "net"
    "sync"

    "context"
    "google.golang.org/grpc"
    "google.golang.org/grpc/resolver"
    "test/proto"
)

// 自定义解析器的 scheme
const myScheme = "consul"

var consulAddress string = "127.0.0.1:8500"

// 构造器（Builder）实现 `resolver.Builder`
type myResolverBuilder struct{}

// Target url;
func (b *myResolverBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
    log.Printf("解析目标: %s", target.Endpoint())
    r := &myResolver{
        target: target,
        cc:     cc,
    }
    r.ResolveNow(resolver.ResolveNowOptions{}) // 立即解析
    return r, nil
}

func (b *myResolverBuilder) Scheme() string {
    return myScheme
}

// 解析器（Resolver）实现 `resolver.Resolver`
type myResolver struct {
    target resolver.Target
    cc     resolver.ClientConn
    mu     sync.Mutex
}

// 解析服务地址
func (r *myResolver) ResolveNow(opts resolver.ResolveNowOptions) {
    r.mu.Lock()
    defer r.mu.Unlock()

    // 假设这里从 Consul 获取服务地址
    addresses, err := queryConsul(consulAddress, r.target.Endpoint())
    if err != nil {
        panic(err)
    }

    // 更新 gRPC 服务器地址
    r.cc.UpdateState(resolver.State{Addresses: addresses})
    log.Println("解析器更新完成")
}

// 关闭解析器
func (r *myResolver) Close() {}

// 注册解析器
func init() {
    consulAddress, _ = GetLocalIP()
    consulAddress = fmt.Sprintf("%s:8500", consulAddress)
    resolver.Register(&myResolverBuilder{})
}

// 从 Consul 查询 `serviceName` 并返回可用的 gRPC 服务器列表
func queryConsul(consulAddress string, serviceName string) ([]resolver.Address, error) {
    // 创建 Consul 客户端
    consulConfig := api.DefaultConfig()
    consulConfig.Address = consulAddress
    client, err := api.NewClient(consulConfig)
    if err != nil {
        return nil, err
    }

    // 查询 Consul 中健康的 gRPC 服务
    services, _, err := client.Health().Service(serviceName, "", true, nil)
    if err != nil {
        return nil, err
    }

    // 转换为 gRPC `resolver.Address`
    var addresses []resolver.Address
    for _, service := range services {
        addr := service.Service.Address
        port := service.Service.Port
        addresses = append(addresses, resolver.Address{
            Addr:       addr + ":" + fmt.Sprintf("%d", port),
            ServerName: serviceName})
    }

    // 如果没有可用服务，返回错误
    if len(addresses) == 0 {
        return nil, fmt.Errorf("未找到任何健康的 %s 服务", serviceName)
    }

    for _, address := range addresses {
        fmt.Println(address)
    }

    return addresses, nil
}

// 获取本机 IPv4 地址（第一个非 127.0.0.1 的）
func GetLocalIP() (string, error) {
    interfaces, err := net.Interfaces()
    if err != nil {
        return "", err
    }

    for _, iface := range interfaces {
        // 忽略未启用的网卡
        if iface.Flags&net.FlagUp == 0 || iface.Flags&net.FlagLoopback != 0 {
            continue
        }

        addrs, err := iface.Addrs()
        if err != nil {
            continue
        }

        for _, addr := range addrs {
            ipNet, ok := addr.(*net.IPNet)
            if !ok || ipNet.IP.IsLoopback() || ipNet.IP.To4() == nil {
                continue
            }
            return ipNet.IP.String(), nil // 找到第一个 IPv4 地址直接返回
        }
    }

    return "", fmt.Errorf("no valid IPv4 address found")
}

func main() {
    // 使用自定义解析器
    conn, err := grpc.Dial(
        "consul:///mx_shop_users",
        grpc.WithInsecure(),
        grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"round_robin"}`),
    )
    if err != nil {
        log.Fatalf("连接 gRPC 服务器失败: %v", err)
    }
    defer conn.Close()

    log.Println("gRPC 连接成功")
    UserSrvClient := proto.NewUserClient(conn)
    rsp, err := UserSrvClient.GetUserList(context.Background(), &proto.PageInfo{
        Pn:    1,
        PSize: 2,
    })
    if err != nil {
        panic(err)
    }
    for index, data := range rsp.Data {
        fmt.Println(index, data)
    }
}

是有负载均衡的

9. 总结

✅ gRPC 预留的 resolver.Resolver 接口 主要用于 解析目标地址，并提供服务器列表。

✅ 核心接口

**resolver.Resolver**：解析地址并返回服务器列表
**resolver.Builder**：创建解析器
**resolver.ClientConn**：通知 gRPC 解析结果
**resolver.State**：存储解析后的服务器地址

✅ 自定义 gRPC 解析器适用场景

Consul、Etcd、Kubernetes 服务发现
动态解析 gRPC 服务器
自定义负载均衡策略

🚀 这样，你的 gRPC 应用就能动态解析服务，并自动负载均衡！态解析并连接多个后端服务器！**

grpc-resolver

1. resolver.Resolver 预留接口

🔹 Resolver 关键方法

2. resolver.Builder 接口（解析器构造器）

🔹 Builder 关键方法

3. resolver.ClientConn 预留接口

🔹 ClientConn 关键方法

4. 解析器的 resolver.State 结构