负载均衡算法 常见的负载均衡算法如下： 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返回的都是那个连接，相当于不做负载均衡： ...

服务发现概述 平时用浏览器上过网都知道，输入一个网址比如google.com就能访问内容，背后是DNS帮我们将google.com解析成IP地址，最终浏览器才能基于TCP协议，从本地连接到这个服务提供商的IP地址。所以DNS属于服务发现的其中一种方式。 所以服务发现提供的就是通过自动化的方式帮助服务在网络中找到彼此，无需手动配置。 一个好的服务发现需要： 服务地址动态变化：服务的 IP 或端口可能因为容器化或自动扩展而频繁改变。 高可用：需要在服务实例宕机时快速感知并移除不健康的实例。 负载均衡：服务发现需要为调用方提供负载均衡能力，选择最佳的服务实例。 服务发现通常与负载均衡同时实现，分为两种方式： 客户端服务发现（如eureka、consul）：在客户端做负载均衡，选择一个实例进行调用，优点是避免集中式LB可能存在的瓶颈，性能较好，但是每个客户端需要维护服务端列表，服务端这部分的负载可能变高。并且更新LB或其他相关组件的策略时需要所有客户端都一起更新，管理不方便。并且需要多语言支持 代理服务发现（如k8s+coreDNS、nginx+consul）：客户端将请求发送到负载均衡器（如 API 网关），由负载均衡器查询服务注册中心并将请求转发给目标服务实例。 独立LB进程：LB与消费者在同一个主机中，但分别作为不同的进程，避免了需要多语言支持，以及LB的更新不需要调用方改代码。 服务发现的核心组件有：注册中心、服务提供者、客户端（服务消费者） 服务发现的关键功能有：服务注册、服务查询、健康检查、动态更新 gRPC服务发现 gRPC使用客户端服务发现，gRPC中称为名称解析（Name Resolution），默认情况下使用DNS-resolver。通过服务发现解析出IP列表后就通过LB组件进行负载均衡并建立连接。 下面基于target=localhost:50052这个服务端地址来进行分析，并且是默认的DNS作为resolver（不用官方例子的50051端口是因为被mac的launchd进程占用了）。 首先gRPC在创建cc（ClientConn）的时候，使用initParsedTargetAndResolverBuilder创建resolver.Builder。这一步决定的是采用什么服务发现机制，默认是DNS。 func (cc *ClientConn) initParsedTargetAndResolverBuilder() error { logger.Infof("original dial target is: %q", cc.target) // 尝试直接解析target并获取相应的resolver.Builder var rb resolver.Builder parsedTarget, err := parseTarget(cc.target) if err == nil { rb = cc.getResolver(parsedTarget.URL.Scheme) if rb != nil { cc.parsedTarget = parsedTarget cc.resolverBuilder = rb return nil } } // target没有指定schema（比如我们的localhost:50052是没有指定schema的）或者无法匹配schema对应的resolver.Builder // 那么使用默认的schema，即dns defScheme := cc.dopts.defaultScheme if internal.UserSetDefaultScheme { defScheme = resolver.GetDefaultScheme() } // 此处canonicalTarget为dns:///localhost:50052 // "//"与第三个"/"之间的是authority canonicalTarget := defScheme + ":///" + cc.target // 再次尝试target并获取相应的resolver.Builder，此处会拿到dns.dnsBuilder parsedTarget, err = parseTarget(canonicalTarget) if err != nil { return err } rb = cc.getResolver(parsedTarget.URL.Scheme) if rb == nil { return fmt.Errorf("could not get resolver for default scheme: %q", parsedTarget.URL.Scheme) } // 保存parsedTarget和resolverBuilder cc.parsedTarget = parsedTarget cc.resolverBuilder = rb return nil } 那么resolverBuilder在什么时候会Build一个resolver出来呢？在ide的帮助下，可以直接定位到这个函数中： ...

启动客户端 客户端的启动也是三部曲： 初始化grpc.ClientConn 创建service对应的Client（比如codegen生成的GreeterClient） 发起rpc调用 第二步比较简单，只是把ClientConn作为GreeterClient的成员变量，重点分析建立连接和RPC调用 初始化ClientConn 初始化ClientConn做了很多准备工作，包括但不限于： 应用选项（DialOption） 构建拦截器调用链（Interceptor） 决定使用什么resolver（resolver.Builder） 检查传输层凭证，比如TLS（TransportCredentials） 解析自定义服务端配置（ServerConfig） … 但还有一些配置是在真正发起RPC调用的时候才会被设置和触发，比如重试限流器、RPC配置选择器、RPC负载均衡器等。 func NewClient(target string, opts ...DialOption) (conn *ClientConn, err error) { cc := &ClientConn{ target: target, conns: make(map[*addrConn]struct{}), dopts: defaultDialOptions(), } // 重试限流器 cc.retryThrottler.Store((*retryThrottler)(nil)) // 配置选择器，动态选择每个RPC的调用配置 cc.safeConfigSelector.UpdateConfigSelector(&defaultConfigSelector{nil}) cc.ctx, cc.cancel = context.WithCancel(context.Background()) // options ... // 确定使用哪个resolver（默认为dns） if err := cc.initParsedTargetAndResolverBuilder(); err != nil { return nil, err } // 内部使用的全局perTarget options for _, opt := range globalPerTargetDialOptions { opt.DialOptionForTarget(cc.parsedTarget.URL).apply(&cc.dopts) } // 初始化拦截器调用链 chainUnaryClientInterceptors(cc) chainStreamClientInterceptors(cc) // 验证安全传输，如TLS if err := cc.validateTransportCredentials(); err != nil { return nil, err } // 解析以json格式指定的配置 // 如负载均衡配置、per-RPC方法超时等 if cc.dopts.defaultServiceConfigRawJSON != nil { scpr := parseServiceConfig(*cc.dopts.defaultServiceConfigRawJSON, cc.dopts.maxCallAttempts) if scpr.Err != nil { return nil, fmt.Errorf("%s: %v", invalidDefaultServiceConfigErrPrefix, scpr.Err) } cc.dopts.defaultServiceConfig, _ = scpr.Config.(*ServiceConfig) } // keepalive对服务端探活 cc.mkp = cc.dopts.copts.KeepaliveParams // 获取authority，作为请求头中的:authority字段 if err = cc.initAuthority(); err != nil { return nil, err } // 注册channelz，用于监测grpc的运行 // 可通过http协议访问/grpc/channelz/v1查看grpc的状态 cc.channelzRegistration(target) channelz.Infof(logger, cc.channelz, "parsed dial target is: %#v", cc.parsedTarget) channelz.Infof(logger, cc.channelz, "Channel authority set to %q", cc.authority) // 连接状态管理器 cc.csMgr = newConnectivityStateManager(cc.ctx, cc.channelz) // 负载均衡器，动态选择每个RPC的子通道 cc.pickerWrapper = newPickerWrapper(cc.dopts.copts.StatsHandlers) // stats cc.metricsRecorderList = stats.NewMetricsRecorderList(cc.dopts.copts.StatsHandlers) cc.initIdleStateLocked() // Safe to call without the lock, since nothing else has a reference to cc. // idle状态管理 cc.idlenessMgr = idle.NewManager((*idler)(cc), cc.dopts.idleTimeout) return cc, nil } 这么一套下来可以看到，初始化ClientConn的时候并没有建立连接，所以猜测是在第一次发起RPC调用的时候才去尝试建立连接。还有一种验证方法是，把服务端关闭，尝试NewClient，是不会返回错误的。 ...

gRPC介绍 gRPC 是一种由 Google 开发的高性能远程过程调用（RPC）框架，适用于分布式系统间的通信。它基于 HTTP/2 进行传输，使用 Protocol Buffers 进行序列化，提供跨平台的兼容性。gRPC 的核心理念是让客户端像调用本地函数一样调用远程服务，简化服务间的调用流程。 通过编写与具体编程语言无关的 IDL （默认是 protobuf） 来定义 RPC 方法，gRPC 框架就会生成语言相关的客户端/服务端代码。 HTTP/2介绍 相比http1，具有更高的传输效率（多路复用：在同一个链连接上同时处理多个请求），更低的延迟（服务端推送，减少请求数量、简化header大小）、带宽利用率更高（头部压缩、数据流优先）、更安全（基于tls）。 http2具体特性有： 帧、消息、流：帧是小通信数据单元；消息由一个或多个帧组成。例如请求的消息和响应的消息；一个连接中包含多个流，每个流包含多个帧。帧通过流id进行标识属于哪个流 二进制分帧：每个消息由若干个帧组成，帧是最小传输单位，并且原来基于文本编码变成基于二进制，进一步减小帧大小 压缩header 多路复用：即在同一连接中的多个stream的传输互不影响 服务端推送 流量控制和资源优先级：流量控制以有效利用多路复用机制，确保只有接收者使用的数据会被传输。优先级机制可以确保重要的资源被优先传输。 启动服务端 通过官方的 helloworld 例子可以看到，服务端的启动分为三步： 创建gRPC的Server 将业务handler注册到Server 调用Server.Serve在端口上进行监听 第一步没什么好说的，注意下第二步注册进去的东西： // 注册进去的ServiceDesc var Greeter_ServiceDesc = grpc.ServiceDesc{ ServiceName: "helloworld.Greeter", HandlerType: (*GreeterServer)(nil), Methods: []grpc.MethodDesc{ { MethodName: "SayHello", Handler: _Greeter_SayHello_Handler, }, }, Streams: []grpc.StreamDesc{}, Metadata: "helloworld/helloworld.proto", } // Method对应的handler func _Greeter_SayHello_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) { in := new(HelloRequest) if err := dec(in); err != nil { return nil, err } if interceptor == nil { return srv.(GreeterServer).SayHello(ctx, in) } info := &grpc.UnaryServerInfo{ Server: srv, FullMethod: Greeter_SayHello_FullMethodName, } handler := func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(GreeterServer).SayHello(ctx, req.(*HelloRequest)) } return interceptor(ctx, in, info, handler) } 前两步比较简单，再来看第三步的如何建立连接并进行处理。类似标准库http的ListenAndServe，本质上就是创建一个死循环等待有新的连接到来，然后开新的goroutine去处理这个连接上的读写事件： ...