玖叶教程网

前端编程开发入门

2024年04月03日

grpc 链接池(4)自定义resolver 、balancer和picker

在分析完源码后golang源码分析：grpc 链接池（3）resolver 、balancer和picker，我们尝试自定义实现相应的插件。grpc 通过服务发现或者直连形式获取到 gRPC server 的实例的 endpoints，然后通知负载均衡器进行 SubConn 更新，对于新加入的 endpoint 进行实例创建，移出废弃的 endpoint， 最后通过状态更新将状态为 Idle 的 SubConn 进行管理，gRPC 在调用 Invoke时，则会通过负载均衡器中的 Picker 去按照某一个负载均衡算法选择一个 SubConn 创建链接，如果创建成功则不再进行其他 SubConn 的尝试，否则会按照一定的退避算法进行重试，直到退避失败或者创建链接成功为止。上述三个组件的功能分别如下：

  • resolver：通过直连、本地配置，或者从服务发现后台，比如k8s、nacos、etcd，consul等存储介质，获取target对应的endpoint列表。
  • balancer：管理连接池的SubConn，创建对应的picker
  • picker：从 SubConn 列表中按照负载均衡算法选择一个 SubConn 创建链接

下面我们通过这样一个实例来分别实现上述组件，并测试正确性。

后台服务有三个endpoint：127.0.0.1:9080，127.0.0.1:9081，127.0.0.1:9082客户端可以随机从三个endpoint中选择 一个来发送请求

package myresolver


import (
  "fmt"
  "strconv"


  // "github.com/golang-leetcode/grpc-go/resolver"
  //引用包不对，导致注册失败
  "google.golang.org/grpc/resolver"
)


const MockResolverScheme = "mock"


type mockResolverBuilder struct {
}


type mockResolver struct {
  target resolver.Target
  cc     resolver.ClientConn
}


func (*mockResolverBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
  r := &mockResolver{
    target: target,
    cc:     cc,
  }
  fmt.Println("target:", target, target.Endpoint, target.URL)
  switch target.Endpoint {
  case "myMock":
    fmt.Println("myMock resolver")
  }
  r.start()
  return r, nil
}


func (*mockResolverBuilder) Scheme() string {
  fmt.Println("get mock scheme")
  return MockResolverScheme
}


func (r mockResolver) start() {
  addrs := make([]resolver.Address, 0)
  for i := 0; i < 3; i++ {
    addrs = append(addrs, resolver.Address{Addr: "127.0.0.1:908" + strconv.FormatInt(int64(i), 10)})
  }
  fmt.Println("addrs", addrs)
  r.cc.UpdateState(resolver.State{Addresses: addrs})
}


func (r *mockResolver) ResolveNow(o resolver.ResolveNowOptions) {
  addrs := make([]resolver.Address, 0)
  for i := 0; i < 3; i++ {
    addrs = append(addrs, resolver.Address{
      ServerName: "resolver.mock.grpc.io" + strconv.FormatInt(int64(i), 10),
      Addr:       "127.0.0.1:908" + strconv.FormatInt(int64(i), 10)})
  }
  fmt.Println("addrs", addrs)
  //2023/02/05 16:55:53 could not greet rpc error: code = Unavailable desc = last connection error: connection error: desc = "transport: Error while dialing dial tcp: lookup tcp///default/resolver.mock.grpc.io: nodename nor servname provided, or not known"
  r.start()
}
func (*mockResolver) Close() {}


func init() {
  resolver.Register(&mockResolverBuilder{})
  fmt.Println(resolver.Get(MockResolverScheme))
}
package mybalancer


import (
  "learn/grpc/picker"


  "google.golang.org/grpc/balancer"
  "google.golang.org/grpc/balancer/base"
)


const Name = "random"


func init() {
  balancer.Register(newBuilder())
}


func newBuilder() balancer.Builder {
  return base.NewBalancerBuilder(Name, picker.NewRandomPickerBuilder(), base.Config{HealthCheck: true})
}
package picker


import (
  "fmt"
  "math/rand"
  "sync"
  "time"


  "google.golang.org/grpc/balancer"
  "google.golang.org/grpc/balancer/base"
)


func NewRandomPickerBuilder() *randomPickerBuilder {
  return &randomPickerBuilder{}
}


type randomPickerBuilder struct {
}


type Conn struct {
  SubConn     balancer.SubConn
  SubConnInfo base.SubConnInfo
}


func (r *randomPickerBuilder) Build(info base.PickerBuildInfo) balancer.Picker {
  if len(info.ReadySCs) == 0 {
    return base.NewErrPicker(balancer.ErrNoSubConnAvailable)
  }
  readyScs := make([]Conn, 0, len(info.ReadySCs))
  for sc, info := range info.ReadySCs {
    readyScs = append(readyScs, Conn{
      SubConn:     sc,
      SubConnInfo: info,
    })
  }
  return &randomPicker{
    subConns: readyScs,
    r:        rand.New(rand.NewSource(time.Now().UnixNano())),
  }
}


type randomPicker struct {
  subConns []Conn
  mu       sync.Mutex
  r        *rand.Rand
}


func (r *randomPicker) Pick(_ balancer.PickInfo) (balancer.PickResult, error) {
  next := r.r.Int() % len(r.subConns)
  sc := r.subConns[next]
  fmt.Printf("picked: %+v\n", sc.SubConnInfo.Address.Addr)
  return balancer.PickResult{
    SubConn: sc.SubConn,
  }, nil
}

这里实现的比较简单，比如我们可以通过target.Endpoint 来实现不同的endpoint获取策略。然后我们启动服务

package hello


import (
  context "context"


  codes "google.golang.org/grpc/codes"
  status "google.golang.org/grpc/status"
)


type ImplementedGreeterServer struct {
  Port string
}


func (i *ImplementedGreeterServer) SayHello(context.Context, *HelloRequest) (*HelloReply, error) {
  return nil, status.Errorf(codes.Unimplemented, "method SayHello  implemented"+i.Port)
}
func (i *ImplementedGreeterServer) SayHello1(context.Context, *HelloRequest) (*HelloReply, error) {
  return nil, status.Errorf(codes.Unimplemented, "method SayHello1  implemented"+i.Port)
}


func (i *ImplementedGreeterServer) mustEmbedUnimplementedGreeterServer() {}


package main


import (
  helloworld "learn/grpc/conn/hello"
  "log"
  "net"
  "strconv"


  "google.golang.org/grpc"
)


func main() {
  for i := 0; i < 3; i++ {
    go func(i int) {
      addr := ":908" + strconv.FormatInt(int64(i), 10)


      srv := grpc.NewServer()
      helloworld.RegisterGreeterServer(srv, &helloworld.ImplementedGreeterServer{
        Port: addr,
      })


      listener, err := net.Listen("tcp", addr)
      if err != nil {
        log.Fatalf("failed to listen: %v", err)
      }


      err = srv.Serve(listener)
      if err != nil {
        log.Fatalf("failed to serve: %v", err)
      }
    }(i)
  }
  select {}
}

然后通过客户端发起请求

package main


import (
  "context"
  "fmt"
  "log"
  "time"


  helloworld "learn/grpc/conn/hello"
  "learn/grpc/mybalancer"
  "learn/grpc/myresolver"


  "google.golang.org/grpc"
)


func main() {
  conn, err := grpc.Dial(myresolver.MockResolverScheme+":///myMock", grpc.WithInsecure(), grpc.WithTimeout(10*time.Second), grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, mybalancer.Name)), grpc.WithBlock()) // grpc.WithBalancerName(builder.Name) //, grpc.WithBlock(),grpc.WithResolvers(r),
  defer conn.Close()
  if err != nil {
    log.Fatalf("did not1 connect: %v", err)
  }
  client := helloworld.NewGreeterClient(conn)
  for i := 0; i < 10; i++ {
    resp, err := client.SayHello(context.Background(), &helloworld.HelloRequest{Name: "hello world!"})
    fmt.Println(i+1, "个", resp, err)
    resp, err = client.SayHello1(context.Background(), &helloworld.HelloRequest{Name: "hello world!"})
    fmt.Println(i+1, "个", resp, err)
    if err != nil {
      log.Println("could not greet", err)
    }
  }
}

这里需要注意的是，我们在创建连接的是通过schema来选择我们自定义的mock resolver，指定balancer的时候，使用的是 grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, mybalancer.Name))来指定的，对于picker来说，它是balancer里面直接引用的，所以不需要在连接的时候指定。

1 个 <nil> rpc error: code = Unimplemented desc = method SayHello  implemented:9082
picked: 127.0.0.1:9081
1 个 <nil> rpc error: code = Unimplemented desc = method SayHello1  implemented:9081
2023/02/05 18:26:08 could not greet rpc error: code = Unimplemented desc = method SayHello1  implemented:9081
picked: 127.0.0.1:9081
2 个 <nil> rpc error: code = Unimplemented desc = method SayHello  implemented:9081
picked: 127.0.0.1:9080
2 个 <nil> rpc error: code = Unimplemented desc = method SayHello1  implemented:9080
2023/02/05 18:26:08 could not greet rpc error: code = Unimplemented desc = method SayHello1  implemented:9080
picked: 127.0.0.1:9080
3 个 <nil> rpc error: code = Unimplemented desc = method SayHello  implemented:9080
picked: 127.0.0.1:9080
3 个 <nil> rpc error: code = Unimplemented desc = method SayHello1  implemented:9080
2023/02/05 18:26:08 could not greet rpc error: code = Unimplemented desc = method SayHello1  implemented:9080
picked: 127.0.0.1:9080
4 个 <nil> rpc error: code = Unimplemented desc = method SayHello  implemented:9080
picked: 127.0.0.1:9080
4 个 <nil> rpc error: code = Unimplemented desc = method SayHello1  implemented:9080
2023/02/05 18:26:08 could not greet rpc error: code = Unimplemented desc = method SayHello1  implemented:9080
picked: 127.0.0.1:9081
5 个 <nil> rpc error: code = Unimplemented desc = method SayHello  implemented:9081
picked: 127.0.0.1:9082
5 个 <nil> rpc error: code = Unimplemented desc = method SayHello1  implemented:9082
2023/02/05 18:26:08 could not greet rpc error: code = Unimplemented desc = method SayHello1  implemented:9082
picked: 127.0.0.1:9080
6 个 <nil> rpc error: code = Unimplemented desc = method SayHello  implemented:9080
picked: 127.0.0.1:9080
6 个 <nil> rpc error: code = Unimplemented desc = method SayHello1  implemented:9080
2023/02/05 18:26:08 could not greet rpc error: code = Unimplemented desc = method SayHello1  implemented:9080
picked: 127.0.0.1:9081
7 个 <nil> rpc error: code = Unimplemented desc = method SayHello  implemented:9081
picked: 127.0.0.1:9080
7 个 <nil> rpc error: code = Unimplemented desc = method SayHello1  implemented:9080
2023/02/05 18:26:08 could not greet rpc error: code = Unimplemented desc = method SayHello1  implemented:9080
picked: 127.0.0.1:9080
8 个 <nil> rpc error: code = Unimplemented desc = method SayHello  implemented:9080
picked: 127.0.0.1:9081
8 个 <nil> rpc error: code = Unimplemented desc = method SayHello1  implemented:9081
2023/02/05 18:26:08 could not greet rpc error: code = Unimplemented desc = method SayHello1  implemented:9081
picked: 127.0.0.1:9081
9 个 <nil> rpc error: code = Unimplemented desc = method SayHello  implemented:9081
picked: 127.0.0.1:9080
9 个 <nil> rpc error: code = Unimplemented desc = method SayHello1  implemented:9080
2023/02/05 18:26:08 could not greet rpc error: code = Unimplemented desc = method SayHello1  implemented:9080
picked: 127.0.0.1:9080
10 个 <nil> rpc error: code = Unimplemented desc = method SayHello  implemented:9080
picked: 127.0.0.1:9082
10 个 <nil> rpc error: code = Unimplemented desc = method SayHello1  implemented:9082
2023/02/05 18:26:08 could not greet rpc error: code = Unimplemented desc = method SayHello1  implemented:9082

看下结果，我们的连接端口是随机的，符合我们的预期。

标签：resolver 

作者:gojiuye , 分类:技术精选 , 浏览:29 , 评论:0

发表评论:

控制面板
您好，欢迎到访网站！
  查看权限
网站分类
最新留言