func main() {// Set up a connection to the server.//建立一个链接conn, err := grpc.Dial(address, grpc.WithInsecure())if err != nil {log.Fatalf("did not connect: %v", err)}defer conn.Close()c := pb.NewGreeterClient(conn)// Contact the server and print out its response.name := defaultNameif len(os.Args) > 1 {name = os.Args[1]}//调用函数r, err := c.SayHello(context.Background(), &pb.HelloRequest{Name: name})if err != nil {log.Fatalf("could not greet: %v", err)}log.Printf("Greeting: %s", r.Message)
}

func DialContext(ctx context.Context, target string, opts ...DialOption) (conn *ClientConn, err error) {cc := &ClientConn{target: target,conns:  make(map[Address]*addrConn),}cc.ctx, cc.cancel = context.WithCancel(context.Background())defer func() {select {case <-ctx.Done():conn, err = nil, ctx.Err()default:}if err != nil {cc.Close()}}()//设置grpc的各种选项for _, opt := range opts {opt(&cc.dopts)}// Set defaults.if cc.dopts.codec == nil {//默认用protobuf编解码cc.dopts.codec = protoCodec{}}if cc.dopts.bs == nil {cc.dopts.bs = DefaultBackoffConfig}creds := cc.dopts.copts.TransportCredentials//验证信息if creds != nil && creds.Info().ServerName != "" {cc.authority = creds.Info().ServerName} else {colonPos := strings.LastIndex(target, ":")if colonPos == -1 {colonPos = len(target)}cc.authority = target[:colonPos]}var ok boolwaitC := make(chan error, 1)//启动一个goroutine启动名字服务器(类似dns)go func() {var addrs []Addressif cc.dopts.balancer == nil {// Connect to target directly if balancer is nil.// 如果没设置负载均衡器，则直接连接addrs = append(addrs, Address{Addr: target})} else {var credsClone credentials.TransportCredentialsif creds != nil {credsClone = creds.Clone()}config := BalancerConfig{DialCreds: credsClone,}//启动负载均衡服务if err := cc.dopts.balancer.Start(target, config); err != nil {waitC <- errreturn}ch := cc.dopts.balancer.Notify()if ch == nil {// There is no name resolver installed.addrs = append(addrs, Address{Addr: target})} else {addrs, ok = <-chif !ok || len(addrs) == 0 {waitC <- errNoAddrreturn}}}for _, a := range addrs {//给每个地址一个conn，连接池if err := cc.resetAddrConn(a, false, nil); err != nil {waitC <- errreturn}}close(waitC)}()var timeoutCh <-chan time.Timeif cc.dopts.timeout > 0 {timeoutCh = time.After(cc.dopts.timeout)}select {case <-ctx.Done():return nil, ctx.Err()case err := <-waitC:if err != nil {return nil, err}case <-timeoutCh:return nil, ErrClientConnTimeout}// If balancer is nil or balancer.Notify() is nil, ok will be false here.// The lbWatcher goroutine will not be created.if ok {go cc.lbWatcher()}return cc, nil
}

func (c *greeterClient) SayHello(ctx context.Context, in *HelloRequest, opts ...grpc.CallOption) (*HelloReply, error) {out := new(HelloReply)//调用实际的发送请求函数err := grpc.Invoke(ctx, "/helloworld.Greeter/SayHello", in, out, c.cc, opts...)if err != nil {return nil, err}return out, nil
}//最后主要是invoke函数
func invoke(ctx context.Context, method string, args, reply interface{}, cc *ClientConn, opts ...CallOption) (e error) {c := defaultCallInfofor _, o := range opts {//调用之前的hookif err := o.before(&c); err != nil {return toRPCErr(err)}}defer func() {for _, o := range opts {//执行完后的hooko.after(&c)}}()//trace相关代码if EnableTracing {c.traceInfo.tr = trace.New("grpc.Sent."+methodFamily(method), method)defer c.traceInfo.tr.Finish()c.traceInfo.firstLine.client = trueif deadline, ok := ctx.Deadline(); ok {c.traceInfo.firstLine.deadline = deadline.Sub(time.Now())}c.traceInfo.tr.LazyLog(&c.traceInfo.firstLine, false)// TODO(dsymonds): Arrange for c.traceInfo.firstLine.remoteAddr to be set.defer func() {if e != nil {c.traceInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{e}}, true)c.traceInfo.tr.SetError()}}()}//统计相关代码if stats.On() {ctx = stats.TagRPC(ctx, &stats.RPCTagInfo{FullMethodName: method})begin := &stats.Begin{Client:    true,BeginTime: time.Now(),FailFast:  c.failFast,}stats.HandleRPC(ctx, begin)}defer func() {//结束后的统计相关代码if stats.On() {end := &stats.End{Client:  true,EndTime: time.Now(),Error:   e,}stats.HandleRPC(ctx, end)}}()topts := &transport.Options{Last:  true,Delay: false,}for {var (err    errort      transport.ClientTransportstream *transport.Stream// Record the put handler from Balancer.Get(...). It is called once the// RPC has completed or failed.put func())// TODO(zhaoq): Need a formal spec of fail-fast.//传输层的配置callHdr := &transport.CallHdr{Host:   cc.authority,Method: method,}if cc.dopts.cp != nil {callHdr.SendCompress = cc.dopts.cp.Type()}gopts := BalancerGetOptions{BlockingWait: !c.failFast,}//得到传输成连接，在http2中一个传输单位是一个流。t, put, err = cc.getTransport(ctx, gopts)if err != nil {// TODO(zhaoq): Probably revisit the error handling.if _, ok := err.(*rpcError); ok {return err}if err == errConnClosing || err == errConnUnavailable {if c.failFast {return Errorf(codes.Unavailable, "%v", err)}continue}// All the other errors are treated as Internal errors.return Errorf(codes.Internal, "%v", err)}if c.traceInfo.tr != nil {c.traceInfo.tr.LazyLog(&payload{sent: true, msg: args}, true)}// 发送请求stream, err = sendRequest(ctx, cc.dopts.codec, cc.dopts.cp, callHdr, t, args, topts)if err != nil {if put != nil {put()put = nil}// Retry a non-failfast RPC when// i) there is a connection error; or// ii) the server started to drain before this RPC was initiated.// 在这两种情况下重试，1 链接错误 2 在rpc初始化之前服务端已经开始服务if _, ok := err.(transport.ConnectionError); ok || err == transport.ErrStreamDrain {if c.failFast {return toRPCErr(err)}continue}return toRPCErr(err)}//收消息err = recvResponse(ctx, cc.dopts, t, &c, stream, reply)if err != nil {if put != nil {put()put = nil}if _, ok := err.(transport.ConnectionError); ok || err == transport.ErrStreamDrain {if c.failFast {return toRPCErr(err)}continue}return toRPCErr(err)}if c.traceInfo.tr != nil {c.traceInfo.tr.LazyLog(&payload{sent: false, msg: reply}, true)}//关闭一个http2流t.CloseStream(stream, nil)if put != nil {put()put = nil}//Errorf会判断返回十分okreturn Errorf(stream.StatusCode(), "%s", stream.StatusDesc())}
}

func sendRequest(ctx context.Context, codec Codec, compressor Compressor, callHdr *transport.CallHdr, t transport.ClientTransport, args interface{}, opts *transport.Options) (_ *transport.Stream, err error) {// 创建一个http2流stream, err := t.NewStream(ctx, callHdr)if err != nil {return nil, err}defer func() {if err != nil {// If err is connection error, t will be closed, no need to close stream here.if _, ok := err.(transport.ConnectionError); !ok {t.CloseStream(stream, err)}}}()var (cbuf       *bytes.BufferoutPayload *stats.OutPayload)//压缩不为空if compressor != nil {cbuf = new(bytes.Buffer)}//统计if stats.On() {outPayload = &stats.OutPayload{Client: true,}}//编码并压缩数据outBuf, err := encode(codec, args, compressor, cbuf, outPayload)if err != nil {return nil, Errorf(codes.Internal, "grpc: %v", err)}//写入流err = t.Write(stream, outBuf, opts)if err == nil && outPayload != nil {outPayload.SentTime = time.Now()stats.HandleRPC(ctx, outPayload)}// t.NewStream(...) could lead to an early rejection of the RPC (e.g., the service/method// does not exist.) so that t.Write could get io.EOF from wait(...). Leave the following// recvResponse to get the final status.if err != nil && err != io.EOF {return nil, err}// Sent successfully.return stream, nil
}

func recvResponse(ctx context.Context, dopts dialOptions, t transport.ClientTransport, c *callInfo, stream *transport.Stream, reply interface{}) (err error) {// Try to acquire header metadata from the server if there is any.defer func() {if err != nil {if _, ok := err.(transport.ConnectionError); !ok {t.CloseStream(stream, err)}}}()c.headerMD, err = stream.Header()if err != nil {return}p := &parser{r: stream}var inPayload *stats.InPayloadif stats.On() {inPayload = &stats.InPayload{Client: true,}}for {//一直读到流关闭if err = recv(p, dopts.codec, stream, dopts.dc, reply, math.MaxInt32, inPayload); err != nil {if err == io.EOF {break}return}}if inPayload != nil && err == io.EOF && stream.StatusCode() == codes.OK {// TODO in the current implementation, inTrailer may be handled before inPayload in some cases.// Fix the order if necessary.stats.HandleRPC(ctx, inPayload)}c.trailerMD = stream.Trailer()return nil
}func recv(p *parser, c Codec, s *transport.Stream, dc Decompressor, m interface{}, maxMsgSize int, inPayload *stats.InPayload) error {//接受数据pf, d, err := p.recvMsg(maxMsgSize)if err != nil {return err}if inPayload != nil {inPayload.WireLength = len(d)}if err := checkRecvPayload(pf, s.RecvCompress(), dc); err != nil {return err}if pf == compressionMade {//解压d, err = dc.Do(bytes.NewReader(d))if err != nil {return Errorf(codes.Internal, "grpc: failed to decompress the received message %v", err)}}if len(d) > maxMsgSize {// TODO: Revisit the error code. Currently keep it consistent with java// implementation.return Errorf(codes.Internal, "grpc: received a message of %d bytes exceeding %d limit", len(d), maxMsgSize)}//数据解码if err := c.Unmarshal(d, m); err != nil {return Errorf(codes.Internal, "grpc: failed to unmarshal the received message %v", err)}if inPayload != nil {inPayload.RecvTime = time.Now()inPayload.Payload = m// TODO truncate large payload.inPayload.Data = dinPayload.Length = len(d)}return nil
}