gRPC阅读（4）——负载均衡
负载均衡算法 常见的负载均衡算法如下： RoundRobin（轮询） Weight-RoundRobin（加权轮询） 不同的后端服务器可能机器的配置和当前系统的负载并不相同，因此它们的抗压能力也不相同。给配置高、负载低的机器配置更高的权重，而配置低、负载高的机器，给其分配较低的权重，降低其系统负载，加权轮询能很好地处理这一问题，并将请求顺序且按照权重分配到后端。 Random（随机） Weight-Random（加权随机） 通过系统的随机算法，根据后端服务器的列表随机选取其中的一台服务器进行访问 源地址哈希法 源地址哈希的思想是根据获取客户端的 IP 地址，通过哈希函数计算得到的一个数值，用该数值对服务器列表的大小进行取模运算，得到的结果便是客服端要访问服务器的序号。采用源地址哈希法进行负载均衡，同一 IP 地址的客户端，当后端服务器列表不变时，它每次都会映射到同一台后端服务器进行访问 最小连接数法 最小连接数算法比较灵活和智能，由于后端服务器的配置不尽相同，对于请求的处理有快有慢，它是根据后端服务器当前的连接情况，动态地选取其中当前积压连接数最少的一台服务器来处理当前的请求，尽可能地提高后端服务的利用效率，将负责合理地分流到每一台服务器 Consistent-Hash（一致性哈希算法） 常见的是 Ketama 算法（虚拟节点），该算法是用来解决 cache 失效导致的缓存穿透的问题的，当然也可以适用于 gRPC 长连接的场景 自适应算法（P2C：多选二，二选一）：即从可用节点列表中随机选择两个节点，计算它们的负载率，选择负载率较低的进行请求 基于最小负载策略：该策略是 linkerd 的默认负载均衡器。当确定发送请求的位置时，linkerd 随机从负载均衡器池中选择两个副本，并选择两者中最少负载的副本。负载由每个副本的未完成请求数决定。该算法为单个副本提供了可管理的负载上限，与具有相似性能的其他算法相比，开销较少 峰值 EWMA（预测）策略：该算法是上述的变体，同样在发送请求时仍然在两个副本之间进行选择。不一样的是，该算法需要保持观察到的延迟的动态平均值，并且使用它来对每个副本的未完成请求的数量进行加权。这种方法对延迟波动更敏感，并通过向较慢的后端发送更少的请求来允许他们恢复时间（可以通过调参来改变对请求延时的敏感度） gRPC负载均衡 接着上篇，通过DNS进行服务发现获取服务器IP list之后，调用ccResolverWrapper.UpdateState更新状态： // UpdateState is called by resolver implementations to report new state to gRPC // which includes addresses and service config. func (ccr *ccResolverWrapper) UpdateState(s resolver.State) error { ... return ccr.cc.updateResolverStateAndUnlock(s, nil) } func (cc *ClientConn) updateResolverStateAndUnlock(s resolver.State, err error) error { ... var ret error // 应用最新的从dns发现的ServiceConfig if cc.dopts.disableServiceConfig { channelz.Infof(logger, cc.channelz, "ignoring service config from resolver (%v) and applying the default because service config is disabled", s.ServiceConfig) cc.maybeApplyDefaultServiceConfig() } else if s.ServiceConfig == nil { cc.maybeApplyDefaultServiceConfig() } else { if sc, ok := s.ServiceConfig.Config.(*ServiceConfig); s.ServiceConfig.Err == nil && ok { configSelector := iresolver.GetConfigSelector(s) if configSelector != nil { if len(s.ServiceConfig.Config.(*ServiceConfig).Methods) != 0 { channelz.Infof(logger, cc.channelz, "method configs in service config will be ignored due to presence of config selector") } } else { configSelector = &defaultConfigSelector{sc} } cc.applyServiceConfigAndBalancer(sc, configSelector) } else { ... } } // ServiceConfig的负载均衡配置 var balCfg serviceconfig.LoadBalancingConfig if cc.sc != nil && cc.sc.lbConfig != nil { balCfg = cc.sc.lbConfig } // 负载均衡器 bw := cc.balancerWrapper cc.mu.Unlock() // 应用服务发现结果 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. } return ret } // ccBalancerWrapper解耦ClientConn和Balancer，在updateClientConnState调用时才会创建Balancer // 并且ccBalancerWrapper使用gracefulswitch.Balancer支持Balancer的优雅切换 func (ccb *ccBalancerWrapper) updateClientConnState(ccs *balancer.ClientConnState) error { errCh := make(chan error) uccs := func(ctx context.Context) { defer close(errCh) if ctx.Err() != nil || ccb.balancer == nil { return } name := gracefulswitch.ChildName(ccs.BalancerConfig) if ccb.curBalancerName != name { ccb.curBalancerName = name channelz.Infof(logger, ccb.cc.channelz, "Channel switches to new LB policy %q", name) } // 通知gracefulBalancer状态更新 err := ccb.balancer.UpdateClientConnState(*ccs) if logger.V(2) && err != nil { logger.Infof("error from balancer.UpdateClientConnState: %v", err) } errCh <- err } onFailure := func() { close(errCh) } ccb.serializer.ScheduleOr(uccs, onFailure) return <-errCh } // gsb更新状态 func (gsb *Balancer) UpdateClientConnState(state balancer.ClientConnState) error { // 获取最新的balancer balToUpdate := gsb.latestBalancer() gsbCfg, ok := state.BalancerConfig.(*lbConfig) if ok { // Switch to the child in the config unless it is already active. if balToUpdate == nil || gsbCfg.childBuilder.Name() != balToUpdate.builder.Name() { var err error // 切换到新的balancer balToUpdate, err = gsb.switchTo(gsbCfg.childBuilder) if err != nil { return fmt.Errorf("could not switch to new child balancer: %w", err) } } // Unwrap the child balancer's config. state.BalancerConfig = gsbCfg.childConfig } if balToUpdate == nil { return errBalancerClosed } // 通知真正的Balancer状态更新 return balToUpdate.UpdateClientConnState(state) } // gsb优雅切换balancer func (gsb *Balancer) switchTo(builder balancer.Builder) (*balancerWrapper, error) { gsb.mu.Lock() if gsb.closed { gsb.mu.Unlock() return nil, errBalancerClosed } bw := &balancerWrapper{ builder: builder, gsb: gsb, lastState: balancer.State{ ConnectivityState: connectivity.Connecting, Picker: base.NewErrPicker(balancer.ErrNoSubConnAvailable), }, subconns: make(map[balancer.SubConn]bool), } balToClose := gsb.balancerPending // nil if there is no pending balancer if gsb.balancerCurrent == nil { gsb.balancerCurrent = bw } else { gsb.balancerPending = bw } gsb.mu.Unlock() // 关闭旧的balancer balToClose.Close() // 创建新的balancer newBalancer := builder.Build(bw, gsb.bOpts) if newBalancer == nil { gsb.mu.Lock() if gsb.balancerPending != nil { gsb.balancerPending = nil } else { gsb.balancerCurrent = nil } gsb.mu.Unlock() return nil, balancer.ErrBadResolverState } bw.Balancer = newBalancer return bw, nil } // pick_first balancer状态更新 func (b *pickfirstBalancer) UpdateClientConnState(state balancer.ClientConnState) error { ... cfg, ok := state.BalancerConfig.(pfConfig) ... // 展开endpoints/addrs var addrs []resolver.Address ... // balancer之前维护了子连接 // 如果连接的地址在addrs中，则保持连接 // 否则断开并用addrs重连 if b.subConn != nil { b.cc.UpdateAddresses(b.subConn, addrs) return nil } // balancer之前没有维护子连接 // 创建新的子连接 var subConn balancer.SubConn subConn, err := b.cc.NewSubConn(addrs, balancer.NewSubConnOptions{ StateListener: func(state balancer.SubConnState) { // 子连接状态更新，回调通知balancer b.updateSubConnState(subConn, state) }, }) ... b.subConn = subConn b.state = connectivity.Idle // 通知cc现在是Connecting状态 b.cc.UpdateState(balancer.State{ ConnectivityState: connectivity.Connecting, Picker: &picker{err: balancer.ErrNoSubConnAvailable}, }) b.subConn.Connect() return nil } 可以看到在pick_first策略下，会将所有地址都试一遍，连上其中一个后，后续picker返回的都是那个连接，相当于不做负载均衡： ...