在go语言中实现GRPC的四种通信模式示例代码

Go  /  管理员 发布于 1年前   445

在客户端和服务端的通信中，可以使用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/