上一篇文章一起学习了Resolver的原理和源码剖析,本篇持续和大家一起学习下和Resolver关系密切的Balancer的相干内容。这里说的负载平衡次要指数据中心内的负载平衡,即RPC间的负载平衡。
传送门 服务发现原理剖析与源码解读
基于go-zero v1.3.5 和 grpc-go v1.47.0
负载平衡
每一个被调用服务都会有多个实例,那么服务的调用方应该将申请,发向被调用服务的哪一个服务实例,这就是负载平衡的业务场景。
负载平衡的第一个关键点是公平性,即负载平衡须要关注被调用服务实例组之间的公平性,不要呈现旱的旱死,涝的涝死的状况。
负载平衡的第二个关键点是正确性,即对于有状态的服务来说,负载平衡须要关怀申请的状态,将申请调度到能解决它的后端实例上,不要呈现不能解决和错误处理的状况。
无状态的负载平衡
无状态的负载平衡是咱们日常工作中接触比拟多的负载平衡模型,它指的是参加负载平衡的后端实例是无状态的,所有的后端实例都是对等的,一个申请不管发向哪一个实例,都会失去雷同的并且正确的处理结果,所以无状态的负载平衡策略不须要关怀申请的状态。上面介绍两种无状态负载平衡算法。
轮询
轮询的负载平衡策略非常简单,只须要将申请按程序调配给多个实例,不必再做其余的解决。例如,轮询策略会将第一个申请调配给第一个实例,而后将下一个申请调配给第二个实例,这样顺次调配上来,调配完一轮之后,再回到结尾调配给第一个实例,再顺次调配。轮询在路由时,不利用申请的状态信息,属于无状态的负载平衡策略,所以它不能用于有状态实例的负载均衡器,否则正确性会呈现问题。在公平性方面,因为轮询策略只是按程序调配申请,所以实用于申请的工作负载和实例的解决能力差异都较小的状况。
权重轮询
权重轮询的负载平衡策略是将每一个后端实例调配一个权重,调配申请的数量和实例的权重成正比轮询。例如有两个实例 A,B,假如咱们设置 A 的权重为 20,B 的权重为 80,那么负载平衡会将 20% 的申请数量调配给 A,80 % 的申请数量调配给 B。权重轮询在路由时,不利用申请的状态信息,属于无状态的负载平衡策略,所以它也不能用于有状态实例的负载均衡器,否则正确性会呈现问题。在公平性方面,因为权重策略会按实例的权重比例来调配申请数,所以,咱们能够利用它解决实例的解决能力差异的问题,认为它的公平性比轮询策略要好。
有状态负载平衡
有状态负载平衡是指,在负载平衡策略中会保留服务端的一些状态,而后依据这些状态依照肯定的算法抉择出对应的实例。
P2C+EWMA
在go-zero中默认应用的是P2C的负载平衡算法。该算法的原理比较简单,即随机从所有可用节点中抉择两个节点,而后计算这两个节点的负载状况,抉择负载较低的一个节点来服务本次申请。为了防止某些节点始终得不到抉择导致不均衡,会在超过肯定的工夫后强制抉择一次。
在该简单平衡算法中,多出采纳了EWMA指数挪动加权均匀的算法,示意是一段时间内的均值。该算法绝对于算数均匀来说对于忽然的网络抖动没有那么敏感,忽然的抖动不会体现在申请的lag中,从而能够让算法更加平衡。
go-zero/zrpc/internal/balancer/p2c/p2c.go:133
atomic.StoreUint64(&c.lag, uint64(float64(olag)*w+float64(lag)*(1-w)))
go-zero/zrpc/internal/balancer/p2c/p2c.go:139
atomic.StoreUint64(&c.success, uint64(float64(osucc)*w+float64(success)*(1-w)))
系数w是一个工夫衰减值,即两次申请的距离越大,则系数w就越小。
go-zero/zrpc/internal/balancer/p2c/p2c.go:124
w := math.Exp(float64(-td) / float64(decayTime))
节点的load值是通过该连贯的申请提早 lag 和以后申请数 inflight 的乘积所得,如果申请的提早越大或者以后正在解决的申请数越多表明该节点的负载越高。
go-zero/zrpc/internal/balancer/p2c/p2c.go:199
func (c *subConn) load() int64 {
// plus one to avoid multiply zero
lag := int64(math.Sqrt(float64(atomic.LoadUint64(&c.lag) + 1)))
load := lag * (atomic.LoadInt64(&c.inflight) + 1)
if load == 0 {
return penalty
}
return load
}
源码剖析
如下源码会波及go-zero和gRPC,请依据给出的代码门路进行辨别
在gRPC中,Balancer和Resolver一样也能够自定义,同样也是通过Register办法进行注册
grpc-go/balancer/balancer.go:53
func Register(b Builder) {
m[strings.ToLower(b.Name())] = b
}
Register的参数Builder为接口,在Builder接口中,Build办法的第一个参数ClientConn也为接口,Build办法的返回值Balancer同样也是接口,定义如下:
能够看出,要想实现自定义的Balancer的话,就必须要实现balancer.Builder接口。
在理解了gRPC提供的Balancer的注册形式之后,咱们看一下go-zero是在什么中央进行Balancer注册的
go-zero/zrpc/internal/balancer/p2c/p2c.go:36
func init() {
balancer.Register(newBuilder())
}
在go-zero中并没有实现 balancer.Builder 接口,而是应用gRPC提供的 base.baseBuilder 进行注册,base.baseBuilder 实现了balancer.Builder 接口。创立baseBuilder的时候调用了 base.NewBalancerBuilder 办法,须要传入 PickerBuilder 参数,PickerBuilder为接口,在go-zero中 p2c.p2cPickerBuilder 实现了该接口。
PickerBuilder接口Build办法返回值 balancer.Picker 也是一个接口,p2c.p2cPicker 实现了该接口。
grpc-go/balancer/base/base.go:65
func NewBalancerBuilder(name string, pb PickerBuilder, config Config) balancer.Builder {
return &baseBuilder{
name: name,
pickerBuilder: pb,
config: config,
}
}
各构造之间的关系如下图所示,其中各构造模块对应的包为:
- balancer:grpc-go/balancer
- base:grpc-go/balancer/base
- p2c: go-zero/zrpc/internal/balancer/p2c
在哪里获取已注册的Balancer?
通过下面的流程步骤,曾经晓得了如何自定义Balancer,以及如何注册自定义的Blancer。既然注册了必定就会获取,接下来看一下是在哪里获取曾经注册的Balancer的。
咱们晓得Resolver是通过解析DialContext的第二个参数target,从而失去Resolver的name,而后依据name获取到对应的Resolver的。获取Balancer同样也是依据名称,Balancer的名称是在创立gRPC Client的时候通过配置项传入的,这里的p2c.Name为注册Balancer时指定的名称 p2c_ewma ,如下:
go-zero/zrpc/internal/client.go:50
func NewClient(target string, opts ...ClientOption) (Client, error) {
var cli client
svcCfg := fmt.Sprintf(`{"loadBalancingPolicy":"%s"}`, p2c.Name)
balancerOpt := WithDialOption(grpc.WithDefaultServiceConfig(svcCfg))
opts = append([]ClientOption{balancerOpt}, opts...)
if err := cli.dial(target, opts...); err != nil {
return nil, err
}
return &cli, nil
}
在上一篇文章中,咱们曾经晓得当创立gRPC客户端的时候,会触发调用自定义Resolver的Build办法,在Build办法外部获取到服务地址列表后,通过cc.UpdateState办法进行状态更新,前面当监听到服务状态变动的时候同样也会调用cc.UpdateState进行状态的更新,而这里的cc指的就是 ccResolverWrapper 对象,这一部分如果遗记的话,能够再去回顾一下解说Resolver的那篇文章,以便能丝滑接入本篇:
go-zero/zrpc/resolver/internal/kubebuilder.go:51
if err := cc.UpdateState(resolver.State{
Addresses: addrs,
}); err != nil {
logx.Error(err)
}
这里有几个重要的模块对象,如下:
- ClientConn:grpc-go/clientconn.go:464
- ccResolverWrapper:grpc-go/resolver_conn_wrapper.go:36
- ccBalancerWrapper:grpc-go/balancer_conn_wrappers.go:48
- Balancer:grpc-go/internal/balancer/gracefulswitch/gracefulswitch.go:46
- balancerWrapper:grpc-go/internal/balancer/gracefulswitch/gracefulswitch.go:247
当监听到服务状态的变更后(首次启动或者通过Watch监听变动)调用 ccResolverWrapper.UpdateState 触发更新状态的流程,各模块间的调用链路如下所示:
获取Balancer的动作是在 ccBalancerWrapper.handleSwitchTo 办法中触发的,代码如下所示:
grpc-go/balancer_conn_wrappers.go:266
builder := balancer.Get(name)
if builder == nil {
channelz.Warningf(logger, ccb.cc.channelzID, "Channel switches to new LB policy %q, since the specified LB policy %q was not registered", PickFirstBalancerName, name)
builder = newPickfirstBuilder()
} else {
channelz.Infof(logger, ccb.cc.channelzID, "Channel switches to new LB policy %q", name)
}
if err := ccb.balancer.SwitchTo(builder); err != nil {
channelz.Errorf(logger, ccb.cc.channelzID, "Channel failed to build new LB policy %q: %v", name, err)
return
}
ccb.curBalancerName = builder.Name()
而后在 Balancer.SwitchTo 办法中,调用了自定义Balancer的Build办法:
grpc-go/internal/balancer/gracefulswitch/gracefulswitch.go:121
newBalancer := builder.Build(bw, gsb.bOpts)
上文有提到Build办法的第一个参数为接口 balancer.ClientConn ,而这里传入的为 balancerWrapper ,所以gracefulswitch.balancerWrapper实现了该接口:
到这里咱们曾经晓得了获取自定义Balancer是在哪里触达的,以及在哪里获取的自定义的Balancer,和balancer.Builder的Build办法在哪里被调用。
通过上文可知这里的balancer.Builder为baseBuilder,所以调用的Build办法为baseBuilder的Build办法,Build办法的定义如下:
grpc-go/balancer/base/balancer.go:39
func (bb *baseBuilder) Build(cc balancer.ClientConn, opt balancer.BuildOptions) balancer.Balancer {
bal := &baseBalancer{
cc: cc,
pickerBuilder: bb.pickerBuilder,
subConns: resolver.NewAddressMap(),
scStates: make(map[balancer.SubConn]connectivity.State),
csEvltr: &balancer.ConnectivityStateEvaluator{},
config: bb.config,
}
bal.picker = NewErrPicker(balancer.ErrNoSubConnAvailable)
return bal
}
Build办法返回了baseBalancer,能够晓得baseBalancer实现了balancer.Balancer接口:
再来回顾下这个流程,其实次要做了如下几件事:
- 在自定义的Resolver中监听服务状态的变更
- 通过UpdateState来更新状态
- 获取自定义的Balancer
- 执行自定义Balancer的Build办法获取Balancer
如何创立连贯?
持续回到ClientConn的updateResolverState办法,在办法的最初调用balancerWrapper.updateClientConnState办法更新客户端的连贯状态:
grpc-go/clientconn.go:664
uccsErr := bw.updateClientConnState(&balancer.ClientConnState{ResolverState: s, BalancerConfig: balCfg})
if ret == nil {
ret = uccsErr // prefer ErrBadResolver state since any other error is
// currently meaningless to the caller.
}
前面的调用链路如下图所示:
最终会调用baseBalancer.UpdateClientConnState办法:
grpc-go/balancer/base/balancer.go:94
func (b *baseBalancer) UpdateClientConnState(s balancer.ClientConnState) error {
// .............
b.resolverErr = nil
addrsSet := resolver.NewAddressMap()
for _, a := range s.ResolverState.Addresses {
addrsSet.Set(a, nil)
if _, ok := b.subConns.Get(a); !ok {
sc, err := b.cc.NewSubConn([]resolver.Address{a}, balancer.NewSubConnOptions{HealthCheckEnabled: b.config.HealthCheck})
if err != nil {
logger.Warningf("base.baseBalancer: failed to create new SubConn: %v", err)
continue
}
b.subConns.Set(a, sc)
b.scStates[sc] = connectivity.Idle
b.csEvltr.RecordTransition(connectivity.Shutdown, connectivity.Idle)
sc.Connect()
}
}
for _, a := range b.subConns.Keys() {
sci, _ := b.subConns.Get(a)
sc := sci.(balancer.SubConn)
if _, ok := addrsSet.Get(a); !ok {
b.cc.RemoveSubConn(sc)
b.subConns.Delete(a)
}
}
// ................
}
当第一次触发调用UpdateClientConnState的时候,如下代码中 ok 为 false:
_, ok := b.subConns.Get(a);
所以会创立新的连贯:
sc, err := b.cc.NewSubConn([]resolver.Address{a}, balancer.NewSubConnOptions{HealthCheckEnabled: b.config.HealthCheck})
这里的 b.cc 即为 balancerWrapper,遗记的盆友能够往上翻看温习一下,也就是会调用 balancerWrapper.NewSubConn创立连贯
grpc-go/internal/balancer/gracefulswitch/gracefulswitch.go:328
func (bw *balancerWrapper) NewSubConn(addrs []resolver.Address, opts balancer.NewSubConnOptions) (balancer.SubConn, error) {
// .............
sc, err := bw.gsb.cc.NewSubConn(addrs, opts)
if err != nil {
return nil, err
}
// .............
bw.subconns[sc] = true
// .............
}
bw.gsb.cc即为ccBalancerWrapper,所以这里会调用ccBalancerWrapper.NewSubConn创立连贯:
grpc-go/balancer_conn_wrappers.go:299
func (ccb *ccBalancerWrapper) NewSubConn(addrs []resolver.Address, opts balancer.NewSubConnOptions) (balancer.SubConn, error) {
if len(addrs) <= 0 {
return nil, fmt.Errorf("grpc: cannot create SubConn with empty address list")
}
ac, err := ccb.cc.newAddrConn(addrs, opts)
if err != nil {
channelz.Warningf(logger, ccb.cc.channelzID, "acBalancerWrapper: NewSubConn: failed to newAddrConn: %v", err)
return nil, err
}
acbw := &acBalancerWrapper{ac: ac}
acbw.ac.mu.Lock()
ac.acbw = acbw
acbw.ac.mu.Unlock()
return acbw, nil
}
最终返回的是acBalancerWrapper对象,acBalancerWrapper实现了balancer.SubConn接口:
调用流程图如下所示:
创立连贯的默认状态为 connectivity.Idle :
grpc-go/clientconn.go:699
func (cc *ClientConn) newAddrConn(addrs []resolver.Address, opts balancer.NewSubConnOptions) (*addrConn, error) {
ac := &addrConn{
state: connectivity.Idle,
cc: cc,
addrs: addrs,
scopts: opts,
dopts: cc.dopts,
czData: new(channelzData),
resetBackoff: make(chan struct{}),
}
// ...........
}
在gRPC中为连贯定义了五种状态,别离如下:
const (
// Idle indicates the ClientConn is idle.
Idle State = iota
// Connecting indicates the ClientConn is connecting.
Connecting
// Ready indicates the ClientConn is ready for work.
Ready
// TransientFailure indicates the ClientConn has seen a failure but expects to recover.
TransientFailure
// Shutdown indicates the ClientConn has started shutting down.
Shutdown
)
在 baseBalancer 中通过b.scStates保留创立的连贯,初始状态也为connectivity.Idle,之后通过sc.Connect()进行连贯:
grpc-go/balancer/base/balancer.go:112
b.subConns.Set(a, sc)
b.scStates[sc] = connectivity.Idle
b.csEvltr.RecordTransition(connectivity.Shutdown, connectivity.Idle)
sc.Connect()
这里sc.Connetc调用的是acBalancerWrapper的Connect办法,能够看到这里创立连贯是异步进行的:
grpc-go/balancer_conn_wrappers.go:406
func (acbw *acBalancerWrapper) Connect() {
acbw.mu.Lock()
defer acbw.mu.Unlock()
go acbw.ac.connect()
}
最初会调用addrConn.connect办法:
grpc-go/clientconn.go:786
func (ac *addrConn) connect() error {
ac.mu.Lock()
if ac.state == connectivity.Shutdown {
ac.mu.Unlock()
return errConnClosing
}
if ac.state != connectivity.Idle {
ac.mu.Unlock()
return nil
}
ac.updateConnectivityState(connectivity.Connecting, nil)
ac.mu.Unlock()
ac.resetTransport()
return nil
}
从connect开始的调用链路如下所示:
在baseBalancer的UpdateSubConnState办法的最初,更新了Picker为自定义的Picker:
grpc-go/balancer/base/balancer.go:221
b.cc.UpdateState(balancer.State{ConnectivityState: b.state, Picker: b.picker})
在addrConn办法的最初会调用ac.resetTransport()真正的进行连贯的创立:
当连贯曾经创立好,处于Ready状态,最初调用baseBalancer.UpdateSubConnState办法,此时s==connectivity.Ready为true,而oldS == connectivity.Ready为false,所以会调用b.regeneratePicker()办法:
if (s == connectivity.Ready) != (oldS == connectivity.Ready) ||
b.state == connectivity.TransientFailure {
b.regeneratePicker()
}
func (b *baseBalancer) regeneratePicker() {
if b.state == connectivity.TransientFailure {
b.picker = NewErrPicker(b.mergeErrors())
return
}
readySCs := make(map[balancer.SubConn]SubConnInfo)
// Filter out all ready SCs from full subConn map.
for _, addr := range b.subConns.Keys() {
sci, _ := b.subConns.Get(addr)
sc := sci.(balancer.SubConn)
if st, ok := b.scStates[sc]; ok && st == connectivity.Ready {
readySCs[sc] = SubConnInfo{Address: addr}
}
}
b.picker = b.pickerBuilder.Build(PickerBuildInfo{ReadySCs: readySCs})
}
在regeneratePicker中获取了处于connectivity.Ready状态可用的连贯,同时更新了picker。还记得b.pickerBuilder吗?b.b.pickerBuilder为在go-zero中自定义实现的base.PickerBuilder接口。
go-zero/zrpc/internal/balancer/p2c/p2c.go:42
func (b *p2cPickerBuilder) Build(info base.PickerBuildInfo) balancer.Picker {
readySCs := info.ReadySCs
if len(readySCs) == 0 {
return base.NewErrPicker(balancer.ErrNoSubConnAvailable)
}
var conns []*subConn
for conn, connInfo := range readySCs {
conns = append(conns, &subConn{
addr: connInfo.Address,
conn: conn,
success: initSuccess,
})
}
return &p2cPicker{
conns: conns,
r: rand.New(rand.NewSource(time.Now().UnixNano())),
stamp: syncx.NewAtomicDuration(),
}
}
最初把自定义的Picker赋值为 ClientConn.blockingpicker.picker属性。
grpc-go/balancer_conn_wrappers.go:347
func (ccb *ccBalancerWrapper) UpdateState(s balancer.State) {
ccb.cc.blockingpicker.updatePicker(s.Picker)
ccb.cc.csMgr.updateState(s.ConnectivityState)
}
如何抉择已创立的连贯?
当初曾经晓得了如何创立连贯,以及连贯其实是在 baseBalancer.scStates 中治理,当连贯的状态发生变化,则会更新 baseBalancer.scStates 。那么接下来咱们来看一下gRPC是如何抉择一个连贯进行申请的发送的。
当gRPC客户端发动调用的时候,会调用ClientConn的Invoke办法,个别不会被动应用该办法进行调用,该办法的调用个别是主动生成:
grpc-go/examples/helloworld/helloworld/helloworld_grpc.pb.go:39
func (c *greeterClient) SayHello(ctx context.Context, in *HelloRequest, opts ...grpc.CallOption) (*HelloReply, error) {
out := new(HelloReply)
err := c.cc.Invoke(ctx, "/helloworld.Greeter/SayHello", in, out, opts...)
if err != nil {
return nil, err
}
return out, nil
}
如下为发动申请的调用链路,最终会调用p2cPicker.Pick办法获取连贯,咱们自定义的负载平衡算法个别都在Pick办法中实现,获取到连贯之后,通过sendMsg发送申请。
grpc-go/stream.go:945
func (a *csAttempt) sendMsg(m interface{}, hdr, payld, data []byte) error {
cs := a.cs
if a.trInfo != nil {
a.mu.Lock()
if a.trInfo.tr != nil {
a.trInfo.tr.LazyLog(&payload{sent: true, msg: m}, true)
}
a.mu.Unlock()
}
if err := a.t.Write(a.s, hdr, payld, &transport.Options{Last: !cs.desc.ClientStreams}); err != nil {
if !cs.desc.ClientStreams {
return nil
}
return io.EOF
}
if a.statsHandler != nil {
a.statsHandler.HandleRPC(a.ctx, outPayload(true, m, data, payld, time.Now()))
}
if channelz.IsOn() {
a.t.IncrMsgSent()
}
return nil
}
源码剖析到此就完结了,因为篇幅无限没法做到八面玲珑,所以本文只列出了源码中的次要门路。
结束语
Balancer相干的源码还是有点简单的,笔者也是读了好几遍才理清脉络,所以如果读了一两遍感觉没有脉络也不必焦急,对照文章的脉络多读几遍就肯定能搞懂。
如果有疑难能够随时找我探讨,在社区群中能够搜寻dawn_zhou找到我。
心愿本篇文章对你有所帮忙,你的点赞是作者继续输入的最大能源。
我的项目地址
https://github.com/zeromicro/go-zero
欢送应用 go-zero
并 star 反对咱们!
微信交换群
关注『微服务实际』公众号并点击 交换群 获取社区群二维码。