grpc 簡(jiǎn)單使用

參考：golang grpc 快速開(kāi)始
以 windows 平臺(tái)為例：

go get google.golang.org/protobuf/cmd/protoc-gen-go 
 go get  google.golang.org/grpc/cmd/protoc-gen-go-grpc
因?yàn)槲耶?dāng)前 的 gopath 就是f:/go 所以 直接把get 包安裝到了 bin 目錄

gopath/bin 要加入到環(huán)境變量

然后就可以試一下 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控制臺(tái)輸出：
> Greeting: Hello world
  • 修改 rpc 服務(wù)的函數(shù)
    在 helloworld/helloworld.proto 添加SayHelloAgain()具有相同請(qǐng)
    求和響應(yīng)類(lèi)型的新方法，改成如下文件
// 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文件
我們 上面新添加了 一個(gè) rpc 函數(shù) ： 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 函數(shù)末尾添加 ：

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())
  • 分別運(yùn)行 greeter_server/main.go 和 greeter_client/main.go

更全面的官方教程在此, 包含普通，客戶(hù)端流，服務(wù)端流，雙向流

簡(jiǎn)單介紹一下四種模式下的基本用法(既然看了，還是要記錄一下的)

服務(wù)端：

  • 普通的調(diào)用
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
}
  • 服務(wù)端流 ( stream.Send 不斷發(fā)響應(yīng))
// 服務(wù)端流 
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
}
  • 客戶(hù)端流 ( stream.Recv() 不斷取值， 遇到 io.EOF 的錯(cuò)誤，就調(diào)用 stream.SendAndClose 返回值，并通知客戶(hù)端函數(shù)執(zhí)行完畢)
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) // 計(jì)算兩個(gè) point之間的距離
        }
        lastPoint = point
    }
}
  • 雙向流 ( stream 入?yún)⒓瓤梢訰ecv 也可以 Send ，每一個(gè) recv 都會(huì)循環(huán)發(fā)送 send ,這樣就模擬了雙向的效果)
// 為了模擬流發(fā)送，他在每一個(gè)請(qǐng)求收到后，循環(huán)發(fā)送所有的對(duì)應(yīng)
// 的值， 并且枷鎖，防止并發(fā)共享一個(gè)屬性導(dǎo)致出現(xiàn)問(wèn)題
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
            }
        }
    }
}

客戶(hù)端
其實(shí)和服務(wù)端基本邏輯一致

  • 調(diào)用服務(wù)流：
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())
    }
}

就是一個(gè)簡(jiǎn)單的 for {stream.Recv()}

下面看一個(gè)簡(jiǎn)單的，grpc 里究竟怎么實(shí)現(xiàn) rpc 協(xié)議的：

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)
}

這里就類(lèi)似 rpc 的風(fēng)格了。(呀復(fù)習(xí)一下， go rpc 的基本寫(xiě)法了)

type HelloService struct {}

func (p *HelloService) Hello(request string, reply *string) error {
    *reply = "hello:" + request
    return nil
}
其中Hello方法必須滿(mǎn)足Go語(yǔ)言的RPC規(guī)則：方法只能有兩個(gè)可序
列化的參數(shù)，其中第二個(gè)參數(shù)是指針類(lèi)型，并且返回一個(gè)error類(lèi)
型，同時(shí)必須是公開(kāi)的方法。

grpc 官方文檔: 生成代碼結(jié)構(gòu)的講解

具體講講生成的 代碼（有一些名稱(chēng)是固定的格式）

服務(wù)端：

對(duì)于 service Bar{ rpc Foo( client_data) returns (server_data) ; } 為例子 ( 流的話(huà)加上關(guān)鍵字 stream 修飾參數(shù)即可 )
注冊(cè)函數(shù) RegisterBarServer 格式為 Register<service name>Server (當(dāng)然你也可以自己寫(xiě)注冊(cè)的函數(shù))

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

普通一元函數(shù) ( 以 Foo 為例子)

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

服務(wù)端流：

Foo(*MsgA, <ServiceName>_FooServer) error

// MsgA   接收到的消息
//  <ServiceName>_FooServer  流接口類(lèi)型   
// <Services name>_<rpc_func_name>Server

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

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

客戶(hù)端流

Foo(<ServiceName>_FooServer) error

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

// 一直 Recv ，最后結(jié)束的時(shí)候發(fā)送一次消息 調(diào)用 SendAndClose  結(jié)束本次響應(yīng)

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

雙流

Foo(<ServiceName>_FooServer) error

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


//同時(shí)首發(fā)數(shù)據(jù)  Send  和 Recv

客戶(hù)端接口

  • 一元方法：
(ctx context.Context, in *MsgA, opts ...grpc.CallOption) (*MsgB, error)
  • 服務(wù)流
Foo(ctx context.Context, in *MsgA, opts ...grpc.CallOption) (<ServiceName>_FooClient, error)
// <ServiceName>_FooClient  代表 服務(wù)端的 流對(duì)象 

type <ServiceName>_FooClient interface {
    Recv() (*MsgB, error)
    grpc.ClientStream
}
  • 客戶(hù)端流
Foo(ctx context.Context, opts ...grpc.CallOption) (<ServiceName>_FooClient, error)

// <ServiceName>_FooClient代表客戶(hù)機(jī)到服務(wù)器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
}

暫時(shí)完畢。。

只有 客戶(hù)端流，才有 CloseAndRecv (客戶(hù)端) 和 SendAndClose （ 服務(wù)端只有這個(gè)，沒(méi)有 send）函數(shù)，其他的沒(méi)有這兩個(gè)函數(shù)

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡(jiǎn)書(shū)系信息發(fā)布平臺(tái)，僅提供信息存儲(chǔ)服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容