go grpc流式和非流式的例子

參考grpc官方：  https://grpc.io/docs/quickstart/go.html

或官方中文翻譯： http://doc.oschina.net/grpc?t=60133

 

安裝proto buf 3、protoc編譯器

1-安裝grpc(需要紅杏出牆)： 

# 因為本朝網絡的安全原因，需要提前export http_proxy https_proxy 來紅杏出牆到golang.org。 比如執行linux命令 export http{,s}_proxy=紅杏出牆IP:紅杏出牆端口號
執行：go get -u google.golang.org/grpc

2-安裝

  下載protoc：

           從 https://github.com/google/protobuf/releases下載預編譯的“protoc編譯器”，用於生成gRPC服務代碼。（文件名為：protoc-<版本>-<平台>.zip）

          下載后，解壓zip文件，並將protoc二進制文件所在bin目錄添加到PATH環境變量中

   安裝protoc的go插件：

go get -u github.com/golang/protobuf/protoc-gen-go

  go get -u github.com/golang/protobuf  //解決go build報錯：” undefined: proto.ProtoPackageIsVersion3 “ 。高版本protoc搭配低版本protobuf就會出現這樣的問題

        將$GOPATH/bin加入到PATH環境變量中

 

非流式RPC: 

1-使用proto buf 3的IDL(接口定義語言)語法編寫 .proto 文件.  關於.proto文件的語法規則參考本博protocol buffer的整理

我的.proto文件路徑是：$GOPATH/src/nonstream/non_stream.proto  

syntax = "proto3"; //必須指定，否則就認為是proto2版本

package nonstream;  //包名，生成go代碼的時候，protoc會用到這個包名,生成的.pb.go文件中有package nonstream，后面的服務端、客戶端go源文件時需要導入該包名

//定義服務 ( 風格：服務名和rpc方法名是大駝峰風格 )
service EchoIface {
     //自己的GO源碼的類需要實現接口Echo函數 
     rpc Echo (Ping) returns (Pong) {} 
}

// 定義消息 （風格： 消息名用大駝峰風格）
message Ping {
  string request = 1;   //(風格： 字段名用小寫下划線風格，如果是枚舉字段名則是大寫下划線風格)
}

message Pong {
  string reply = 1;
} 

2- 使用protoc生成go代碼：

protoc命令的語法： protoc -I /PATH/TO/.proto_DIR/   /PATH/TO/TARGET.proto --go_out=plugins=grpc:/PATH/TO/OUTPUT_DIR  #不指定-I表示從當前工作目錄尋找.proto文件。-I一般是指向項目根目錄。
#本例編譯.proto文件使用下面的命令(需要cd到.proto文件所在目錄，再執行)
protoc non_stream.proto --go_out=plugins=grpc:.  #執行完這條命令后，就可以在當前目錄下看到non_stream.pb.go源碼文件了。 命令中:后面的.表示當前目錄，加入要生成代碼到/tmp目錄，則使用選項--go-out=plugins=grpc:/tmp

# 題外話： --go-out 選項中有無plugins=grpc的區別：
1 protoc --go_out=. ./helloworld.proto                   // -–go_out=后直接跟路徑(點號表示當前路徑)。只生成序列化/反序列化代碼文件，不需要rpc通訊。
protoc --go_out=plugins=grpc:.  ./helloworld.proto     // --go_out=后跟了grpc插件和目錄。生成序列化反序列化代碼 和 客戶端/服務端通訊代碼。

 

3-編寫非流式服務端go代碼：

package main

import (
    "context"
    "log"
    "net"
    "strings"

    pb "nonstream"  //剛才.proto文件定義的包名
    "google.golang.org/grpc"
)

type server struct{}

/*server結構體需要實現EchoIface接口的Echo方法, 注意這里的參數多了context，返回值多了error。這可以查看protoc生成的.pb.go文件就發現有下面的定義：
type EchoIfaceServer interface {
    Echo(context.Context, *Ping) (*Pong, error)
}
*/
func (s *server) Echo(ctx context.Context, in *pb.Ping) (*pb.Pong, error) {
    log.Printf("Received from client: %s", in.Request)
    u := strings.ToUpper(in.Request)  //注意.proto文件中字段是小寫下划線，這里變成了大駝峰了
    return &pb.Pong{Reply: u + "..." + in.Request + "..."}, nil
}

func main() {
    lis, err := net.Listen("tcp", ":5555") //1.指定監聽地址:端口號
    if err != nil {
        log.Fatalf("failed to listen: %v", err)
    }

    s := grpc.NewServer() //2.新建gRPC實例
    pb.RegisterEchoIfaceServer(s, &server{})  //3.在gRPC服務器注冊我們的服務實現。參數2是接口(滿足服務定義的方法)。在.pb.go文件中搜索Register關鍵字即可找到這個函數簽名
    if err := s.Serve(lis); err != nil {   //4.Serve()阻塞等待
        log.Fatalf("failed to serve: %v", err)
    }
}

 

4-編寫非流式客戶端go代碼：

 

package main

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

    pb "nonstream"
    "google.golang.org/grpc"
)


func main() {
    //1.建立連接
    conn, err := grpc.Dial("127.0.0.1:5555", grpc.WithInsecure())  //如果需要授權認證或tls加密，則可以使用DialOptions來設置grpc.Dial
    if err != nil {
        log.Fatalf("grpc.Dial: %v", err)
    }
    defer conn.Close()

    c := pb.NewEchoIfaceClient(conn) //2.新建一個客戶端stub來執行rpc方法

    ctx, cancel := context.WithTimeout(context.Background(), time.Second)  //超時1秒執行cancel
    defer cancel()

    r, err := c.Echo(ctx, &pb.Ping{Request: "yahoo"})  //3.調用rpc方法
    if err != nil {
        log.Fatalf("c.Echo: %v", err)
    }

    fmt.Printf("echo from server: %s\n", r.Reply)
}

 

 

---------------------------------- 分割線 ------------------------------

單向流式RPC -- 服務流式響應

流式的好處之一是可以實現異步

1- 編寫.proto文件：

// 這個文件的路徑: $GOPATH/src/stream/stream.proto 。包名是stream,后面的服務端/客戶端代碼會導入這個包
syntax = "proto3";

package stream;

service EchoIface {
      rpc Echo(Ping) returns (stream Pong) {}  //server-side-stream
}

message Ping {
      string request = 1;
}

message Pong {
      string reply = 1;
}

2-使用protoc生成go代碼(此處省略，具體參考上面protoc的例子)

3-服務端代碼：

package main

import (
    "log"
    "net"
    "strconv"

    "google.golang.org/grpc"
    pb "stream"
)

type server struct{}

/* 從.pb.go文件中，我們發現了下面的定義：
      type EchoIfaceServer interface {
           Echo(*Ping, EchoIface_EchoServer) error
      }
   而參數EchoIface_EchoServer的定義為（有Send函數）：
      type EchoIface_EchoServer interface {
          Send(*Pong) error
          grpc.ServerStream
   }
*/
func (s *server) Echo(in *pb.Ping, es pb.EchoIface_EchoServer) error {
    n := in.Request
    for i := 0; i < 10; i++ {  //發10次Hello
        es.Send(&pb.Pong{Reply: "Hello " + n + ":" + strconv.Itoa(i)})
    }
    return nil
}

func main() {
    conn, err := net.Listen("tcp", ":6666")
    if err != nil {
        log.Fatalf("net.Listen: %v", err)
    }
    svr := grpc.NewServer()

    pb.RegisterEchoIfaceServer(svr, &server{})
    if err := svr.Serve(conn); err != nil {
        log.Fatalf("s.Serve(): %v", err)
    }
}

4-客戶端代碼：

package main

import (
    "context"
    "io"
    "log"

    "google.golang.org/grpc"
    pb "stream"
)

func main() {
    conn, err := grpc.Dial("127.0.0.1:6666", grpc.WithInsecure())
    if err != nil {
        log.Fatalf("grpc.Dial: %v", err)
    }
    defer conn.Close()

    c := pb.NewEchoIfaceClient(conn)

    /* 從protoc生成的.pb.go文件中，我們發現如下定義：
               type EchoIfaceClient interface {
                      Echo(ctx context.Context, in *Ping, opts ...grpc.CallOption) (EchoIface_EchoClient, error)
               }
               而上面接口函數的第一個返回值在.pb.go文件中的的定義是：
                       type EchoIface_EchoClient interface {
                         Recv() (*Pong, error)
                        grpc.ClientStream
                   }
    */
    stream, err := c.Echo(context.Background(), &pb.Ping{Request: "google"})
    if err != nil {
        log.Fatalf("c.Echo: %v", err)
    }

    for {
        pong, err := stream.Recv()
        if err == io.EOF {
            break
        }

        if err != nil {
            log.Printf("stream.Recv: %v", err)
        }

        log.Printf("echo from server: %s", pong.Reply)
    }

}

 

雙向流式RPC，客戶端流式RPC

無非就是：

    1-編寫.proto文件的服務定義時，在需要變成流式的參數或返回值前加stream修飾。

    2-對.pb.go文件執行正則搜索，找到函數/方法的定義。再根據定義寫go代碼

 但是要注意一些差異(細節代碼參考)：

     1-客戶端流式RPC (客戶端一次或多次發數據到服務端，服務端響應一次)：

               客戶端： 使用 流 的Send()方法 發送請求，發送完后使用 流 的CloseAndRecv方法讓gRPC服務端知道客戶端已經完成請求並且期望獲得一個響應。如果CloseAndRecv()返回的err不為nil，那么返回的第一個值就是一個有效的服務端響應。

              服務端：  使用 流 的Recv() 方法接收客戶端消息，並且用 流 的SendAndClose() 方法返回它的單個響應。

     2-雙向流式RPC中（客戶端：等待接收，並同時發送一次或多次數據到服務端，最后一次發送結束標識--CloseSend()方法。 服務端：來一個，處理一個，響應一個，不用等待客戶端發送完成才處理。）：

             客戶端：一個goroutine用來執行 流的 Recv()方法 等待服務端發來的流式響應(阻塞等待狀態)。另一個goroutine用來將一次或多次請求通過流的Send()方法發送到服務端，發完后，使用CloseSend()結束發送。

             服務端： 服務端用 流 的Recv()方法接收客戶端的請求流，接收一個、處理一個、發送一個響應，直到出錯，或因為客戶端CloseSend()導致的服務端接收完請求。

更多細節，結合源代碼https://github.com/grpc/grpc-go/blob/master/examples/route_guide和官方文檔中譯版看

 

不需要入參結構體或返回結構體的情況:

方法1:  可以在.proto文件中定義空的消息(也就是空結構體,這樣做的好處是以后可以給結構體添加字段, 接口函數簽名不變).          

message MyRespone {
}

務必注意,在源代碼編寫的時候,不能直接直接返回空指針, 否則會報"rpc error: code = Internal desc = grpc: error while marshaling: proto: Marshal called with nil".  意思大概就是不能序列化nil指針

if err != nil {
       return nil, err  //錯誤! 正確寫法 return &pb.MyRespone{},err
}

方法2: 使用protobuf的內置空類型.

.proto文件需要類似這樣定義(以上面非流式的例子做修改成不需要返回結構體為例子)

 syntax = "proto3"; 
 import "google/protobuf/empty.proto";  //這個就是protobuf內置的空類型
 package nonstream; 
 
 service EchoIface {
      rpc Echo (Ping) returns (google.protobuf.Empty) {} 
 }
 
 message Ping {
   string request = 1;
 }

源碼文件的編寫需要這樣(以上面非流式的例子做修改):

import "github.com/golang/protobuf/ptypes/empty"
 ...
 func (s *server) Echo(ctx context.Context, in *pb.Ping) (*empty.Empty, error)