golang源码分析：grpc 链接池（5）自定义组件和框架交互流程

func (*mockResolverBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
      r.start()
      r.cc.UpdateState(resolver.State{Addresses: addrs})

func DialContext(ctx context.Context, target string, opts ...DialOption) (conn *ClientConn, err error) {
      rWrapper, err := newCCResolverWrapper(cc, resolverBuilder)
      if cc.dopts.block {
              cc.Connect()
      s := cc.GetState()
      if s == connectivity.Ready {
      cc.balancerWrapper = newCCBalancerWrapper(cc, balancer.BuildOptions{
      rWrapper, err := newCCResolverWrapper(cc, resolverBuilder)
      cc.resolverWrapper = rWrapper

func newCCResolverWrapper(cc *ClientConn, rb resolver.Builder) (*ccResolverWrapper, error) {
    ccr := &ccResolverWrapper{
    cc:   cc,
    done: grpcsync.NewEvent(),
  }
  ccr.resolver, err = rb.Build(cc.parsedTarget, ccr, rbo)

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

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

func (ccb *ccBalancerWrapper) updateClientConnState(ccs *balancer.ClientConnState) error {
      ccb.updateCh.Put(&ccStateUpdate{ccs: ccs})

type ccStateUpdate struct {
  ccs *balancer.ClientConnState
}

func (ccb *ccBalancerWrapper) watcher() {
      case u := <-ccb.updateCh.Get():
              case *ccStateUpdate:
        ccb.handleClientConnStateChange(update.ccs)
      case *scStateUpdate:
        ccb.handleSubConnStateChange(update)
      case *exitIdleUpdate:
        ccb.handleExitIdle()
      case *resolverErrorUpdate:
        ccb.handleResolverError(update.err)
      case *switchToUpdate:
        ccb.handleSwitchTo(update.name)
      case *subConnUpdate:
        ccb.handleRemoveSubConn(update.acbw)

    func (cc *ClientConn) Connect() {
      cc.balancerWrapper.exitIdle()

    func (ccb *ccBalancerWrapper) exitIdle() {
      ccb.updateCh.Put(&exitIdleUpdate{})

func (ccb *ccBalancerWrapper) switchTo(name string) {
      ccb.updateCh.Put(&switchToUpdate{name: name})
func (ccb *ccBalancerWrapper) resolverError(err error) {
  ccb.updateCh.Put(&resolverErrorUpdate{err: err})
func (ccb *ccBalancerWrapper) updateSubConnState(sc balancer.SubConn, s connectivity.State, err error) {
    ccb.updateCh.Put(&scStateUpdate{
    sc:    sc,
    state: s,
    err:   err,
  }

func (ccb *ccBalancerWrapper) handleClientConnStateChange(ccs *balancer.ClientConnState) {
      ccs.ResolverState.Addresses = addrs
      ccb.resultCh.Put(ccb.balancer.UpdateClientConnState(*ccs))

func (ccb *ccBalancerWrapper) handleSubConnStateChange(update *scStateUpdate) {
      ccb.balancer.UpdateSubConnState(update.sc, balancer.SubConnState{ConnectivityState: update.state, ConnectionError: update.err})

func (ccb *ccBalancerWrapper) handleSwitchTo(name string) {
      builder := balancer.Get(name)
      if err := ccb.balancer.SwitchTo(builder); err != nil {

func (ccb *ccBalancerWrapper) NewSubConn(addrs []resolver.Address, opts balancer.NewSubConnOptions) (balancer.SubConn, error) {
      ac, err := ccb.cc.newAddrConn(addrs, opts)
      acbw := &acBalancerWrapper{ac: ac}
      ac.acbw = acbw

balancer.Register( base.NewBalancerBuilder

return base.NewBalancerBuilder(Name, picker.NewRandomPickerBuilder(), base.Config{HealthCheck: true})

func NewBalancerBuilder(name string, pb PickerBuilder, config Config) balancer.Builder {

type baseBuilder struct {
  name          string
  pickerBuilder PickerBuilder
  config        Config
}

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,
    state:    connectivity.Connecting,
  }

func (b *baseBalancer) UpdateSubConnState(sc balancer.SubConn, state balancer.SubConnState) {
  case connectivity.Idle:
    sc.Connect()
  case connectivity.Shutdown:
    // When an address was removed by resolver, b called RemoveSubConn but
    // kept the sc's state in scStates. Remove state for this sc here.
    delete(b.scStates, sc)
  case connectivity.TransientFailure:

 func (gsb *Balancer) ExitIdle() {
    for sc := range balToUpdate.subconns {
       sc.Connect()

func (b *baseBalancer) UpdateClientConnState(s balancer.ClientConnState) error {
      for _, a := range s.ResolverState.Addresses {
        if _, ok := b.subConns.Get(a); !ok {
          sc, err := b.cc.NewSubConn([]resolver.Address{a}, balancer.NewSubConnOptions{HealthCheckEnabled: b.config.HealthCheck})
          sc.Connect()

func (cc *ClientConn) applyServiceConfigAndBalancer(sc *ServiceConfig, configSelector iresolver.ConfigSelector, addrs []resolver.Address) {
      cc.balancerWrapper.switchTo(newBalancerName)

func (gsb *Balancer) SwitchTo(builder balancer.Builder) error {
    bw := &balancerWrapper{
    gsb: gsb,

    lastState: balancer.State{
      ConnectivityState: connectivity.Connecting,
      Picker:            base.NewErrPicker(balancer.ErrNoSubConnAvailable),
    },
    subconns: make(map[balancer.SubConn]bool),
  }
      newBalancer := builder.Build(bw, gsb.bOpts)
      bw.Balancer = newBalancer

func (r *randomPicker) Pick(_ balancer.PickInfo) (balancer.PickResult, error) {

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

func newClientStreamWithParams(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, mc serviceconfig.MethodConfig, onCommit, doneFunc func(), opts ...CallOption) (_ iresolver.ClientStream, err error) {
      if err := cs.newAttemptLocked(false /* isTransparent */); err != nil {

func (cs *clientStream) newAttemptLocked(isTransparent bool) (retErr error) {
      t, done, err := cs.cc.getTransport(ctx, cs.callInfo.failFast, cs.callHdr.Method)

func (cc *ClientConn) getTransport(ctx context.Context, failfast bool, method string) (transport.ClientTransport, func(balancer.DoneInfo), error) {
    t, done, err := 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) {
      p := pw.picker
      pickResult, err := p.Pick(info)
      acw, ok := pickResult.SubConn.(*acBalancerWrapper)
      if t := acw.getAddrConn().getReadyTransport(); t != nil {
func (ac *addrConn) getReadyTransport() transport.ClientTransport {
      if ac.state == connectivity.Ready {
    return ac.transport

 func (acbw *acBalancerWrapper) getAddrConn() *addrConn {
      return acbw.ac

func (cc *ClientConn) newAddrConn(addrs []resolver.Address, opts balancer.NewSubConnOptions) (*addrConn, error) {

func (ac *addrConn) updateConnectivityState(s connectivity.State, lastErr error) {
      ac.cc.handleSubConnStateChange(ac.acbw, s, lastErr)

func (ac *addrConn) connect() error {
      ac.updateConnectivityState(connectivity.Connecting, nil)
      ac.resetTransport()

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

func (ac *addrConn) tryAllAddrs(addrs []resolver.Address, connectDeadline time.Time) error {
      for _, addr := range addrs {
        err := ac.createTransport(addr, copts, connectDeadline)

func (ac *addrConn) createTransport(addr resolver.Address, copts transport.ConnectOptions, connectDeadline time.Time) error {
        newTr, err := transport.NewClientTransport(connectCtx, ac.cc.ctx, addr, copts, func() { prefaceReceived.Fire() }, onGoAway, onClose)

func (acbw *acBalancerWrapper) Connect() {
      go acbw.ac.connect()

func (acbw *acBalancerWrapper) UpdateAddresses(addrs []resolver.Address) {
        newAC, err := cc.newAddrConn(addrs, opts)
        acbw.ac = newAC
          if acState != connectivity.Idle {
      go newAC.connect()
    }

func (gsb *Balancer) UpdateSubConnState(sc balancer.SubConn, state balancer.SubConnState) {
    if state.ConnectivityState == connectivity.Shutdown {
    bw.gsb.mu.Lock()
    delete(bw.subconns, sc)
     balToUpdate.UpdateSubConnState(sc, state)

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