DialContext

DialContext是客户端建立连接的入口函数，我们看看在这个函数里面做了哪些事情：

  1. func DialContext(ctx context.Context, target string, opts ...DialOption) (conn *ClientConn, err error) {
  3.     // 1.创建ClientConn结构体
  4.     cc := &ClientConn{
  5.         target:            target,
  6.         ...
  7.     }
  9.     // 2.解析target
  10.     cc.parsedTarget = grpcutil.ParseTarget(cc.target, cc.dopts.copts.Dialer != nil)
  12.     // 3.根据解析的target找到合适的resolverBuilder
  13.     resolverBuilder := cc.getResolver(cc.parsedTarget.Scheme)
  15.     // 4.创建Resolver
  16.     rWrapper, err := newCCResolverWrapper(cc, resolverBuilder)
  18.     // 5.完事
  19.     return cc, nil
  20. }

显而易见，在省略了亿点点细节之后，我们发现建立连接的过程其实也很简单，我们梳理一遍：
因为gRPC没有提供服务注册，服务发现的功能，所以需要开发者自己编写服务发现的逻辑：也就是Resolver——解析器。
在得到了解析的结果，也就是一连串的IP地址之后，需要对其中的IP进行选择，也就是Balancer。

Builder的获取

我们从Builder开始讲起。

  1. cc.parsedTarget = grpcutil.ParseTarget(cc.target, cc.dopts.copts.Dialer != nil)

关于ParseTarget的逻辑我们用简单一句话来概括：获取开发者传入的target参数的地址类型，在后续查找适合这种类型地址的Builder。
然后我们来看查找Builder的这部分操作，这部分代码比较简单，我在代码中加了一些注释：

  1. resolverBuilder := cc.getResolver(cc.parsedTarget.Scheme)
  3. func (cc *ClientConn) getResolver(scheme string) resolver.Builder {
  4.     // 先查看是否在配置中存在resolver
  5.     for _, rb := range cc.dopts.resolvers {
  6.         if scheme == rb.Scheme() {
  7.             return rb
  8.         }
  9.     }
  11.     // 如果配置中没有相应的resolver，再从注册的resolver中寻找
  12.     return resolver.Get(scheme)
  13. }
  15. // 可以看出，ResolverBuilder是从m这个map里面找到的
  16. func Get(scheme string) Builder {
  17.     if b, ok := m[scheme]; ok {
  18.         return b
  19.     }
  20.     return nil
  21. }

看到这里我们可以推测：对于每个**ResolverBuilder**，是需要提前注册的。
我们找到Resolver的代码中，果然发现他在init()的时候注册了自己。

  1. func init() {
  2.     resolver.Register(&passthroughBuilder{})
  3. }
  5. // 注册Resolver，即是把自己加入map中
  6. func Register(b Builder) {
  7.     m[b.Scheme()] = b
  8. }

至此，我们已经研究完了Builder的注册和获取。

ResolverWrapper的创建

回到ClientConn的创建过程中，在获取到了ResolverBuilder之后，进行下一步的操作：

  1. rWrapper, err := newCCResolverWrapper(cc, resolverBuilder)
  2. ...
  3. cc.resolverWrapper = rWrapper
  4. ...

在创建ResolverBuilder后，设置给cc，就没什么逻辑了
gRPC为了实现插件式的Resolver，因此采用了装饰器模式，创建了一个ResolverWrapper
我们看看在创建ResolverWrapper的细节：

  1. func newCCResolverWrapper(cc *ClientConn, rb resolver.Builder) (*ccResolverWrapper, error) {
  2.     ccr := &ccResolverWrapper{
  3.         cc:   cc,
  4.         done: grpcsync.NewEvent(),
  5.     }
  7.   // 根据传入的Builder，创建resolver，并放入wrapper中
  8.   ccr.resolver, err = rb.Build(cc.parsedTarget, ccr, rbo)
  9.      return ccr, nil
  10. }

好，到了这里我们可以暂停一下。
我们停下来思考一下我们需要实现的功能：为了解耦Resolver和Balancer，我们希望能够有一个中间的部分，接收到Resolver解析到的地址，然后对它们进行负载均衡。因此，在接下来的代码阅读过程中，我们可以带着这个问题：**Resolver**和**Balancer**的通信过程是什么样的？
再看上面的代码，ClientConn的创建已经结束了。那么我们可以推测，剩下的逻辑就在rb.Build(cc.parsedTarget, ccr, rbo)这一行代码里面。

此时的Builder是什么？

由于demo中没有使用解析器，会进入以下逻辑

  1. if resolverBuilder == nil {
  2.         channelz.Infof(logger, cc.channelzID, "scheme %q not registered, fallback to default scheme", cc.parsedTarget.Scheme)
  3.         cc.parsedTarget = resolver.Target{
  4.             Scheme:   resolver.GetDefaultScheme(),
  5.             Endpoint: target,
  6.         }
  7.         resolverBuilder = cc.getResolver(cc.parsedTarget.Scheme)
  8.         if resolverBuilder == nil {
  9.             return nil, fmt.Errorf("could not get resolver for default scheme: %q", cc.parsedTarget.Scheme)
  10.         }
  11.     }

调用cc.getResolver获取resolverBuilder

  1. func (cc *ClientConn) getResolver(scheme string) resolver.Builder {
  2.     for _, rb := range cc.dopts.resolvers {
  3.         if scheme == rb.Scheme() {
  4.             return rb
  5.         }
  6.     }
  7.     return resolver.Get(scheme)
  8. }

注意次的Scheme，是默认的passthrough
image.png

passthrough哪来的？

框架自带，用init注册

  1. package passthrough
  3. import "google.golang.org/grpc/resolver"
  5. const scheme = "passthrough"
  7. type passthroughBuilder struct{}
  9. func (*passthroughBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) {
  10.     r := &passthroughResolver{
  11.         target: target,
  12.         cc:     cc,
  13.     }
  14.     r.start()
  15.     return r, nil
  16. }
  18. func (*passthroughBuilder) Scheme() string {
  19.     return scheme
  20. }
  22. type passthroughResolver struct {
  23.     target resolver.Target
  24.     cc     resolver.ClientConn
  25. }
  27. func (r *passthroughResolver) start() {
  28.     r.cc.UpdateState(resolver.State{Addresses: []resolver.Address{{Addr: r.target.Endpoint}}})
  29. }
  31. func (*passthroughResolver) ResolveNow(o resolver.ResolveNowOptions) {}
  33. func (*passthroughResolver) Close() {}
  35. func init() {
  36.     resolver.Register(&passthroughBuilder{})
  37. }

Build

build的逻辑很简单，代用了start()
start的逻辑也很简单，调用了r.cc.UpdateState

1. r.cc.UpdateState

注意，此时的cc 是 ccr ，类型为*ccResolverWrapper

  1. func (ccr *ccResolverWrapper) UpdateState(s resolver.State) {
  2.     if ccr.done.HasFired() {
  3.         return
  4.     }
  5.     channelz.Infof(logger, ccr.cc.channelzID, "ccResolverWrapper: sending update to cc: %v", s)
  6.     if channelz.IsOn() {
  7.         ccr.addChannelzTraceEvent(s)
  8.     }
  9.     ccr.curState = s
  10.     ccr.poll(ccr.cc.updateResolverState(ccr.curState, nil))
  11. }

将ccr.curState 赋值为s后调用ccr.cc.updateResolverState()

2. ccr.cc.updateResolverState

  1. func (cc *ClientConn) updateResolverState(s resolver.State, err error) error {
  2.     .......
  3.     var ret error
  4.     if cc.dopts.disableServiceConfig || s.ServiceConfig == nil {
  5.         // 第一次进入这个函数调用
  6.         cc.maybeApplyDefaultServiceConfig(s.Addresses)
  7.         // TODO: do we need to apply a failing LB policy if there is no
  8.         // default, per the error handling design?
  9.     } else {
  10.         if sc, ok := s.ServiceConfig.Config.(*ServiceConfig); s.ServiceConfig.Err == nil && ok {
  11.             configSelector := iresolver.GetConfigSelector(s)
  12.             if configSelector != nil {
  13.                 if len(s.ServiceConfig.Config.(*ServiceConfig).Methods) != 0 {
  14.                     channelz.Infof(logger, cc.channelzID, "method configs in service config will be ignored due to presence of config selector")
  15.                 }
  16.             } else {
  17.                 configSelector = &defaultConfigSelector{sc}
  18.             }
  19.             cc.applyServiceConfigAndBalancer(sc, configSelector, s.Addresses)
  20.         } else {
  21.             ret = balancer.ErrBadResolverState
  22.             if cc.balancerWrapper == nil {
  23.                 var err error
  24.                 if s.ServiceConfig.Err != nil {
  25.                     err = status.Errorf(codes.Unavailable, "error parsing service config: %v", s.ServiceConfig.Err)
  26.                 } else {
  27.                     err = status.Errorf(codes.Unavailable, "illegal service config type: %T", s.ServiceConfig.Config)
  28.                 }
  29.                 cc.safeConfigSelector.UpdateConfigSelector(&defaultConfigSelector{cc.sc})
  30.                 cc.blockingpicker.updatePicker(base.NewErrPicker(err))
  31.                 cc.csMgr.updateState(connectivity.TransientFailure)
  32.                 cc.mu.Unlock()
  33.                 return ret
  34.             }
  35.         }
  36.     }
  38.     var balCfg serviceconfig.LoadBalancingConfig
  39.     if cc.dopts.balancerBuilder == nil && cc.sc != nil && cc.sc.lbConfig != nil {
  40.         balCfg = cc.sc.lbConfig.cfg
  41.     }
  43.     cbn := cc.curBalancerName
  44.     bw := cc.balancerWrapper
  45.     cc.mu.Unlock()
  46.     if cbn != grpclbName {
  47.         // Filter any grpclb addresses since we don't have the grpclb balancer.
  48.         for i := 0; i < len(s.Addresses); {
  49.             if s.Addresses[i].Type == resolver.GRPCLB {
  50.                 copy(s.Addresses[i:], s.Addresses[i+1:])
  51.                 s.Addresses = s.Addresses[:len(s.Addresses)-1]
  52.                 continue
  53.             }
  54.             i++
  55.         }
  56.     }
  57.     uccsErr := bw.updateClientConnState(&balancer.ClientConnState{ResolverState: s, BalancerConfig: balCfg})
  58.     if ret == nil {
  59.         ret = uccsErr // prefer ErrBadResolver state since any other error is
  60.         // currently meaningless to the caller.
  61.     }
  62.     return ret
  63. }

Balancer创建

3. cc.maybeApplyDefaultServiceConfig

上面ccr.cc.updateResolverState，会调用cc.maybeApplyDefaultServiceConfig

  1. func (cc *ClientConn) maybeApplyDefaultServiceConfig(addrs []resolver.Address) {
  2.     if cc.sc != nil {
  3.         cc.applyServiceConfigAndBalancer(cc.sc, nil, addrs)
  4.         return
  5.     }
  6.     if cc.dopts.defaultServiceConfig != nil {
  7.         cc.applyServiceConfigAndBalancer(cc.dopts.defaultServiceConfig, &defaultConfigSelector{cc.dopts.defaultServiceConfig}, addrs)
  8.     } else {
  9.         cc.applyServiceConfigAndBalancer(emptyServiceConfig, &defaultConfigSelector{emptyServiceConfig}, addrs)
  10.     }
  11. }

4. cc.applyServiceConfigAndBalancer

  1. func (cc *ClientConn) applyServiceConfigAndBalancer(sc *ServiceConfig, configSelector iresolver.ConfigSelector, addrs []resolver.Address) {
  2.     .....
  3.     if cc.dopts.balancerBuilder == nil {
  4.         var newBalancerName string
  5.         if cc.sc != nil && cc.sc.lbConfig != nil {
  6.             newBalancerName = cc.sc.lbConfig.name
  7.         } else {
  8.             var isGRPCLB bool
  9.             for _, a := range addrs {
  10.                 if a.Type == resolver.GRPCLB {
  11.                     isGRPCLB = true
  12.                     break
  13.                 }
  14.             }
  15.             if isGRPCLB {
  16.                 newBalancerName = grpclbName
  17.             } else if cc.sc != nil && cc.sc.LB != nil {
  18.                 newBalancerName = *cc.sc.LB
  19.             } else {
  20.                 newBalancerName = PickFirstBalancerName
  21.             }
  22.         }
  23.         cc.switchBalancer(newBalancerName)
  24.     } else if cc.balancerWrapper == nil {
  25.         // Balancer dial option was set, and this is the first time handling
  26.         // resolved addresses. Build a balancer with dopts.balancerBuilder.
  27.         cc.curBalancerName = cc.dopts.balancerBuilder.Name()
  28.         cc.balancerWrapper = newCCBalancerWrapper(cc, cc.dopts.balancerBuilder, cc.balancerBuildOpts)
  29.     }
  30. }

image.png
可以看到newBalancerName 为 pick_first

5. cc.switchBalancer

  1. func (cc *ClientConn) switchBalancer(name string) {
  2.     // 此时的builder是grpc.pickfirstBuilder
  3.     builder := balancer.Get(name)
  4.     ......
  5.     cc.curBalancerName = builder.Name()
  6.     cc.balancerWrapper = newCCBalancerWrapper(cc, builder, cc.balancerBuildOpts)
  7. }

默认balancerBuilder的获取与resolverBuilder一样，都是提前注册好的

  1. func init() {
  2.     balancer.Register(newPickfirstBuilder())
  3. }

6. newCCBalancerWrapper

可以看到Balancer也有一个Wrapper

  1. type ccBalancerWrapper struct {
  2.     cc         *ClientConn
  3.     balancerMu sync.Mutex // synchronizes calls to the balancer
  4.     balancer   balancer.Balancer
  5.     scBuffer   *buffer.Unbounded
  6.     done       *grpcsync.Event
  8.     mu       sync.Mutex
  9.     subConns map[*acBalancerWrapper]struct{}  // 没看懂这里为什么要map
  10. }
  1. func newCCBalancerWrapper(cc *ClientConn, b balancer.Builder, bopts balancer.BuildOptions) *ccBalancerWrapper {
  2.     ccb := &ccBalancerWrapper{
  3.         cc:       cc,
  4.         scBuffer: buffer.NewUnbounded(),
  5.         done:     grpcsync.NewEvent(),
  6.         subConns: make(map[*acBalancerWrapper]struct{}),
  7.     }
  8.     go ccb.watcher()
  9.     ccb.balancer = b.Build(ccb, bopts)
  10.     return ccb
  11. }
  13. // demo原因 这里返回的是pickfirstBuilder
  14. func (*pickfirstBuilder) Build(cc balancer.ClientConn, opt balancer.BuildOptions) balancer.Balancer {
  15.     return &pickfirstBalancer{cc: cc}
  16. }

�

addrConn连接

回到 2. ccr.cc.updateResolverState

7. bw.updateClientConnState

  1. func (ccb *ccBalancerWrapper) updateClientConnState(ccs *balancer.ClientConnState) error {
  2.     ccb.balancerMu.Lock()
  3.     defer ccb.balancerMu.Unlock()
  4.     return ccb.balancer.UpdateClientConnState(*ccs)
  5. }
  7. func (b *pickfirstBalancer) UpdateClientConnState(cs balancer.ClientConnState) error {
  8.     .....
  9.     if b.sc == nil { // 第一次进来 b.sc balancer.SubConn 肯定是nil，所以进入if
  10.         var err error
  11.         b.sc, err = b.cc.NewSubConn(cs.ResolverState.Addresses, balancer.NewSubConnOptions{})
  12.         if err != nil {
  13.             if logger.V(2) {
  14.                 logger.Errorf("pickfirstBalancer: failed to NewSubConn: %v", err)
  15.             }
  16.             b.state = connectivity.TransientFailure
  17.             b.cc.UpdateState(balancer.State{ConnectivityState: connectivity.TransientFailure,
  18.                 Picker: &picker{err: fmt.Errorf("error creating connection: %v", err)},
  19.             })
  20.             return balancer.ErrBadResolverState
  21.         }
  22.         b.state = connectivity.Idle
  23.         b.cc.UpdateState(balancer.State{ConnectivityState: connectivity.Idle, Picker: &picker{result: balancer.PickResult{SubConn: b.sc}}})
  24.         b.sc.Connect()
  25.     } else {
  26.         b.sc.UpdateAddresses(cs.ResolverState.Addresses)
  27.         b.sc.Connect()
  28.     }
  29.     return nil
  30. }

8. b.cc.NewSubConn

  1. func (ccb *ccBalancerWrapper) NewSubConn(addrs []resolver.Address, opts balancer.NewSubConnOptions) (balancer.SubConn, error) {
  2.     if len(addrs) <= 0 {
  3.         return nil, fmt.Errorf("grpc: cannot create SubConn with empty address list")
  4.     }
  5.     ccb.mu.Lock()
  6.     defer ccb.mu.Unlock()
  7.     if ccb.subConns == nil {
  8.         return nil, fmt.Errorf("grpc: ClientConn balancer wrapper was closed")
  9.     }
  11.     // 初始化addrConn
  12.     ac, err := ccb.cc.newAddrConn(addrs, opts)
  13.     if err != nil {
  14.         return nil, err
  15.     }
  16.     // 把addrConn再包装下
  17.     acbw := &acBalancerWrapper{ac: ac}
  18.     acbw.ac.mu.Lock()
  19.     ac.acbw = acbw
  20.     acbw.ac.mu.Unlock()
  21.     ccb.subConns[acbw] = struct{}{}
  22.     return acbw, nil
  23. }

addrConn为什么还需要包装，额，是要实现接口，不得不说grpc的设计模式很棒棒

  1. // acBalancerWrapper is a wrapper on top of ac for balancers.
  2. // It implements balancer.SubConn interface.
  3. type acBalancerWrapper struct {
  4.     mu sync.Mutex
  5.     ac *addrConn
  6. }

9. ccb.cc.newAddrConn

  1. func (cc *ClientConn) newAddrConn(addrs []resolver.Address, opts balancer.NewSubConnOptions) (*addrConn, error) {
  2.     ac := &addrConn{
  3.         state:        connectivity.Idle,
  4.         cc:           cc,
  5.         addrs:        addrs,
  6.         scopts:       opts,
  7.         dopts:        cc.dopts,
  8.         czData:       new(channelzData),
  9.         resetBackoff: make(chan struct{}),
  10.     }
  11.     ac.ctx, ac.cancel = context.WithCancel(cc.ctx)
  12.     // Track ac in cc. This needs to be done before any getTransport(...) is called.
  13.     cc.mu.Lock()
  14.     .....
  15.     // 这里设置了cc的conns字段
  16.     // 因此得知cc.conns == ccb.subConns
  17.     cc.conns[ac] = struct{}{}
  18.     cc.mu.Unlock()
  19.     return ac, nil
  20. }

回到7.updateClientConnState

10. b.cc.UpdateState

  1. func (ccb *ccBalancerWrapper) UpdateState(s balancer.State) {
  2.     ccb.mu.Lock()
  3.     defer ccb.mu.Unlock()
  4.     if ccb.subConns == nil {
  5.         return
  6.     }
  7.     ccb.cc.blockingpicker.updatePicker(s.Picker)
  8.     ccb.cc.csMgr.updateState(s.ConnectivityState)
  9. }

11. ccb.cc.blockingpicker.updatePicker

  1. // pickerWrapper是balance.picker的包装器。它阻塞某些拾取动作，并在拾取更新时解除阻塞
  2. // 它是在DialContext中初始化的
  3. type pickerWrapper struct {
  4.     mu         sync.Mutex
  5.     done       bool
  6.     blockingCh chan struct{}
  7.     picker     balancer.Picker
  8. }
  1. // 更新pickerWrapper，将picker 设置为balance.picker
  2. func (pw *pickerWrapper) updatePicker(p balancer.Picker) {
  3.     pw.mu.Lock()
  4.     if pw.done {
  5.         pw.mu.Unlock()
  6.         return
  7.     }
  8.     pw.picker = p
  9.     // pw.blockingCh should never be nil.
  10.     close(pw.blockingCh)
  11.     pw.blockingCh = make(chan struct{})
  12.     pw.mu.Unlock()
  13. }

12. ccb.cc.csMgr.updateState

  1. // 根据上面流程，这里state是Idle状态
  2. // 这里只是保证connectivityStateManager的state，与传过来的state 保持一致
  3. func (csm *connectivityStateManager) updateState(state connectivity.State) {
  4.     csm.mu.Lock()
  5.     defer csm.mu.Unlock()
  6.     if csm.state == connectivity.Shutdown {
  7.         return
  8.     }
  9.     if csm.state == state {
  10.         return
  11.     }
  12.     csm.state = state
  13.     channelz.Infof(logger, csm.channelzID, "Channel Connectivity change to %v", state)
  14.     if csm.notifyChan != nil {
  15.         // There are other goroutines waiting on this channel.
  16.         close(csm.notifyChan)
  17.         csm.notifyChan = nil
  18.     }
  19. }

回到7.updateClientConnState

13. b.sc.Connect()

  1. func (acbw *acBalancerWrapper) Connect() {
  2.     acbw.mu.Lock()
  3.     defer acbw.mu.Unlock()
  4.     acbw.ac.connect()
  5. }
  8. func (ac *addrConn) connect() error {
  9.     ac.mu.Lock()
  10.     if ac.state == connectivity.Shutdown {
  11.         ac.mu.Unlock()
  12.         return errConnClosing
  13.     }
  15.     // 必须是Idle才能过来进行连接
  16.     if ac.state != connectivity.Idle {
  17.         ac.mu.Unlock()
  18.         return nil
  19.     }
  20.     // 更新锁内的连通性状态，以防止后续或并发调用多次重置传输
  21.     // 很明显，将ac的状态更新为Connecting了
  22.     ac.updateConnectivityState(connectivity.Connecting, nil)
  23.     ac.mu.Unlock()
  25.     // Start a goroutine connecting to the server asynchronously.
  26.     go ac.resetTransport()
  27.     return nil
  28. }

14. ac.resetTransport

  1. func (ac *addrConn) resetTransport() {
  2.     for i := 0; ; i++ {
  3.         if i > 0 {
  4.             ac.cc.resolveNow(resolver.ResolveNowOptions{})
  5.         }
  7.         ac.mu.Lock()
  8.         if ac.state == connectivity.Shutdown {
  9.             ac.mu.Unlock()
  10.             return
  11.         }
  12.         .......
  13.         ac.updateConnectivityState(connectivity.Connecting, nil)
  14.         ac.transport = nil
  15.         ac.mu.Unlock()
  17.         newTr, addr, reconnect, err := ac.tryAllAddrs(addrs, connectDeadline)
  18.         if err != nil {
  19.             .....
  20.         }
  22.         ac.mu.Lock()
  23.         if ac.state == connectivity.Shutdown {
  24.             ac.mu.Unlock()
  25.             newTr.Close()
  26.             return
  27.         }
  28.         ac.curAddr = addr
  29.         ac.transport = newTr
  30.         ac.backoffIdx = 0
  32.         hctx, hcancel := context.WithCancel(ac.ctx)
  33.         ac.startHealthCheck(hctx)
  34.         ac.mu.Unlock()
  36.         // 阻塞，直到创建的传输关闭。当这种情况发生时，从addr列表的顶部重新启动。
  37.         <-reconnect.Done()
  38.         hcancel()
  39.         重新启动连接-循环的顶部将设置state为connecting。这是违反当前的连接语义文档，但是它允许优雅
  40.         的行为尚未分派RPC -不幸的时间将导致RPC失败，即使TRANSIENT_FAILURE状态(由文档调用)将是瞬
  41.         时的。
  42.         理想情况下，我们应该在这里过渡到Idle并阻塞，直到出现RPC活动导致平衡器请求重新连接关联的
  43.         SubConn。
  44.     }
  45. }

15. ac.tryAllAddrs

tryAllAddrs尝试创建到该地址的连接，成功一个就返回传输、地址和一个Event。当返回的传输断开连接时触发事件
这里埋个问题，第一个成功就返回了，那其他的链接是怎么链接上的

  1. func (ac *addrConn) tryAllAddrs(addrs []resolver.Address, connectDeadline time.Time) (transport.ClientTransport, resolver.Address, *grpcsync.Event, error) {
  2.     var firstConnErr error
  3.     for _, addr := range addrs {
  4.         ac.mu.Lock()
  5.         if ac.state == connectivity.Shutdown {
  6.             ac.mu.Unlock()
  7.             return nil, resolver.Address{}, nil, errConnClosing
  8.         }
  10.         ac.cc.mu.RLock()
  11.         ac.dopts.copts.KeepaliveParams = ac.cc.mkp
  12.         ac.cc.mu.RUnlock()
  14.         copts := ac.dopts.copts
  15.         if ac.scopts.CredsBundle != nil {
  16.             copts.CredsBundle = ac.scopts.CredsBundle
  17.         }
  18.         ac.mu.Unlock()
  20.         channelz.Infof(logger, ac.channelzID, "Subchannel picks a new address %q to connect", addr.Addr)
  22.         newTr, reconnect, err := ac.createTransport(addr, copts, connectDeadline)
  23.         if err == nil {
  24.             return newTr, addr, reconnect, nil
  25.         }
  26.         if firstConnErr == nil {
  27.             firstConnErr = err
  28.         }
  29.         ac.cc.updateConnectionError(err)
  30.     }
  32.     // Couldn't connect to any address.
  33.     return nil, resolver.Address{}, nil, firstConnErr
  34. }

16. ac.createTransport

createTransport创建到addr的连接。它返回传输和成功情况下的事件。事件在返回的传输时触发断开连接。

  1. func (ac *addrConn) createTransport(addr resolver.Address, copts transport.ConnectOptions, connectDeadline time.Time) (transport.ClientTransport, *grpcsync.Event, error) {
  2.     prefaceReceived := make(chan struct{})
  3.     onCloseCalled := make(chan struct{})
  4.     reconnect := grpcsync.NewEvent()
  6.     // addr.ServerName takes precedent over ClientConn authority, if present.
  7.     if addr.ServerName == "" {
  8.         addr.ServerName = ac.cc.authority
  9.     }
  11.     once := sync.Once{}
  12.     onGoAway := func(r transport.GoAwayReason) {
  13.         ac.mu.Lock()
  14.         ac.adjustParams(r)
  15.         once.Do(func() {
  16.             if ac.state == connectivity.Ready {
  17.                 // Prevent this SubConn from being used for new RPCs by setting its
  18.                 // state to Connecting.
  19.                 //
  20.                 // TODO: this should be Idle when grpc-go properly supports it.
  21.                 ac.updateConnectivityState(connectivity.Connecting, nil)
  22.             }
  23.         })
  24.         ac.mu.Unlock()
  25.         reconnect.Fire()
  26.     }
  28.     onClose := func() {
  29.         ac.mu.Lock()
  30.         once.Do(func() {
  31.             if ac.state == connectivity.Ready {
  32.                 // Prevent this SubConn from being used for new RPCs by setting its
  33.                 // state to Connecting.
  34.                 //
  35.                 // TODO: this should be Idle when grpc-go properly supports it.
  36.                 ac.updateConnectivityState(connectivity.Connecting, nil)
  37.             }
  38.         })
  39.         ac.mu.Unlock()
  40.         close(onCloseCalled)
  41.         reconnect.Fire()
  42.     }
  44.     onPrefaceReceipt := func() {
  45.         close(prefaceReceived)
  46.     }
  48.     connectCtx, cancel := context.WithDeadline(ac.ctx, connectDeadline)
  49.     defer cancel()
  50.     if channelz.IsOn() {
  51.         copts.ChannelzParentID = ac.channelzID
  52.     }
  54.     newTr, err := transport.NewClientTransport(connectCtx, ac.cc.ctx, addr, copts, onPrefaceReceipt, onGoAway, onClose)
  55.     if err != nil {
  56.         // newTr is either nil, or closed.
  57.         channelz.Warningf(logger, ac.channelzID, "grpc: addrConn.createTransport failed to connect to %v. Err: %v. Reconnecting...", addr, err)
  58.         return nil, nil, err
  59.     }
  61.     select {
  62.     case <-time.After(time.Until(connectDeadline)):
  63.         // We didn't get the preface in time.
  64.         newTr.Close()
  65.         channelz.Warningf(logger, ac.channelzID, "grpc: addrConn.createTransport failed to connect to %v: didn't receive server preface in time. Reconnecting...", addr)
  66.         return nil, nil, errors.New("timed out waiting for server handshake")
  67.     case <-prefaceReceived:
  68.         // We got the preface - huzzah! things are good.
  69.     case <-onCloseCalled:
  70.         // The transport has already closed - noop.
  71.         return nil, nil, errors.New("connection closed")
  72.         // TODO(deklerk) this should bail on ac.ctx.Done(). Add a test and fix.
  73.     }
  74.     return newTr, reconnect, nil
  75. }

17. transport.NewClientTransport

  1. func NewClientTransport(connectCtx, ctx context.Context, addr resolver.Address, opts ConnectOptions, onPrefaceReceipt func(), onGoAway func(GoAwayReason), onClose func()) (ClientTransport, error) {
  2.     return newHTTP2Client(connectCtx, ctx, addr, opts, onPrefaceReceipt, onGoAway, onClose)
  3. }

newHTTP2Client基于HTTP2构造一个连接的ClientTransport到addr，并开始在其上接收消息。如果失败，则返回非nil错误失败。

  1. func newHTTP2Client(connectCtx, ctx context.Context, addr resolver.Address, opts ConnectOptions, onPrefaceReceipt func(), onGoAway func(GoAwayReason), onClose func()) (_ *http2Client, err error) {
  2.     scheme := "http"
  3.     ...
  4.     // 经理千心万苦，终于到了dial操作
  5.     conn, err := dial(connectCtx, opts.Dialer, addr, opts.UseProxy, opts.UserAgent)
  6.     ...
  7.     // Keepalive相关
  8.     kp := opts.KeepaliveParams
  9.     ...
  11.     var (
  12.         isSecure bool
  13.         authInfo credentials.AuthInfo
  14.     )
  15.     // transportCreds,perRPCCreds https，凭证相关
  16.     transportCreds := opts.TransportCredentials
  17.     perRPCCreds := opts.PerRPCCredentials
  19.     if b := opts.CredsBundle; b != nil {
  20.         if t := b.TransportCredentials(); t != nil {
  21.             transportCreds = t
  22.         }
  23.         if t := b.PerRPCCredentials(); t != nil {
  24.             perRPCCreds = append(perRPCCreds, t)
  25.         }
  26.     }
  27.     .....
  29.     // 窗口，writeBufSize，readBufSize，maxHeaderListSize相关
  30.     dynamicWindow := true
  31.     icwz := int32(initialWindowSize)
  32.     if opts.InitialConnWindowSize >= defaultWindowSize {
  33.         icwz = opts.InitialConnWindowSize
  34.         dynamicWindow = false
  35.     }
  36.     writeBufSize := opts.WriteBufferSize
  37.     readBufSize := opts.ReadBufferSize
  38.     maxHeaderListSize := defaultClientMaxHeaderListSize
  39.     if opts.MaxHeaderListSize != nil {
  40.         maxHeaderListSize = *opts.MaxHeaderListSize
  41.     }
  42.     // 初始化http2Client
  43.     t := &http2Client{
  44.         ctx:                   ctx,
  45.         ctxDone:               ctx.Done(), // Cache Done chan.
  46.         cancel:                cancel,
  47.         userAgent:             opts.UserAgent,
  48.         conn:                  conn,
  49.         remoteAddr:            conn.RemoteAddr(),
  50.         localAddr:             conn.LocalAddr(),
  51.         authInfo:              authInfo,
  52.         readerDone:            make(chan struct{}),
  53.         writerDone:            make(chan struct{}),
  54.         goAway:                make(chan struct{}),
  55.         framer:                newFramer(conn, writeBufSize, readBufSize, maxHeaderListSize),
  56.         fc:                    &trInFlow{limit: uint32(icwz)},
  57.         scheme:                scheme,
  58.         activeStreams:         make(map[uint32]*Stream),
  59.         isSecure:              isSecure,
  60.         perRPCCreds:           perRPCCreds,
  61.         kp:                    kp,
  62.         statsHandler:          opts.StatsHandler,
  63.         initialWindowSize:     initialWindowSize,
  64.         onPrefaceReceipt:      onPrefaceReceipt,
  65.         nextID:                1,
  66.         maxConcurrentStreams:  defaultMaxStreamsClient,
  67.         streamQuota:           defaultMaxStreamsClient,
  68.         streamsQuotaAvailable: make(chan struct{}, 1),
  69.         czData:                new(channelzData),
  70.         onGoAway:              onGoAway,
  71.         onClose:               onClose,
  72.         keepaliveEnabled:      keepaliveEnabled,
  73.         bufferPool:            newBufferPool(),
  74.     }
  76.     ....
  77.     // 健康检查，跟server端逻辑一样
  78.     if t.keepaliveEnabled {
  79.         t.kpDormancyCond = sync.NewCond(&t.mu)
  80.         go t.keepalive()
  81.     }
  82.     // 为传入消息启动阅读器例行程序。每个传输都有一个专用的从网络中读取HTTP2帧
  83.     // 的goroutine。然后将帧发送给相应的流实体。
  84.     // 属于数据交互部分了
  85.     go t.reader()
  87.     // 看grpc抓包分析的部分，有个数据帧是Magic,这个就是发送Magic逻辑
  88.     n, err := t.conn.Write(clientPreface)
  89.     ...
  90.     // 看grpc抓包分析的部分，有个数据帧是SETTINGS,这个就是发送SETTINGS逻辑
  91.     ...
  92.     err = t.framer.fr.WriteSettings(ss...)
  93.     if err != nil {
  94.         t.Close()
  95.         return nil, connectionErrorf(true, err, "transport: failed to write initial settings frame: %v", err)
  96.     }
  98.     // 连接建立，如果窗口大小需要调整,发WINDOW_UPDATE数据帧
  99.     if delta := uint32(icwz - defaultWindowSize); delta > 0 {
  100.         if err := t.framer.fr.WriteWindowUpdate(0, delta); err != nil {
  101.             t.Close()
  102.             return nil, connectionErrorf(true, err, "transport: failed to write window update: %v", err)
  103.         }
  104.     }
  106.     // 为该连接分配一个ID
  107.     t.connectionID = atomic.AddUint64(&clientConnectionCounter, 1)
  109.     // 这里是将自带缓冲区写入net.conn
  110.     if err := t.framer.writer.Flush(); err != nil {
  111.         return nil, err
  112.     }
  113.     go func() {
  114.         t.loopy = newLoopyWriter(clientSide, t.framer, t.controlBuf, t.bdpEst)
  115.         err := t.loopy.run()
  116.         if err != nil {
  117.             if logger.V(logLevel) {
  118.                 logger.Errorf("transport: loopyWriter.run returning. Err: %v", err)
  119.             }
  120.         }
  121.         // If it's a connection error, let reader goroutine handle it
  122.         // since there might be data in the buffers.
  123.         if _, ok := err.(net.Error); !ok {
  124.             t.conn.Close()
  125.         }
  126.         close(t.writerDone)
  127.     }()
  128.     return t, nil
  129. }
  1. const (
  2.     // ClientPreface is the string that must be sent by new
  3.     // connections from clients.
  4.     ClientPreface = "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"
  6.     // SETTINGS_MAX_FRAME_SIZE default
  7.     // http://http2.github.io/http2-spec/#rfc.section.6.5.2
  8.     initialMaxFrameSize = 16384
  10.     // NextProtoTLS is the NPN/ALPN protocol negotiated during
  11.     // HTTP/2's TLS setup.
  12.     NextProtoTLS = "h2"
  14.     // http://http2.github.io/http2-spec/#SettingValues
  15.     initialHeaderTableSize = 4096
  17.     initialWindowSize = 65535 // 6.9.2 Initial Flow Control Window Size
  19.     defaultMaxReadFrameSize = 1 << 20
  20. )
  22. var (
  23.     clientPreface = []byte(ClientPreface)
  24. )
  26. const (
  27.     FrameData         FrameType = 0x0
  28.     FrameHeaders      FrameType = 0x1
  29.     FramePriority     FrameType = 0x2
  30.     FrameRSTStream    FrameType = 0x3
  31.     FrameSettings     FrameType = 0x4
  32.     FramePushPromise  FrameType = 0x5
  33.     FramePing         FrameType = 0x6
  34.     FrameGoAway       FrameType = 0x7
  35.     FrameWindowUpdate FrameType = 0x8
  36.     FrameContinuation FrameType = 0x9
  37. )
  39. var frameName = map[FrameType]string{
  40.     FrameData:         "DATA",
  41.     FrameHeaders:      "HEADERS",
  42.     FramePriority:     "PRIORITY",
  43.     FrameRSTStream:    "RST_STREAM",
  44.     FrameSettings:     "SETTINGS",
  45.     FramePushPromise:  "PUSH_PROMISE",
  46.     FramePing:         "PING",
  47.     FrameGoAway:       "GOAWAY",
  48.     FrameWindowUpdate: "WINDOW_UPDATE",
  49.     FrameContinuation: "CONTINUATION",
  50. }

18. dial

拿到4层连接

  1. func dial(ctx context.Context, fn func(context.Context, string) (net.Conn, error), addr resolver.Address, useProxy bool, grpcUA string) (net.Conn, error) {
  2.     address := addr.Addr
  3.     networkType, ok := networktype.Get(addr)
  4.     if fn != nil {
  5.         if networkType == "unix" && !strings.HasPrefix(address, "\x00") {
  6.             // For backward compatibility, if the user dialed "unix:///path",
  7.             // the passthrough resolver would be used and the user's custom
  8.             // dialer would see "unix:///path". Since the unix resolver is used
  9.             // and the address is now "/path", prepend "unix://" so the user's
  10.             // custom dialer sees the same address.
  11.             return fn(ctx, "unix://"+address)
  12.         }
  13.         return fn(ctx, address)
  14.     }
  15.     if !ok {
  16.         networkType, address = parseDialTarget(address)
  17.     }
  18.     if networkType == "tcp" && useProxy {
  19.         return proxyDial(ctx, address, grpcUA)
  20.     }
  21.     return (&net.Dialer{}).DialContext(ctx, networkType, address)
  22. }
  24. func proxyDial(ctx context.Context, addr string, grpcUA string) (conn net.Conn, err error) {
  25.     newAddr := addr
  26.     proxyURL, err := mapAddress(ctx, addr)
  27.     if err != nil {
  28.         return nil, err
  29.     }
  30.     if proxyURL != nil {
  31.         newAddr = proxyURL.Host
  32.     }
  34.     conn, err = (&net.Dialer{}).DialContext(ctx, "tcp", newAddr)
  35.     if err != nil {
  36.         return
  37.     }
  38.     if proxyURL != nil {
  39.         // proxy is disabled if proxyURL is nil.
  40.         conn, err = doHTTPConnectHandshake(ctx, conn, addr, proxyURL, grpcUA)
  41.     }
  42.     return
  43. }

回到14. ac.resetTransport

  1. // 设置ac的相关信息
  2. ac.curAddr = addr
  3. ac.transport = newTr
  4. ac.backoffIdx = 0
  6. hctx, hcancel := context.WithCancel(ac.ctx)
  7. //开启 健康检查
  8. ac.startHealthCheck(hctx)
  9. ac.mu.Unlock()
  12. //如果请求并配置了健康检查，startHealthCheck将启动健康检查流(RPC)来监视此连接的健康统计信息。
  13. 当LB通道健康检查满足以下要求时，启用LB通道健康检查:
  14. 1. 它不被用户使用WithDisableHealthCheck DialOption禁用
  15. 2. internal.HealthCheckFunc是通过导入grpc/health包设置的
  16. 3.提供了一个带有非空healthCheckConfig字段的服务配置
  17. 4. 负载均衡器请求它
  19. //如果没有启动健康检查流，它将addrConn设置为READY。
  20. //呼叫者必须保持ac.mu。
  21. //因此我们的直连demo，不进行HealthCheck
  22. func (ac *addrConn) startHealthCheck(ctx context.Context) {
  23.     ...
  24. }

至此，连接建立完成

总结

image.png
�