golang源码分析：grpc 链接池（3）resolver 、balancer和picker

// Resolver watches for the updates on the specified target.
// Updates include address updates and service config updates.
type Resolver interface {
// ResolveNow will be called by gRPC to try to resolve the target name
// again. It's just a hint, resolver can ignore this if it's not necessary.
//
// It could be called multiple times concurrently.
ResolveNow(ResolveNowOptions)
// Close closes the resolver.
Close()
}

func DialContext(ctx context.Context, target string, opts ...DialOption) (conn *ClientConn, err error) {
  resolverBuilder, err := cc.parseTargetAndFindResolver()
  rWrapper, err := newCCResolverWrapper(cc, resolverBuilder)
  cc.resolverWrapper = rWrapper

func (cc *ClientConn) parseTargetAndFindResolver() (resolver.Builder, error) {
rb = cc.getResolver(parsedTarget.Scheme)

func (cc *ClientConn) getResolver(scheme string) resolver.Builder {
return resolver.Get(scheme)

func init() {
  resolver.Register(&mockResolverBuilder{})
 }

func newCCResolverWrapper(cc *ClientConn, rb resolver.Builder) (*ccResolverWrapper, error) {
  ccr.resolver, err = rb.Build(cc.parsedTarget, ccr, rbo)

func (ac *addrConn) createTransport(addr resolver.Address, copts transport.ConnectOptions, connectDeadline time.Time) error {
ac.cc.resolveNow(resolver.ResolveNowOptions{})

func (ac *addrConn) resetTransport() {
  if err := ac.tryAllAddrs(addrs, connectDeadline); err != nil {
  ac.cc.resolveNow(resolver.ResolveNowOptions{})

func (cc *ClientConn) resolveNow(o resolver.ResolveNowOptions) {
  r := cc.resolverWrapper
  go r.resolveNow(o)

func (ccr *ccResolverWrapper) resolveNow(o resolver.ResolveNowOptions) {
  ccr.resolverMu.Lock()
  if !ccr.done.HasFired() {
  ccr.resolver.ResolveNow(o)

func (ccr *ccResolverWrapper) UpdateState(s resolver.State) error {
if err := ccr.cc.updateResolverState(ccr.curState, nil); err == balancer.ErrBadResolverState {
return balancer.ErrBadResolverState
}

func (cc *ClientConn) updateResolverState(s resolver.State, err error) error {
  cc.applyServiceConfigAndBalancer(sc, configSelector, s.Addresses)
  bw := cc.balancerWrapper
  uccsErr := bw.updateClientConnState(&balancer.ClientConnState{ResolverState: s, BalancerConfig: balCfg})

 grpc.WithDefaultServiceConfig(fmt.Sprintf(`{"LoadBalancingPolicy": "%s"}`, mybalancer.Name)

type Balancer interface {
// UpdateClientConnState is called by gRPC when the state of the ClientConn
// changes.  If the error returned is ErrBadResolverState, the ClientConn
// will begin calling ResolveNow on the active name resolver with
// exponential backoff until a subsequent call to UpdateClientConnState
// returns a nil error.  Any other errors are currently ignored.
UpdateClientConnState(ClientConnState) error
// ResolverError is called by gRPC when the name resolver reports an error.
ResolverError(error)
// UpdateSubConnState is called by gRPC when the state of a SubConn
// changes.
UpdateSubConnState(SubConn, SubConnState)
// Close closes the balancer. The balancer is not required to call
// ClientConn.RemoveSubConn for its existing SubConns.
Close()
}

type ClientConnState struct {
ResolverState resolver.State
// The parsed load balancing configuration returned by the builder's
// ParseConfig method, if implemented.
BalancerConfig serviceconfig.LoadBalancingConfig
}

func (ccb *ccBalancerWrapper) watcher() {
  for {
  select {
    case u := <-ccb.updateCh.Get():
    switch update := u.(type) {
    case *ccStateUpdate:
       ccb.handleClientConnStateChange(update.ccs)

func (ccb *ccBalancerWrapper) updateClientConnState(ccs *balancer.ClientConnState) error {
  ccb.updateCh.Put(&ccStateUpdate{ccs: ccs})
  select {
    case res = <-ccb.resultCh.Get():
    ccb.resultCh.Load()

func (ccb *ccBalancerWrapper) handleClientConnStateChange(ccs *balancer.ClientConnState) {
   ccb.resultCh.Put(ccb.balancer.UpdateClientConnState(*ccs))

func (b *baseBalancer) UpdateClientConnState(s balancer.ClientConnState) error {
  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})
      sc.Connect()
  b.regeneratePicker()
  b.cc.UpdateState(balancer.State{ConnectivityState: b.state, Picker: b.picker})

func (b *baseBalancer) UpdateSubConnState(sc balancer.SubConn, state balancer.SubConnState) {
  if (s == connectivity.Ready) != (oldS == connectivity.Ready) ||
  b.state == connectivity.TransientFailure {
  b.regeneratePicker()
  }
b.cc.UpdateState(balancer.State{ConnectivityState: b.state, Picker: b.picker})

func (ccb *ccBalancerWrapper) UpdateState(s balancer.State) {
  ccb.cc.blockingpicker.updatePicker(s.Picker)
  ccb.cc.csMgr.updateState(s.ConnectivityState)

func (r *randomPickerBuilder) Build(info base.PickerBuildInfo) balancer.Picker {

type PickerBuildInfo struct {
// ReadySCs is a map from all ready SubConns to the Addresses used to
// create them.
ReadySCs map[balancer.SubConn]SubConnInfo
}

func (b *baseBalancer) regeneratePicker() {
  for _, addr := range b.subConns.Keys() {
     sci, _ := b.subConns.Get(addr)
      if st, ok := b.scStates[sc]; ok && st == connectivity.Ready {
      readySCs[sc] = SubConnInfo{Address: addr}
      }
  }
  b.picker = b.pickerBuilder.Build(PickerBuildInfo{ReadySCs: readySCs})

// ClientConnState describes the state of a ClientConn relevant to the
// balancer.
type ClientConnState struct {
ResolverState resolver.State
// The parsed load balancing configuration returned by the builder's
// ParseConfig method, if implemented.
BalancerConfig serviceconfig.LoadBalancingConfig
}

func (gsb *Balancer) UpdateClientConnState(state balancer.ClientConnState) error {
return balToUpdate.UpdateClientConnState(state)

func (cc *ClientConn) getTransport(ctx context.Context, failfast bool, method string) (transport.ClientTransport, func(balancer.DoneInfo), error) {
return cc.blockingpicker.pick(ctx, failfast, balancer.PickInfo{
Ctx:            ctx,
FullMethodName: method,
})
}

func (pw *pickerWrapper) pick(ctx context.Context, failfast bool, info balancer.PickInfo) (transport.ClientTransport, func(balancer.DoneInfo), error) {
   pickResult, err := p.Pick(info)