grpc 简单使用

参考：golang grpc 快速开始
以 windows 平台为例：

• 下载 安装 go —— 略

• 下载 protoc
  参考：https://grpc.io/docs/protoc-installation/
  win平台直接到 github 下载 zip 防盗本地目录解压即可
  下载 地址 https://github.com/protocolbuffers/protobuf/releases
  把 解压后的 bin 目录加到环境变量，以便控制台找到 protoc 命令
  

  

  把这个bin 目录加到环境变量

• 安装 protoc-gen-go 和 protoc-gen-go-grpc 包

go get google.golang.org/protobuf/cmd/protoc-gen-go 
 go get  google.golang.org/grpc/cmd/protoc-gen-go-grpc

因为我当前 的 gopath 就是f:/go 所以 直接把get 包安装到了 bin 目录

gopath/bin 要加入到环境变量

然后就可以试一下 grpc 了

• 直接用 官方的例子

git clone -b v1.35.0 https://github.com/grpc/grpc-go

cd grpc-go/examples/helloworld
go run greeter_server/main.go
go run greeter_client/main.go
# greeter_client控制台输出：
> Greeting: Hello world

• 修改 rpc 服务的函数
  在 helloworld/helloworld.proto 添加SayHelloAgain()具有相同请
  求和响应类型的新方法，改成如下文件

// The greeting service definition.
service Greeter {
  // Sends a greeting
  rpc SayHello (HelloRequest) returns (HelloReply) {}
  // Sends another greeting
  rpc SayHelloAgain (HelloRequest) returns (HelloReply) {}
}

// The request message containing the user's name.
message HelloRequest {
  string name = 1;
}

// The response message containing the greetings
message HelloReply {
  string message = 1;
}

• 重新编译更新的 .proto 文件

protoc --go_out=. --go_opt=paths=source_relative \
    --go-grpc_out=. --go-grpc_opt=paths=source_relative \
    helloworld/helloworld.proto

image.png

重新生成 了 helloworld/helloworld.pb.go和helloworld/helloworld_grpc.pb.go文件
我们 上面新添加了 一个 rpc 函数 ： SayHelloAgain

• 修改 greeter_server/main.go
  添加下面的代码

func (s *server) SayHelloAgain(ctx context.Context, in 
*pb.HelloRequest) (*pb.HelloReply, error) {
        return &pb.HelloReply{Message: "Hello again " + in.GetName()}, nil
}

• 修改 greeter_client/main.go

main 函数末尾添加 ：

r, err = c.SayHelloAgain(ctx, &pb.HelloRequest{Name: name})
if err != nil {
        log.Fatalf("could not greet: %v", err)
}
log.Printf("Greeting: %s", r.GetMessage())

• 分别运行 greeter_server/main.go 和 greeter_client/main.go

更全面的官方教程在此, 包含普通，客户端流，服务端流，双向流

简单介绍一下四种模式下的基本用法(既然看了，还是要记录一下的)

服务端：

• 普通的调用

func (s *routeGuideServer) GetFeature(ctx context.Context, point 
*pb.Point) (*pb.Feature, error) {
    for _, feature := range s.savedFeatures {
        if proto.Equal(feature.Location, point) {
            return feature, nil
        }
    }
    // No feature was found, return an unnamed feature
    return &pb.Feature{Location: point}, nil
}

• 服务端流 ( stream.Send 不断发响应)

// 服务端流 
func (s *routeGuideServer) ListFeatures(rect *pb.Rectangle, 
stream pb.RouteGuide_ListFeaturesServer) error {
    for _, feature := range s.savedFeatures {
        if inRange(feature.Location, rect) {
            if err := stream.Send(feature); err != nil {
                return err
            }
        }
    }
    return nil
}

• 客户端流 ( stream.Recv() 不断取值， 遇到 io.EOF 的错误，就调用 stream.SendAndClose 返回值，并通知客户端函数执行完毕)

func (s *routeGuideServer) RecordRoute(stream pb.RouteGuide_RecordRouteServer) error {
    var pointCount, featureCount, distance int32
    var lastPoint *pb.Point
    startTime := time.Now()
    for {
        point, err := stream.Recv()
        if err == io.EOF {
            endTime := time.Now()
            return stream.SendAndClose(&pb.RouteSummary{
                PointCount:   pointCount,
                FeatureCount: featureCount,
                Distance:     distance,
                ElapsedTime:  int32(endTime.Sub(startTime).Seconds()),
            })
        }
        if err != nil {
            return err
        }
        pointCount++
        for _, feature := range s.savedFeatures {
            if proto.Equal(feature.Location, point) {
                featureCount++
            }
        }
        if lastPoint != nil {
            distance += calcDistance(lastPoint, point) // 计算两个 point之间的距离
        }
        lastPoint = point
    }
}

• 双向流 ( stream 入参既可以Recv 也可以 Send ，每一个 recv 都会循环发送 send ,这样就模拟了双向的效果)

// 为了模拟流发送，他在每一个请求收到后，循环发送所有的对应
// 的值， 并且枷锁，防止并发共享一个属性导致出现问题
func (s *routeGuideServer) RouteChat(stream pb.RouteGuide_RouteChatServer) error {
    for {
        in, err := stream.Recv()
        if err == io.EOF {
            return nil
        }
        if err != nil {
            return err
        }
        key := serialize(in.Location)

        s.mu.Lock()
        s.routeNotes[key] = append(s.routeNotes[key], in)
        // Note: this copy prevents blocking other clients while serving this one.
        // We don't need to do a deep copy, because elements in the slice are
        // insert-only and never modified.
        rn := make([]*pb.RouteNote, len(s.routeNotes[key]))
        copy(rn, s.routeNotes[key])
        s.mu.Unlock()

        for _, note := range rn {
            if err := stream.Send(note); err != nil {
                return err
            }
        }
    }
}

客户端
其实和服务端基本逻辑一致

• 调用服务流：

func printFeatures(client pb.RouteGuideClient, rect *pb.Rectangle) {
    log.Printf("Looking for features within %v", rect)
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()
    stream, err := client.ListFeatures(ctx, rect)
    if err != nil {
        log.Fatalf("%v.ListFeatures(_) = _, %v", client, err)
    }
    for {
        feature, err := stream.Recv()
        if err == io.EOF {
            break
        }
        if err != nil {
            log.Fatalf("%v.ListFeatures(_) = _, %v", client, err)
        }
        log.Printf("Feature: name: %q, point:(%v, %v)", feature.GetName(),
            feature.GetLocation().GetLatitude(), feature.GetLocation().GetLongitude())
    }
}

就是一个简单的 for {stream.Recv()}

下面看一个简单的，grpc 里究竟怎么实现 rpc 协议的：

func (c *routeGuideClient) GetFeature(ctx context.Context, in *Point, opts ...grpc.CallOption) (*Feature, error) {
    out := new(Feature)
    err := c.cc.Invoke(ctx, "/routeguide.RouteGuide/GetFeature", in, out, opts...)
    if err != nil {
        return nil, err
    }
    return out, nil
}

Invoke 所在接口：

type ClientConnInterface interface {
    // Invoke performs a unary RPC and returns after the response is received
    // into reply.
    Invoke(ctx context.Context, method string, args interface{}, reply interface{}, opts ...CallOption) error
    // NewStream begins a streaming RPC.
    NewStream(ctx context.Context, desc *StreamDesc, method string, opts ...CallOption) (ClientStream, error)
}

这里就类似 rpc 的风格了。(呀复习一下， go rpc 的基本写法了)

type HelloService struct {}

func (p *HelloService) Hello(request string, reply *string) error {
    *reply = "hello:" + request
    return nil
}
其中Hello方法必须满足Go语言的RPC规则：方法只能有两个可序
列化的参数，其中第二个参数是指针类型，并且返回一个error类
型，同时必须是公开的方法。

grpc 官方文档: 生成代码结构的讲解

具体讲讲生成的 代码（有一些名称是固定的格式）

服务端：

对于 service Bar{ rpc Foo( client_data) returns (server_data) ; } 为例子 ( 流的话加上关键字 stream 修饰参数即可 )
注册函数 RegisterBarServer 格式为 Register<service name>Server (当然你也可以自己写注册的函数)

func RegisterBarServer(s *grpc.Server, srv BarServer)

普通一元函数 ( 以 Foo 为例子)

Foo(context.Context, *MsgA) (*MsgB, error)
// MsgA 接受的消息， MsgB 发送的消息

服务端流：

Foo(*MsgA, <ServiceName>_FooServer) error

// MsgA   接收到的消息
//  <ServiceName>_FooServer  流接口类型   
// <Services name>_<rpc_func_name>Server

<ServiceName>_FooServer 接口定义如下 （ 所以可以使用 Send 方法了）：

type <ServiceName>_FooServer interface {
    Send(*MsgB) error
    grpc.ServerStream
}

客户端流

Foo(<ServiceName>_FooServer) error

type <ServiceName>_FooServer interface {
    SendAndClose(*MsgA) error
    Recv() (*MsgB, error)
    grpc.ServerStream
}

// 一直 Recv ，最后结束的时候发送一次消息 调用 SendAndClose  结束本次响应

// 流消息结束， Recv返回(nil, io.EOF)  

双流

Foo(<ServiceName>_FooServer) error

type <ServiceName>_FooServer interface {
    Send(*MsgA) error
    Recv() (*MsgB, error)
    grpc.ServerStream
}


//同时首发数据  Send  和 Recv

客户端接口

• 一元方法：

(ctx context.Context, in *MsgA, opts ...grpc.CallOption) (*MsgB, error)

• 服务流

Foo(ctx context.Context, in *MsgA, opts ...grpc.CallOption) (<ServiceName>_FooClient, error)
// <ServiceName>_FooClient  代表 服务端的 流对象 

type <ServiceName>_FooClient interface {
    Recv() (*MsgB, error)
    grpc.ClientStream
}

• 客户端流

Foo(ctx context.Context, opts ...grpc.CallOption) (<ServiceName>_FooClient, error)

// <ServiceName>_FooClient代表客户机到服务器stream的MsgA
type <ServiceName>_FooClient interface {
    Send(*MsgA) error
    CloseAndRecv() (*MsgB, error)
    grpc.ClientStream
}

• 双向流

type <ServiceName>_FooClient interface {
    Send(*MsgA) error
    Recv() (*MsgB, error)
    grpc.ClientStream
}

暂时完毕。。

只有 客户端流，才有 CloseAndRecv (客户端) 和 SendAndClose （ 服务端只有这个，没有 send）函数，其他的没有这两个函数

有疑问加站长微信联系（非本文作者）