vlambda博客
学习文章列表

syntax = "proto3";
package demo;
service DemoService { rpc SayHi(HiRequest) returns (HiResponse);}
message HiRequest { string name = 1;}
message HiResponse { string message = 1;}

package demo
...type DemoServiceClient interface { SayHiOK(ctx context.Context, in *HiRequest, opts ...grpc.CallOption) (*HiResponse, error)}
type demoServiceClient struct { cc *grpc.ClientConn}
// NewDemoServiceClient 业务代码中此方法来构造 *demoServiceClient 实例func NewDemoServiceClient(cc *grpc.ClientConn) DemoServiceClient { return &demoServiceClient{cc}}
func (c *demoServiceClient) SayHiOK(ctx context.Context, in *HiRequest, opts ...grpc.CallOption) (*HiResponse, error) { out := new(HiResponse) err := c.cc.Invoke(ctx, "/proto.DemoService/SayHiOK", in, out, opts...) if err != nil { return nil, err } return out, nil}...

import "mypkg/internal/ns"
...// 使用上节中封装的 ns.Dial 方法构造 *grpc.ClientConnconn, _ := ns.Dial("my-caller-service", "my-callee-service")// 构造 *demoServiceClientcli := demo.NewDemoServiceClient(conn)// 使用 *demoServiceClient 发起 RPC 调用res, _ := cli.SayHiOK(ctx, &proto.HiRequest{Name: "world"})...

// m 定义为一个全局 map，用于存放 [resolver 协议名 -> resolverBuilder] 键值对var m = make(map[string]Builder)
// Register 方法将指定 [resolver 协议名 -> resolverBuilder] 键值对存入 mapfunc Register(b Builder) { m[b.Scheme()] = b}
// Get 方法根据传入的 resolver 协议名返回对应的 resolverBuilderfunc Get(scheme string) Builder { if b, ok := m[scheme]; ok { return b } return nil}

func DialContext(ctx context.Context, target string, opts ...DialOption) (conn *ClientConn, err error) { // 构造 ClientConn 实例 cc := &ClientConn{ target: target, csMgr: &connectivityStateManager{}, conns: make(map[*addrConn]struct{}), dopts: defaultDialOptions(), blockingpicker: newPickerWrapper(), czData: new(channelzData), firstResolveEvent: grpcsync.NewEvent(), } cc.retryThrottler.Store((*retryThrottler)(nil)) cc.ctx, cc.cancel = context.WithCancel(context.Background())
 for _, opt := range opts { opt.apply(&cc.dopts) } ...
 // 如果用户指定了 timeout 超时配置，那么初始化一个带超时的 ctx // 如果 defer 阶段已超时，则抛出 j 错误 if cc.dopts.timeout > 0 { var cancel context.CancelFunc ctx, cancel = context.WithTimeout(ctx, cc.dopts.timeout) defer cancel() } defer func() { select { case <-ctx.Done(): conn, err = nil, ctx.Err() default: } }() ...
 // Name Resolver 核心逻辑，初始化 resolverBuilder，代码中首先会判断下用户是否指定 resolverBuilder // - 如果有指定 resolverBuilder，则直接使用此 resolverBuilder。 // - 如果用户没有指定 resolverBuilder，那么 grpc 做如下操作： // - 通过 parseTarget 方法解析用户传入的 target，本例中即 `ns://my-caller-service:@my-callee-service`，获取 Scheme（协议名）、authority（包含 callerService、calleeService）。 // - 查询指定协议对应的 resolverBuilder。 if cc.dopts.resolverBuilder == nil { // 解析用户传入的 target cc.parsedTarget = parseTarget(cc.target) // 通过协议名查表获取对应的 resolverBuilder cc.dopts.resolverBuilder = resolver.Get(cc.parsedTarget.Scheme) // 如果表中没查到对应的 resolverBuilder，则使用默认协议查询对应的 resolverBuilder // 默认协议为 `passthrough`，它会从用户解析的 target 中直接读取 endpoint 地址 if cc.dopts.resolverBuilder == nil { cc.parsedTarget = resolver.Target{ Scheme: resolver.GetDefaultScheme(), Endpoint: target, } cc.dopts.resolverBuilder = resolver.Get(cc.parsedTarget.Scheme) } } else { cc.parsedTarget = resolver.Target{Endpoint: target} } ...
 // 使用上面初始化的 resolverBuilder 构建 resolver // 初始化 resolverWrapper rWrapper, err := newCCResolverWrapper(cc) if err != nil { return nil, fmt.Errorf("failed to build resolver: %v", err) } cc.mu.Lock() cc.resolverWrapper = rWrapper cc.mu.Unlock()
 // 如果客户端配置了 WithBlock option，则会轮询 ClientConn 状态，如果 ClientConn 就绪，则返回 ClientConn。 // 如果直到 ctx 超时或被 Cancel ClientConn 依然未就绪，则抛出 ctx.Err() 错误。 if cc.dopts.block { for { s := cc.GetState() // 1. 如果 ClientConn 状态为 Ready 则返回此 ClientConn // 2. 如果 ClientConn 状态并非 Ready，且用户配置了 FailOnNonTempDialError，当前 ClientConn 状态为 TransientFailure，且 lbPicker 尝试和服务端实例建立连接时产生错误。根据错误性质做如下处理： // 2.1. 如果此错误是非临时性的错误，则抛出此错误 // 2.2. 如果此错误是临时性的错误，则继续轮询 ClientConn 状态，直至 ctx 超时或被外部 Cancel if s == connectivity.Ready { break } else if cc.dopts.copts.FailOnNonTempDialError && s == connectivity.TransientFailure { if err = cc.blockingpicker.connectionError(); err != nil { terr, ok := err.(interface { Temporary() bool }) if ok && !terr.Temporary() { return nil, err } } } if !cc.WaitForStateChange(ctx, s) { return nil, ctx.Err() } } }
 return cc, nil}

func newCCResolverWrapper(cc *ClientConn) (*ccResolverWrapper, error) { ... ccr := &ccResolverWrapper{ cc: cc, addrCh: make(chan []resolver.Address, 1), scCh: make(chan string, 1), }
 var err error // rb.Build() 调用指定 resolveBuilder 的 Build 方法，本例中会执行我们定义的 nsResolverBuilder.Builder() 方法 ccr.resolver, err = rb.Build(cc.parsedTarget, ccr, resolver.BuildOption{DisableServiceConfig: cc.dopts.disableServiceConfig}) if err != nil { return nil, err } return ccr, nil}

package ns
// init 将定义好的 NS Builder 注册到 resolver 包中func init() { resolver.Register(NewBuilder())}
...func (*nsResolverBuilder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOption) (resolver.Resolver, error) { ctx, cancel := context.WithCancel(context.Background()) r := &nsResolver{ target: target, cc: cc, ctx: ctx, cancel: cancel, } // 启动协程，轮询并更新指定 CalleeService 服务的实例变化 go r.watcher() return r, nil}...

package ns
...// watcher 轮询并更新指定 CalleeService 服务的实例变化func (r *nsResolver) watcher() { r.updateCC() ticker := time.NewTicker(syncNSInterval) for { select { // 当* nsResolver Close 时退出 case <-r.ctx.Done(): ticker.Stop() return case <-ticker.C: // 调用* nsResolver.updagteCC() 方法，更新实例地址 r.updateCC() } }}
// updateCC 更新 resolver.Resolver.ClientConn 配置func (r *nsResolver) updateCC() { // 从 NS 服务获取指定 target 的实例列表 instances, err := r.getInstances(r.target) // 如果获取实例列表失败，或者实例列表为空，则不更新 resolver 中实例列表 if err != nil || len(instances.CalleeIns) == 0 { logz.Warn("[mis] error retrieving instances from Mis", logz.Any("target", r.target), logz.Error(err)) return } ...
 // 组装实例列表 []resolver.Address // resolver.Address 结构体表示 grpc server 端实例地址 var newAddrs []resolver.Address for k := range instances.calleeIns { newAddrs = append(newAddrs, resolver.Address{ Addr: instances.CalleeIns[k], }) } ...
 // 更新实例列表 // grpc 底层 LB 组件对每个服务端实例创建一个 subConnection。并根据设定的 LB 策略，选择合适的 subConnection 处理某次 RPC 请求。 // 此处代码比较复杂，后续在 LB 相关原理文章中再做概述。 r.cc.UpdateState(resolver.State{Addresses: newAddrs})}...