feat: add the logic of gRPC Load Balancing Strategy(LBS)

and this test case

Author: coderi421

README.md

@@ -38,7 +38,7 @@ Microservice framework implemented based on Golang.
 │  │  └─translator
 │  │      └─gin
 │  └─user 举例 user 服务
-│      ├─client
+│      ├─client 用于本地测试 user 服务的客户端 rpc 服务
 │      └─srv
 │          ├─config 服务的配置项，Log error 等子服务的相关逻辑全部注册到配置中
 │          ├─controller 表示层
@@ -67,12 +67,12 @@ Microservice framework implemented based on Golang.
 │  ├─registry
 │  │  └─consul 服务注册中心相关逻辑，参考 kratos
 │  └─server
-│      ├─restserver
-│      │  ├─middlewares
+│      ├─restserver http 服务的初始化配置
+│      │  ├─middlewares http 服务的中间件
 │      │  │  └─auth
-│      │  ├─pprof
-│      │  └─validation
-│      └─rpcserver
+│      │  ├─pprof http 服务的 pprof 相关逻辑
+│      │  └─validation http 服务的参数校验
+│      └─rpcserver rpc 服务的初始化配置
 │          ├─client.go rpc 客户端的初始化配置
 │          ├─server.go rpc 服务端的初始化配置
 │          ├─clientinterceptors 客户端的拦截器：超时连接器
@@ -81,13 +81,13 @@ Microservice framework implemented based on Golang.
 │          │  └─discovery 服务发现，负载均衡
 │          │      ├─builder.go 服务发现的构建器
 │          │      └─resolver.go 服务发现的解析器,负载均衡的逻辑在这里实现 UpdateState 核心
-│          ├─selector
-│          │  ├─node
-│          │  │  ├─direct
-│          │  │  └─ewma
-│          │  ├─p2c
-│          │  ├─random
-│          │  └─wrr
+│          ├─selector 重写 grpc 接口，具体服务相关的实现 
+│          │  ├─node gRPC 服务节点
+│          │  │  ├─direct 直连节点
+│          │  │  └─ewma ewma算法节点，用于实现 p2c 负载均衡策略
+│          │  ├─p2c  负载均衡 [Power of Two Random Choices] 算法
+│          │  ├─random 负载均衡随机算法
+│          │  └─wrr 负载均衡 加权轮询 [Weighted Round Robin] 算法
 │          └─serverinterceptors 服务端的拦截器：超时，crash恢复
 ├─pkg
 │  ├─app 包括的项目的启动服务，配置文件的读取，命令行工具，以及其他选项

app/pkg/options/registry.go

@@ -12,6 +12,7 @@ type RegistryOptions struct {
 	Scheme string `json:"scheme" mapstructure:"scheme,omitempty"`
 }
 
+// NewRegistryOptions 创建一个默认注册中心的配置
 func NewRegistryOptions() *RegistryOptions {
 	return &RegistryOptions{
 		Address: "127.0.0.1:8500",

app/pkg/options/server.go

@@ -1,35 +1,50 @@
 package options
 
 import (
-	"github.com/CoderI421/gframework/pkg/errors"
 	"github.com/spf13/pflag"
 )
 
 // ServerOptions service discovery center options console, etcd, nacos etc.
 type ServerOptions struct {
-	Name string `json:"name" mapstructure:"name"`
-	Host string `json:"host" mapstructure:"host"`
-	Port int    `json:"port" mapstructure:"port"`
+	Name              string   `json:"name,omitempty" mapstructure:"name"`                 // 服务名称
+	Host              string   `json:"host,omitempty" mapstructure:"host"`                 // 服务地址
+	Port              int      `json:"port,omitempty" mapstructure:"port"`                 // 服务端口
+	HttpPort          int      `json:"http_port,omitempty" mapstructure:"http_port"`       // http端口
+	EnableProfiling   bool     `json:"profiling,omitempty" mapstructure:"profiling"`       // 是否开启性能分析
+	EnableLimit       bool     `json:"limit,omitempty" mapstructure:"limit"`               // 是否开启限流
+	EnableMetrics     bool     `json:"metrics,omitempty" mapstructure:"metrics"`           // 是否开启指标
+	EnableHealthCheck bool     `json:"health_check,omitempty" mapstructure:"health_check"` // 是否开启健康检查
+	Middlewares       []string `json:"middlewares,omitempty" mapstructure:"middlewares"`   // 中间件
 }
 
 func NewServerOptions() *ServerOptions {
 	return &ServerOptions{
-		Host: "127.0.0.1",
-		Port: 8078,
-		Name: "user-srv",
+		Host:            "127.0.0.1",
+		Port:            8078,
+		HttpPort:        8079,
+		Name:            "user-srv",
+		EnableProfiling: true,
+		EnableLimit:     true,
+		EnableMetrics:   true,
 	}
 }
 
-func (o *ServerOptions) Validate() (errs []error) {
-	if o.Host == "" || o.Port == 0 {
-		errs = append(errs, errors.New("Host and Port must be"))
-	}
+func (so *ServerOptions) Validate() (errs []error) {
+	errs = []error{}
 	return
 }
 
-func (o *ServerOptions) AddFlags(fs *pflag.FlagSet) {
-	fs.StringVar(&o.Host, "server.host", o.Host, "server host default is 127.0.0.1")
-
-	fs.IntVar(&o.Port, "server.port", o.Port, "server port default is 8078")
-	fs.StringVar(&o.Name, "server.name", o.Name, "server name default is mxshop-user-srv")
+func (so *ServerOptions) AddFlags(fs *pflag.FlagSet) {
+	fs.BoolVar(&so.EnableProfiling, "server.enable-profiling", so.EnableProfiling,
+		"enable-profiling, if true, will add <host>:<port>/debug/pprof/, default is true")
+	fs.BoolVar(&so.EnableMetrics, "server.enable-metrics", so.EnableMetrics,
+		"enable-metrics, if true, will add /metrics, default is true")
+	fs.BoolVar(&so.EnableHealthCheck, "server.enable-health-check", so.EnableHealthCheck,
+		"enable-health-check, if true, will add health check route, default is true")
+	//fs.StringVarP 带有简写命令
+	// （接收值的变量，命令名称，默认值，描述）
+	fs.StringVar(&so.Host, "server.host", so.Host, "server host default is 127.0.0.1")
+	fs.IntVar(&so.Port, "server.port", so.Port, "server port default is 8078")
+	fs.IntVar(&so.HttpPort, "server.http-port", so.HttpPort, "server http port default is 8079")
+	fs.StringVar(&so.Name, "server.name", so.Name, "server name default is user-srv")
 }

app/user/client/client.go

@@ -5,14 +5,21 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/hashicorp/consul/api"
-
 	v1 "github.com/CoderI421/gframework/api/user/v1"
 	"github.com/CoderI421/gframework/gmicro/registry/consul"
 	"github.com/CoderI421/gframework/gmicro/server/rpcserver"
+	"github.com/CoderI421/gframework/gmicro/server/rpcserver/selector"
+	"github.com/CoderI421/gframework/gmicro/server/rpcserver/selector/random"
+	"github.com/hashicorp/consul/api"
 )
 
 func main() {
+	//客户端，设置全局负载均衡策略，这里选择了 random
+	// 这个逻辑中，balancerName是selector,在 gmicro/server/rpcserver/balancer.go中规定的
+	selector.SetGlobalSelector(random.NewBuilder())
+	// selector.SetGlobalSelector(random.NewBuilder()) 设定的全局的 selector 然后这里，调用 selector 进行注册
+	rpcserver.InitBuilder()
+
 	conf := api.DefaultConfig()
 	conf.Address = "127.0.0.1:8500"
 	conf.Scheme = "http"
@@ -27,6 +34,7 @@ func main() {
 
 	conn, err := rpcserver.DialInsecure(
 		context.Background(), rpcserver.WithDiscovery(r),
+		rpcserver.WithBalancerName("selector"),
 		/*
 			第3个/是为了第二个参数是空的
 			默认格式：direct://<authority>/127.0.0.1:8078
@@ -42,9 +50,13 @@ func main() {
 	}
 	defer conn.Close()
 	uc := v1.NewUserClient(conn)
-	re, err := uc.GetUserList(context.Background(), &v1.PageInfo{})
-	if err != nil {
-		panic(err)
+	// 在终端中，使用 --server.port=*** --server.http-port 命令启动启动多个，这里就会轮询调用，测试负载均衡
+	for {
+		re, err := uc.GetUserList(context.Background(), &v1.PageInfo{})
+		if err != nil {
+			panic(err)
+		}
+		fmt.Println(re)
+		time.Sleep(time.Second * 5)
 	}
-	fmt.Println(re)
 }

gmicro/app/app.go

@@ -58,7 +58,7 @@ func (a *App) Run() error {
 	a.mux.Unlock()
 
 	go func() {
-		// start the rpc server
+		// start the rpc server goroutine
 		if a.opts.rpcServer != nil {
 			err := a.opts.rpcServer.Start()
 			if err != nil {

gmicro/server/rpcserver/balancer.go

@@ -0,0 +1,101 @@
+package rpcserver
+
+import (
+	"google.golang.org/grpc/balancer"
+	"google.golang.org/grpc/balancer/base"
+	"google.golang.org/grpc/metadata"
+
+	"github.com/CoderI421/gframework/gmicro/registry"
+	"github.com/CoderI421/gframework/gmicro/server/rpcserver/selector"
+)
+
+const (
+	balancerName = "selector"
+)
+
+var (
+	_ base.PickerBuilder = &balancerBuilder{}
+	_ balancer.Picker    = &balancerPicker{}
+)
+
+// InitBuilder initializes the builder for the selector balancer.
+// 初始化 grpc balancer
+func InitBuilder() {
+	b := base.NewBalancerBuilder(
+		balancerName,
+		&balancerBuilder{
+			// 获取全局的 selector
+			builder: selector.GlobalSelector(),
+		},
+		base.Config{HealthCheck: true},
+	)
+	balancer.Register(b)
+}
+
+type balancerBuilder struct {
+	builder selector.Builder
+}
+
+// Build creates a grpc Picker.
+func (b *balancerBuilder) Build(info base.PickerBuildInfo) balancer.Picker {
+	if len(info.ReadySCs) == 0 {
+		// Block the RPC until a new picker is available via UpdateState().
+		return base.NewErrPicker(balancer.ErrNoSubConnAvailable)
+	}
+	nodes := make([]selector.Node, 0, len(info.ReadySCs))
+	for conn, info := range info.ReadySCs {
+		ins, _ := info.Address.Attributes.Value("rawServiceInstance").(*registry.ServiceInstance)
+		nodes = append(nodes, &grpcNode{
+			Node:    selector.NewNode("grpc", info.Address.Addr, ins),
+			subConn: conn,
+		})
+	}
+	p := &balancerPicker{
+		//自己传递的算法 注意先后顺序
+		selector: b.builder.Build(),
+	}
+	p.selector.Apply(nodes)
+	return p
+}
+
+// balancerPicker is a grpc picker.
+type balancerPicker struct {
+	selector selector.Selector
+}
+
+// Pick pick instances.
+func (p *balancerPicker) Pick(info balancer.PickInfo) (balancer.PickResult, error) {
+	n, done, err := p.selector.Select(info.Ctx)
+	if err != nil {
+		return balancer.PickResult{}, err
+	}
+
+	return balancer.PickResult{
+		SubConn: n.(*grpcNode).subConn,
+		Done: func(di balancer.DoneInfo) {
+			done(info.Ctx, selector.DoneInfo{
+				Err:           di.Err,
+				BytesSent:     di.BytesSent,
+				BytesReceived: di.BytesReceived,
+				ReplyMD:       Trailer(di.Trailer),
+			})
+		},
+	}, nil
+}
+
+// Trailer is a grpc trailder MD.
+type Trailer metadata.MD
+
+// Get get a grpc trailer value.
+func (t Trailer) Get(k string) string {
+	v := metadata.MD(t).Get(k)
+	if len(v) > 0 {
+		return v[0]
+	}
+	return ""
+}
+
+type grpcNode struct {
+	selector.Node
+	subConn balancer.SubConn
+}

gmicro/server/rpcserver/selector/balancer.go

@@ -0,0 +1,38 @@
+package selector
+
+import (
+	"context"
+	"time"
+)
+
+// Balancer is balancer interface
+type Balancer interface {
+	Pick(ctx context.Context, nodes []WeightedNode) (selected WeightedNode, done DoneFunc, err error)
+}
+
+// BalancerBuilder build balancer
+type BalancerBuilder interface {
+	Build() Balancer
+}
+
+// WeightedNode calculates scheduling weight in real time
+type WeightedNode interface {
+	Node
+
+	// Raw returns the original node
+	Raw() Node
+
+	// Weight is the runtime calculated weight
+	Weight() float64
+
+	// Pick the node
+	Pick() DoneFunc
+
+	// PickElapsed is time elapsed since the latest pick
+	PickElapsed() time.Duration
+}
+
+// WeightedNodeBuilder is WeightedNode Builder
+type WeightedNodeBuilder interface {
+	Build(Node) WeightedNode
+}

gmicro/server/rpcserver/selector/default_node.go

@@ -0,0 +1,66 @@
+package selector
+
+import (
+	"strconv"
+
+	"github.com/CoderI421/gframework/gmicro/registry"
+)
+
+// DefaultNode is selector node
+type DefaultNode struct {
+	scheme   string
+	addr     string
+	weight   *int64
+	version  string
+	name     string
+	metadata map[string]string
+}
+
+// Scheme is node scheme
+func (n *DefaultNode) Scheme() string {
+	return n.scheme
+}
+
+// Address is node address
+func (n *DefaultNode) Address() string {
+	return n.addr
+}
+
+// ServiceName is node serviceName
+func (n *DefaultNode) ServiceName() string {
+	return n.name
+}
+
+// InitialWeight is node initialWeight
+func (n *DefaultNode) InitialWeight() *int64 {
+	return n.weight
+}
+
+// Version is node version
+func (n *DefaultNode) Version() string {
+	return n.version
+}
+
+// Metadata is node metadata
+func (n *DefaultNode) Metadata() map[string]string {
+	return n.metadata
+}
+
+// NewNode new node
+func NewNode(scheme, addr string, ins *registry.ServiceInstance) Node {
+	n := &DefaultNode{
+		scheme: scheme,
+		addr:   addr,
+	}
+	if ins != nil {
+		n.name = ins.Name
+		n.version = ins.Version
+		n.metadata = ins.Metadata
+		if str, ok := ins.Metadata["weight"]; ok {
+			if weight, err := strconv.ParseInt(str, 10, 64); err == nil {
+				n.weight = &weight
+			}
+		}
+	}
+	return n
+}