用Golang开始gRPC

2 年 ago
文, 翔
6 minutes

简而言之

gRPC是一个现代化且高性能的开源RPC（远程过程调用）框架，可在任何环境中运行。它使用Protocol Buffers来序列化数据，从而实现快速通信。它能够适应各种语言和平台，并支持基于http/2的双向流传输。使用Protocol Buffers可以简单地定义服务（用于通信的数据和函数），并可以明确定义API规范。

※本章内容参考自gRPC。

Git：k-washi/example-golang-gRPC

开始项目

VSCode的设置

    vscode-proto3 & Clang-Format extensions をインストール

如果你想要自动格式化文件，你还需要设置以下内容。

    Clang-Formatをインストール
brew install clang-format #MacOSX

#other OS  
#http://www.codepool.biz/vscode-format-c-code-windows-linux.html

Protocol buffers设置

安装Protocol Buffers

# MacOSX
brew install protobuf #MacOSX

# Linux
#https://github.com/google/protobuf/releases

#example
# Make sure you grab the latest version
curl -OL https://github.com/google/protobuf/releases/download/v3.5.1/protoc-3.5.1-linux-x86_64.zip
# Unzip
unzip protoc-3.5.1-linux-x86_64.zip -d protoc3
# Move protoc to /usr/local/bin/
sudo mv protoc3/bin/* /usr/local/bin/
# Move protoc3/include to /usr/local/include/
sudo mv protoc3/include/* /usr/local/include/
# Optional: change owner
sudo chown [user] /usr/local/bin/protoc
sudo chown -R [user] /usr/local/include/google

Golang的配置

安装软件包。

go get -u google.golang.org/grpc
go get -d -u github.com/golang/protobuf/protoc-gen-go

确认执行

首先，定义服务。

syntax = "proto3";

package greet;
option go_package="greetpb";

service GreetService{}

通过以下命令，将此定义文件转换为适用于golang的协议缓冲区（greet.pb.go）。通过在golang的客户端和服务器中使用此转换后的库文件（greet.pb.go）中的函数，可以实现通过gRPC进行通信。

protoc greet/greetpb/greet.proto  --go_out=plugins=grpc:.

客户端通过以下方式建立连接并使用客户端API（NewGreetServiceClient）进行通信。

package main

import (
    "fmt"
    "log"
  //protocol buffersライブラリー
    "github.com/k-washi/example-golang-gRPC/greet/greetpb"
    "google.golang.org/grpc"
)

func main() {
    fmt.Println("Hello. I'm a client")
    conn, err := grpc.Dial("localhost:50051", grpc.WithInsecure())
    if err != nil {
        log.Fatalf("Could not connect: %v", err)
    }
    defer conn.Close()
    c := greetpb.NewGreetServiceClient(conn)
    fmt.Printf("Created client %f", c)
}

服务器定义了一个server结构体，将服务注册到服务器上，然后启动服务器。

package main

import (
    "fmt"
    "log"
    "net"
    "github.com/k-washi/example-golang-gRPC/greet/greetpb"
    "google.golang.org/grpc"
)

type server struct{}

func main() {
    lis, err := net.Listen("tcp", ":50051")
    if err != nil {
        log.Fatalf("Failed to listen: %v", err)
    }

    s := grpc.NewServer()
    greetpb.RegisterGreetServiceServer(s, &server{})

    if err := s.Serve(lis); err != nil {
        log.Fatalf("failed to serve: %v", err)
    }
}

通过运行以下命令，启动服务器并执行客户端以进行确认。


#server setup
go run greet/greet_server/server.go 

#client setup
go run greet/greet_client/client.go

通信方式

gRPC提供多种通信方式，根据数据特性需要灵活选择。

    • Unary – もっとも一般的なgRPCのAPIで、クライアントからのメッセージ1つに対して、サーバーは1回のメッセージを返す

 

    • Server streaming – HTTP/2の恩恵を受けたAPIで、クライアントからのメッセージ1つに対して、サーバーは複数回のメッセージを返す

 

    Client streaming – HTTP/2の恩恵を受けたAPIで、クライアントからの複数のメッセージに対して、サーバーは1回のメッセージを返す

服务的定义

    1. 定义发送消息的类型

 

    1. 定义一元、服务器流式传输、客户端流式传输的请求和响应消息的类型

 

    1. 将服务定义为 rpc 服务名称（流式客户端类型）returns（流式响应类型）{}。对于需要多次发送的服务消息，添加 stream。

 

    使用以下命令将其转换为协议缓冲区。
protoc streaming/greetpb/greet.proto  --go_out=plugins=grpc:.

syntax = "proto3";

package greet;
option go_package="greetpb";

//リクエストメッセージの型を定義
message Greeting {
  string first_name = 1;
  string last_name = 2;
}

// Unary リクエスト
message GreetRequest {
  Greeting greeting = 1;
}

// Unary レスポンス
message GreetResponse {
  string result = 1;
}

// Server streaming リクエスト
message GreetManyTimesRequest {
  Greeting greeting = 1;
}

// Server streaming レスポンス
message GreetManyTimesResponse {
  string result = 1;
}

// Client streaming リクエスト
message LongGreetRequest {
  Greeting greeting = 1;
}

// Client streaming レスポンス
message LongGreetResponse {
  string result = 1;
}

// サービスの定義
service GreetService{
  //Unary
  rpc Greet(GreetRequest) returns (GreetResponse) {};

  //Server Streaming
  rpc GreetManyTimes(GreetManyTimesRequest) returns (stream GreetManyTimesResponse) {};

  //Client Streaming
  rpc LongGreet(stream LongGreetRequest) returns (LongGreetResponse) {};
}

客户

使用Golang实现Unary、Server streaming和Client streaming的客户端。基本上，根据服务定义创建数据类型，发送请求，然后等待响应的形式。这些操作需要使用protocol buffers库中的包。如果需要等待服务器的多个消息，可以使用循环语法（for）进行接收并应用。如果需要向服务器发送多个消息，也可以使用循环语法，但基本上只是按照每条消息发送。有关gRPC的错误，可以使用”google.golang.org/grpc/status”库进行提取和定义。


package main

import (
    "context"
    "fmt"
    "io"
    "log"
    "time"

    "google.golang.org/grpc/codes"

    "github.com/k-washi/example-golang-gRPC/streaming/greetpb"
    "google.golang.org/grpc"
    "google.golang.org/grpc/status"
)

func main() {
    fmt.Println("Hello. I'm a client")
    conn, err := grpc.Dial("localhost:50051", grpc.WithInsecure())
    if err != nil {
        log.Fatalf("Could not connect: %v", err)
    }
    defer conn.Close()
    c := greetpb.NewGreetServiceClient(conn)

    doUnary(c)
    doServerStreaming(c)
    doClientStreaming(c)

}

func doUnary(c greetpb.GreetServiceClient) {
    fmt.Println("Starting to do a Unary RPC ...")
    req := &greetpb.GreetRequest{
        Greeting: &greetpb.Greeting{
            FirstName: "tanaka",
            LastName:  "tarou",
        },
    }

    res, err := c.Greet(context.Background(), req)

    if err != nil {
        respErr, ok := status.FromError(err)
        if ok {
            fmt.Println(respErr.Message())
            fmt.Println(respErr.Code())
            if respErr.Code() == codes.InvalidArgument {
                fmt.Println("We probably sent a empty string")

            } else if respErr.Code() == codes.DeadlineExceeded {
                fmt.Println("Timeout was hit! Deadline was exceeded")
            }
        } else {
            log.Printf("error while calling greet rpc: %v", err)
        }
    }
    fmt.Printf("Response from greet server %v", res)
}

func doServerStreaming(c greetpb.GreetServiceClient) {
    fmt.Println("Starting to do a Server Streaming RPC ...")
    req := &greetpb.GreetManyTimesRequest{
        Greeting: &greetpb.Greeting{
            FirstName: "tanaka",
            LastName:  "tarou",
        },
    }
    resStream, err := c.GreetManyTimes(context.Background(), req)
    if err != nil {
        log.Fatalf("error while calling GreetManyTImes RPC: %v", err)
    }
    for {
        msg, err := resStream.Recv()
        if err == io.EOF {
            break
        }
        if err != nil {
            log.Fatalf("error while reading stream: %v", err)
        }
        log.Printf("Response from GreetManyTImes: %v", msg.GetResult())
    }

}

func doClientStreaming(c greetpb.GreetServiceClient) {
    fmt.Println("Starting to do a Client Streaming RPC")

    requests := []*greetpb.LongGreetRequest{
        &greetpb.LongGreetRequest{
            Greeting: &greetpb.Greeting{
                FirstName: "one",
            },
        },
        &greetpb.LongGreetRequest{
            Greeting: &greetpb.Greeting{
                FirstName: "two",
            },
        },
        &greetpb.LongGreetRequest{
            Greeting: &greetpb.Greeting{
                FirstName: "three",
            },
        },
        &greetpb.LongGreetRequest{
            Greeting: &greetpb.Greeting{
                FirstName: "four",
            },
        },
    }

    stream, err := c.LongGreet(context.Background())
    if err != nil {
        log.Fatalf("error while calling LongGreet %v", err)
    }
    for _, req := range requests {
        fmt.Printf("Sending req: %v\n", req)
        stream.Send(req)
        time.Sleep(1000 * time.Millisecond)
    }

    res, err := stream.CloseAndRecv()
    if err != nil {
        log.Fatalf("error while receiving response from LongGreet: %v", err)
    }
    fmt.Printf("LongGreet Response: %v", res)

}

服务器

在 `server` 结构体中，以与 `GreetServiceServer` 在 protocol buffers 中定义的相同格式嵌入函数。
消息的发送和接收也使用了在 protocol buffers 中定义的消息。此外，还可以使用消息对应的 `Get` 函数来获取请求。

package main

import (
    "context"
    "fmt"
    "io"
    "log"
    "net"
    "strconv"
    "time"

    "google.golang.org/grpc/codes"
    "google.golang.org/grpc/status"

    "github.com/k-washi/example-golang-gRPC/streaming/greetpb"

    "google.golang.org/grpc"
)

//server greet.pb.go GreetServiceClient
type server struct{}

func (*server) Greet(ctx context.Context, req *greetpb.GreetRequest) (*greetpb.GreetResponse, error) {
    fmt.Printf("Greet func was invoked with %v", req)
    firstName := req.GetGreeting().GetFirstName()
    if firstName == "" {
        return nil, status.Errorf(
            codes.InvalidArgument,
            "Recived a empty string",
        )
    }

    if ctx.Err() == context.Canceled {
        return nil, status.Error(codes.Canceled, "the client canceld the request")
    }

    result := "Hello " + firstName
    //client config deadline
    /*
        res := &greetpb.GreetWithDeadlineResponse{
            Result: result,
        }
    */
    res := &greetpb.GreetResponse{
        Result: result,
    }
    return res, nil
}

func (*server) GreetManyTimes(req *greetpb.GreetManyTimesRequest, stream greetpb.GreetService_GreetManyTimesServer) error {
    fmt.Printf("GreetManyTimes func was invoked with %v", req)
    firstName := req.GetGreeting().GetFirstName()
    for i := 0; i < 10; i++ {
        result := "Hello" + firstName + " number " + strconv.Itoa(i)
        res := &greetpb.GreetManyTimesResponse{
            Result: result,
        }
        stream.Send(res)
        time.Sleep(1000 * time.Millisecond)
    }
    return nil
}

func (*server) LongGreet(stream greetpb.GreetService_LongGreetServer) error {
    fmt.Println("LongGreet function was invoked with a streaming request")
    result := ""
    for {
        req, err := stream.Recv()
        if err == io.EOF {
            //we have finished reading the client stream
            return stream.SendAndClose(&greetpb.LongGreetResponse{
                Result: result,
            })
        }
        if err != nil {
            log.Fatalf("Error while reading client stream: %v", err)
        }
        firstName := req.GetGreeting().GetFirstName()
        result += "Hello " + firstName + "! "

    }
}

func main() {
    fmt.Println("Hello World!")

    lis, err := net.Listen("tcp", "0.0.0.0:50051")
    if err != nil {
        log.Fatalf("Failed to listen: %v", err)
    }

    s := grpc.NewServer()
    greetpb.RegisterGreetServiceServer(s, &server{})

    if err := s.Serve(lis); err != nil {
        log.Fatalf("failed to serve: %v", err)
    }
}

文献引用

    • gRPCって何？

 

    • Goで始めるgRPC入門

 

    今から学ぶgRPCの基礎
✖
广告
将在 10 秒后关闭
bannerAds