GRPC通信模式 • Overstarry Site

04 Dec 2021

6 minutes reading time

在我们日常的客户端和服务端的通信中，往往是简单的请求-响应风格的通信，这里一个请求会得到一个响应。借助gRPC，我们可以实现不同的进程间通信模式。

接下来就由我来介绍 gRPC的4种通信模式：一元 RPC, 服务端流式 RPC, 客户端流式 RPC, 客户端流式 RPC（双向流式），并会带有一些简单的例子展示各种模式，在这里客户端和服务端都由 go 编写。

#一元 RPC

我们先来介绍一元 RPC，这是一种单向通信，客户端调用服务端的远程方法时，客户端发送请求至服务端并获得一个响应。

假设我们现在需要构建一个商城系统，商城系统有一个订单服务，提供了根据订单号查询订单的功能。

下面来实现这个功能，先使用 protocol buffer 来定义服务，然后再使用 gRPC 实现一元 RPC。

syntax = "proto3";

import "google/protobuf/wrappers.proto";
service OrderManagement {
  rpc getOrder(google.protobuf.StringValue) returns (Order) {}
}

message Order{
  string id = 1;
  string desc = 2;
  float price = 3;
  string destination = 4;
  repeated string items = 5;
}

接下来编写相应的服务端实现代码:

func (s *server) GetOrder(ctx context.Context, orderId *wrapper.StringValue) (*pb.Order, error) {
ord, exists := orderMap[orderId.Value]
if exists {
return &ord, status.New(codes.OK, "").Err()
}

return nil, status.Errorf(codes.NotFound, "Order does not exist. : ", orderId)

}

由于只是简单的例子，所以使用了一个简单的 map 来存储订单信息，这里我们可以使用数据库来替代。

接下来我们来实现客户端的代码，这里我采用 go 语言编写，代码比较简单，就是创建到服务器的连接并初始化客户端的存根，调用相应方法。

package main

import (
	"context"
	"log"
	"time"

	pb "github.com/overstarry/grpc-demo/shopping/v1"
	"google.golang.org/grpc"
	wrapper "google.golang.org/protobuf/types/known/wrapperspb"
)

const (
	address = "localhost:8081"
)

func main() {
	conn, err := grpc.Dial(address, grpc.WithInsecure(), grpc.WithBlock())
	if err != nil {
		log.Fatalf("did not connect: %v", err)
	}
	defer conn.Close()
	client := pb.NewOrderManagementClient(conn)
	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
	defer cancel()

	// Get Order
	retrievedOrder, err := client.GetOrder(ctx, &wrapper.StringValue{Value: "106"})
	log.Print("GetOrder Response -> : ", retrievedOrder)

这样就实现了简单的一元 RPC, 还是十分简单易懂的，适用于大部分的场景。

#服务端流式 RPC

服务端流式 RPC 与一元 RPC 相比，服务端接受到客户端请求后，会发一个响应的序列，也称为流。在所有响应流发生完后，服务端会以 trailer 元数据发送给客户端，以标记流的结束。

下面通过一个例子讲述服务端流式 RPC. 继续上面的订单服务，这时需要一个订单搜索的功能，提供一个关键字就能得到想要的结果。

1 编写 proto

  rpc searchOrders(google.protobuf.StringValue) returns (stream Order) {}

2 实现服务端代码

接下来实现相应的服务端代码

func (s *server) SearchOrders(searchQuery *wrappers.StringValue, stream pb.OrderManagement_SearchOrdersServer) error {

for key, order := range orderMap {
log.Print(key, order)
for _, itemStr := range order.Items {
log.Print(itemStr)
if strings.Contains(itemStr, searchQuery.Value) {
// Send the matching orders in a stream
err := stream.Send(&order)
if err != nil {
return fmt.Errorf("error sending message to stream : %v", err)
}
log.Print("Matching Order Found : " + key)
break
}
}
}
return nil
}

这里的逻辑很简单，主要是查找符合的订单，通过流依次发出去，使用了 send 方法。

3 客户端

接下来实现相应的客户端代码，客户端的代码和一元 RPC 模式很相似，区别在于服务端往流中发生了多个响应，因此需要循环处理多个响应，使用 recv 方法接收消息，直到流结束。

代码:

searchStream, _ := client.SearchOrders(ctx, &wrapper.StringValue{Value: "overstarry"})
for {
searchOrder, err := searchStream.Recv()
if err == io.EOF {
log.Print("EOF")
break
}

if err == nil {
log.Print("Search Result : ", searchOrder)
}
}

#客户端流式 RPC

客户端流式 RPC 和服务端流式 RPC 恰恰相反，客户端会发送多个请求给服务端，服务端会发送一个响应给客户端，但服务端不一定要处理完所有请求才发送响应。

接下来继续使用一个例子讲述客户端流式 RPC 的代码如何处理。

我们在订单服务中添加一个方法 updateOrders 用来更新多个订单，客户端以 stream 的形式向服务端发送订单列表，服务端接收请求后处理完毕，会发送一个订单状态的响应给客户端。

1 proto 代码

  rpc updateOrders(stream Order) returns (google.protobuf.StringValue) {}

可以看到与服务端流式 RPC 不同的是，客户端流RPC 使用了 stream order 作为参数。

定义完 proto，通过命令生成新的代码。

2 服务端实现

具体就不多说了，看看实现的代码。

func (s *server) UpdateOrders(stream pb.OrderManagement_UpdateOrdersServer) error {

ordersStr := "Updated Order IDs : "
for {
order, err := stream.Recv()
if err == io.EOF {
// Finished reading the order stream.
return stream.SendAndClose(&wrapper.StringValue{Value: "Orders processed " + ordersStr})
}

if err != nil {
return err
}
// Update order
orderMap[order.Id] = *order

log.Printf("Order ID : %s - %s", order.Id, "Updated")
ordersStr += order.Id + ", "
}
}

调用 SendAndClose 方法就可以向客户端发送响应。

3 客户端

客户端通过 updateStream.Send 发送多条消息，一旦消息全部发送完成，就可以接收服务端的响应。

    if err := updateStream.Send(&updOrder3); err != nil {
log.Fatalf("%v.Send(%v) = %v", updateStream, updOrder3, err)
}

updateRes, err := updateStream.CloseAndRecv()
if err != nil {
log.Fatalf("%v.CloseAndRecv() got error %v, want %v", updateStream, err, nil)
}
log.Printf("Update Orders Res : %s", updateRes)

#双向流 RPC 模式

在双向流 RPC 模式中，客户端通过流的形式向服务端发送请求，服务端接受请求后通过流的形式发送响应。调用必须由客户端发起。

#参考

https://grpc.io/docs/languages/go/quickstart/