go微服务框架kratos学习笔记七(kratos warden 负载均衡 balancer)

水深无声 2023-07-01 07:29 57阅读 0赞

# go微服务框架kratos学习笔记七(kratos warden 负载均衡 balancer) #

### 文章目录 ###

*  go微服务框架kratos学习笔记七(kratos warden 负载均衡 balancer)
 *   *  demo
     *   *  demo server
         *  demo client
         *  池
         *  dao
         *  service
         *  p2c
         *  roundrobin
     *  grpc官方负载均衡工作流程
     *  源码

--------------------

本节看看kratos的学习负载均衡策略的使用。

kratos 的负载均衡和服务发现一样也是基于grpc官方api实现的。

grpc官方的负载均衡自带了一个`round-robin`轮询策略、即像一个for循环一样挨个服的发请求、但这显然不能满足我们的需求、于是kratos自带了两种负载均衡策略：

**WRR (Weighted Round Robin)**  
该算法在加权轮询法基础上增加了动态调节权重值，用户可以在为每一个节点先配置一个初始的权重分，之后算法会根据节点cpu、延迟、服务端错误率、客户端错误率动态打分，在将打分乘用户自定义的初始权重分得到最后的权重值。

**P2C (Pick of two choices)**  
本算法通过随机选择两个node选择优胜者来避免羊群效应，并通过ewma尽量获取服务端的实时状态。  
服务端： 服务端获取最近500ms内的CPU使用率（需要将cgroup设置的限制考虑进去，并除于CPU核心数），并将CPU使用率乘与1000后塞入每次grpc请求中的的Trailer中夹带返回： cpu\_usage uint64 encoded with string cpu\_usage : 1000  
客户端： 主要参数：  
server\_cpu：通过每次请求中服务端塞在trailer中的cpu\_usage拿到服务端最近500ms内的cpu使用率  
inflight：当前客户端正在发送并等待response的请求数（pending request）  
latency: 加权移动平均算法计算出的接口延迟  
client\_success:加权移动平均算法计算出的请求成功率（只记录grpc内部错误，比如context deadline）  
目前客户端，已经默认使用p2c负载均衡算法

// NewClient returns a new blank Client instance with a default client interceptor.
    // opt can be used to add grpc dial options.
    func NewClient(conf *ClientConfig, opt ...grpc.DialOption) *Client {
        
    	c := new(Client)
    	if err := c.SetConfig(conf); err != nil {
        
    		panic(err)
    	}
    	c.UseOpt(grpc.WithBalancerName(p2c.Name))
    	c.UseOpt(opt...)
    	return c
    }

## demo ##

本节使用在笔记四[kratos warden-direct方式client调用][kratos warden-direct_client] 使用的direct服务发现方式、和相关代码。

**demo操作**  
1、分别在两个docker中启动一个grpc demo服务。  
2、启动一个client demo服务采用默认p2c负载均衡方式调用grpc SayHello()方法

### demo server ###

1、先启动demo服务 (其实就是一个kratos工具new出来的demo服务、代码可参考笔记四、或者在最后的github地址里面获取整个demo完整代码):  
![format_png][]

### demo client ###

package dao
    
    import (
    	"context"
    
    	"github.com/bilibili/kratos/pkg/net/rpc/warden"
    
    	"google.golang.org/grpc"
    
    	"fmt"
    	demoapi "call-server/api"
    	"google.golang.org/grpc/balancer/roundrobin"
    )
    
    // target server addrs.
    const target = "direct://default/10.0.75.2:30001,10.0.75.2:30002" // NOTE: example
    
    // NewClient new member grpc client
    func NewClient(cfg *warden.ClientConfig, opts ...grpc.DialOption) (demoapi.DemoClient, error) {
        
    	client := warden.NewClient(cfg, opts...)
    	conn, err := client.Dial(context.Background(), target)
    	if err != nil {
        
    		return nil, err
    	}
    	// 注意替换这里：
    	// NewDemoClient方法是在"api"目录下代码生成的
    	// 对应proto文件内自定义的service名字，请使用正确方法名替换
    	return demoapi.NewDemoClient(conn), nil
    }
    
    // NewClient new member grpc client
    func NewGrpcConn(cfg *warden.ClientConfig, opts ...grpc.DialOption) (*grpc.ClientConn, error) {
        
    	fmt.Println("-----tag: NewGrpcConn...")
    	//opts = append(opts, grpc.WithBalancerName(roundrobin.Name))
    	client := warden.NewClient(cfg, opts...)
    	
    	conn, err := client.Dial(context.Background(), target)
    	if err != nil {
        
    		return nil, err
    	}
    
    	return conn, nil
    }

target 填上两个服务ip

其中我多加了一个NewGrpcConn() 函数 、主要用来提取grpc连接。这里我用了kratos自带的pool类型来做连接池。

### 池 ###

关于这个池、它在 kratos `pkg/container/pool` 有两种实现方式 `Slice`和`List`方式。

package pool
    
    import (
    	"context"
    	"errors"
    	"io"
    	"time"
    
    	xtime "github.com/bilibili/kratos/pkg/time"
    )
    
    var (
    	// ErrPoolExhausted connections are exhausted.
    	ErrPoolExhausted = errors.New("container/pool exhausted")
    	// ErrPoolClosed connection pool is closed.
    	ErrPoolClosed = errors.New("container/pool closed")
    
    	// nowFunc returns the current time; it's overridden in tests.
    	nowFunc = time.Now
    )
    
    // Config is the pool configuration struct.
    type Config struct {
        
    	// Active number of items allocated by the pool at a given time.
    	// When zero, there is no limit on the number of items in the pool.
    	Active int
    	// Idle number of idle items in the pool.
    	Idle int
    	// Close items after remaining item for this duration. If the value
    	// is zero, then item items are not closed. Applications should set
    	// the timeout to a value less than the server's timeout.
    	IdleTimeout xtime.Duration
    	// If WaitTimeout is set and the pool is at the Active limit, then Get() waits WatiTimeout
    	// until a item to be returned to the pool before returning.
    	WaitTimeout xtime.Duration
    	// If WaitTimeout is not set, then Wait effects.
    	// if Wait is set true, then wait until ctx timeout, or default flase and return directly.
    	Wait bool
    }
    
    type item struct {
        
    	createdAt time.Time
    	c         io.Closer
    }
    
    func (i *item) expired(timeout time.Duration) bool {
        
    	if timeout <= 0 {
        
    		return false
    	}
    	return i.createdAt.Add(timeout).Before(nowFunc())
    }
    
    func (i *item) close() error {
        
    	return i.c.Close()
    }
    
    // Pool interface.
    type Pool interface {
        
    	Get(ctx context.Context) (io.Closer, error)
    	Put(ctx context.Context, c io.Closer, forceClose bool) error
    	Close() error
    }

### dao ###

dao中添加一个连接池。

package dao
    
    import (
    	"context"
    	"time"
    
    	demoapi "call-server/api"
    	"call-server/internal/model"
    
    	"github.com/bilibili/kratos/pkg/cache/memcache"
    	"github.com/bilibili/kratos/pkg/cache/redis"
    	"github.com/bilibili/kratos/pkg/conf/paladin"
    	"github.com/bilibili/kratos/pkg/database/sql"
    	"github.com/bilibili/kratos/pkg/net/rpc/warden"
    	"github.com/bilibili/kratos/pkg/sync/pipeline/fanout"
    	xtime "github.com/bilibili/kratos/pkg/time"
    	//grpcempty "github.com/golang/protobuf/ptypes/empty"
    	//"github.com/pkg/errors"
    
    	"github.com/google/wire"
    	"github.com/bilibili/kratos/pkg/container/pool"
    	"io"
    	"reflect"
    	"google.golang.org/grpc"
    
    )
    
    var Provider = wire.NewSet(New, NewDB, NewRedis, NewMC)
    
    //go:generate kratos tool genbts
    // Dao dao interface
    type Dao interface {
        
    	Close()
    	Ping(ctx context.Context) (err error)
    	// bts: -nullcache=&model.Article{ID:-1} -check_null_code=$!=nil&&$.ID==-1
    	Article(c context.Context, id int64) (*model.Article, error)
    	//SayHello(c context.Context, req *demoapi.HelloReq) (resp *grpcempty.Empty, err error)
    
    	//get an demo grpcConn/grpcClient/ from rpc pool
    	GrpcConnPut(ctx context.Context, cc *grpc.ClientConn) (err error)
    	GrpcConn(ctx context.Context) (gcc *grpc.ClientConn, err error)
    	GrpcClient(ctx context.Context) (cli demoapi.DemoClient, err error)
    }
    
    // dao dao.
    type dao struct {
        
    	db         *sql.DB
    	redis      *redis.Redis
    	mc         *memcache.Memcache
    	cache      *fanout.Fanout
    	demoExpire int32
    	rpcPool    pool.Pool 
    }
    
    // New new a dao and return.
    func New(r *redis.Redis, mc *memcache.Memcache, db *sql.DB) (d Dao, cf func(), err error) {
        
    	return newDao(r, mc, db)
    }
    
    func newDao(r *redis.Redis, mc *memcache.Memcache, db *sql.DB) (d *dao, cf func(), err error) {
        
    	var cfg struct {
        
    		DemoExpire xtime.Duration
    	}
    	if err = paladin.Get("application.toml").UnmarshalTOML(&cfg); err != nil {
        
    		return
    	}
    
    	// new pool
    	pool_config := &pool.Config{
        
    		Active:      0,
    		Idle:        0,
    		IdleTimeout: xtime.Duration(0 * time.Second),
    		WaitTimeout: xtime.Duration(30 * time.Millisecond),
    	}
    
    	rpcPool := pool.NewSlice(pool_config)
    	rpcPool.New = func(ctx context.Context) (cli io.Closer, err error) {
        
    		wcfg := &warden.ClientConfig{
        }
    		paladin.Get("grpc.toml").UnmarshalTOML(wcfg)
    		if cli, err = NewGrpcConn(wcfg); err != nil {
        
    			return
    		}
    
    		return
    	}
    
    	d = &dao{
        
    		db:         db,
    		redis:      r,
    		mc:         mc,
    		cache:      fanout.New("cache"),
    		demoExpire: int32(time.Duration(cfg.DemoExpire) / time.Second),
    		rpcPool:    rpcPool,
    	}
    	cf = d.Close
    	return
    }
    
    // Close close the resource.
    func (d *dao) Close() {
        
    	d.cache.Close()
    }
    
    // Ping ping the resource.
    func (d *dao) Ping(ctx context.Context) (err error) {
        
    	return nil
    }
    
    func (d *dao) GrpcClient(ctx context.Context) (cli demoapi.DemoClient, err error) {
        
    	var cc io.Closer
    	if cc, err = d.rpcPool.Get(ctx); err != nil {
        
    		return
    	}
    
    	cli = demoapi.NewDemoClient(reflect.ValueOf(cc).Interface().(*grpc.ClientConn))
    	return
    }
    
    func (d *dao) GrpcConnPut(ctx context.Context, cc *grpc.ClientConn) (err error) {
        
    	err = d.rpcPool.Put(ctx, cc, false)
    	return
    }
    
    func (d *dao) GrpcConn(ctx context.Context) (gcc *grpc.ClientConn, err error) {
        
    	var cc io.Closer
    	if cc, err = d.rpcPool.Get(ctx); err != nil {
        
    		return
    	}
    
    	gcc = reflect.ValueOf(cc).Interface().(*grpc.ClientConn)
    	return
    }

### service ###

// SayHello grpc demo func.
    func (s *Service) SayHello(ctx context.Context, req *pb.HelloReq) (reply *empty.Empty, err error) {
        
    	reply = new(empty.Empty)
    	var cc demoapi.DemoClient
    	var gcc *grpc.ClientConn
    	if gcc, err = s.dao.GrpcConn(ctx); err != nil {
        
    		return
    	}
    	defer s.dao.GrpcConnPut(ctx, gcc)
    	cc = demoapi.NewDemoClient(gcc)
    	//if cc, err = s.dao.GrpcClient(ctx); err != nil {
        
    	//	return
    	//}
    	cc.SayHello(ctx, req)
    	fmt.Printf("hello %s", req.Name)
    	return
    }

好了现在测试 、 布局如下 :

![format_png 1][]

### p2c ###

![aHR0cHM6Ly9maWxlcy5jbmJsb2dzLmNvbS9maWxlcy9haWx1bWl5YW5hL3AyYy5naWY][]

### roundrobin ###

轮询方式只需要在NewGrpcConn()里面加语一句配置项即可，它会覆盖掉p2c的配置项。

`opts = append(opts, grpc.WithBalancerName(roundrobin.Name))`

![aHR0cHM6Ly9maWxlcy5jbmJsb2dzLmNvbS9maWxlcy9haWx1bWl5YW5hL3JvdW5kcm9iaW4uZ2lm][]

## grpc官方负载均衡工作流程 ##

我们目前也只是使用了Api、最后来瞧瞧官方grpc的工作流程 :

gRPC开源组件官方并未直接提供服务注册与发现的功能实现，但其设计文档已提供实现的思路，并在不同语言的gRPC代码API中已提供了命名解析和负载均衡接口供扩展。

![format_png 2][]

1.  服务启动后，gPRC客户端通过resolve发起一个名称解析请求。名称会被解析为一个或更多的IP地址，每个地址指明它是一个服务器地址还是一个负载均衡器地址，并且包含一个Opt指明哪一个客户端的负载均衡策略应该被使用(例如: 轮询调度或grpclb)。
2.  客户端实现一个负载均衡策略。  
    注意：如果任何一个被解析器返回的地址是均衡器地址，那么这个客户端会使用grpclb策略，而不管请求的Opt配置的是哪种负载均衡策略。否则，客户端会使用一个Opt项配置负载均衡策略。如果没有负载均衡策略，那么客户端会使用默认的取第一个可用服务器地址的策略。
3.  负载均衡策略对每一个服务器地址创建一个子通道。
4.  当调用rpc请求时，负载均衡策略会决定应该发送到哪个子通道(例如: 哪个服务器)。  
    grpclb策略下，客户端按负载均衡器返回的顺序发送请求到服务器。如果服务器列表为空，调用将会阻塞直到收到一个非空的列表。

## 源码 ##

本节测试代码 : https://github.com/ailumiyana/kratos-note/tree/master/warden/balancer

[kratos warden-direct_client]: https://www.cnblogs.com/ailumiyana/p/12114703.html
[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9pbWcyMDE4LmNuYmxvZ3MuY29tL2Jsb2cvMTM4NDU1NS8yMDIwMDEvMTM4NDU1NS0yMDIwMDExOTE4NDQxMDYzNy05NjcwOTM0OTYucG5n?x-oss-process=image/format,png
[format_png 1]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9pbWcyMDE4LmNuYmxvZ3MuY29tL2Jsb2cvMTM4NDU1NS8yMDIwMDEvMTM4NDU1NS0yMDIwMDExOTE4NDMyNTc3MS01NTk1OTAzMzEucG5n?x-oss-process=image/format,png
[aHR0cHM6Ly9maWxlcy5jbmJsb2dzLmNvbS9maWxlcy9haWx1bWl5YW5hL3AyYy5naWY]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9maWxlcy5jbmJsb2dzLmNvbS9maWxlcy9haWx1bWl5YW5hL3AyYy5naWY
[aHR0cHM6Ly9maWxlcy5jbmJsb2dzLmNvbS9maWxlcy9haWx1bWl5YW5hL3JvdW5kcm9iaW4uZ2lm]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9maWxlcy5jbmJsb2dzLmNvbS9maWxlcy9haWx1bWl5YW5hL3JvdW5kcm9iaW4uZ2lm
[format_png 2]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9pbWcyMDE4LmNuYmxvZ3MuY29tL2Jsb2cvMTM4NDU1NS8yMDIwMDEvMTM4NDU1NS0yMDIwMDExOTE4NDM0MDE0Ny0yMjQzNzE4NjcucG5n?x-oss-process=image/format,png