gRPC 流：最佳实践和性能见解-Golang-PHP中文网

grpc 流：最佳实践和性能见解

介绍

grpc 流允许 protobuf 消息从客户端流式传输到服务器、从服务器流式传输到客户端，或者双向流式传输。
这一强大的功能可用于构建实时应用程序，例如聊天应用程序、实时监控仪表板等。

在本文中，我们将探讨如何正确使用 grpc 流。

先决条件

grpc基础知识
go 编程语言的基础知识（示例代码是用 go 编写的，但这个概念也可以应用于其他语言）
代码示例可在 github 上获取

良好实践

让我们检查一下使用 grpc 流的良好实践：

使用一元请求进行一元请求

一个常见的错误是对一元请求使用流式传输。
例如，考虑以下 grpc 服务定义：

service myservice {
  rpc getsomething (somethingrequest) returns (stream somethingresponse) {}
}

登录后复制

如果客户端只需要发送一个请求并接收一个响应，
您不需要使用流式传输。相反，我们可以按如下方式定义服务：

service myservice {
  rpc getsomething (somethingrequest) returns (somethingresponse) {}
}

登录后复制

通过对一元请求使用流式传输，我们增加了不必要的复杂性
到代码，这可能会使其更难理解和维护，而不是
从使用流媒体中获得任何好处。

比较一元请求和流请求的 golang 代码示例：

一元请求：

type somethingunary struct {
    pb.unimplementedsomethingunaryserver
}

func (s *somethingunary) getsomething(ctx context.context, req *pb.somethingrequest) (*pb.somethingresponse, error) {
    return &pb.somethingresponse{
        message: "hello " + req.name,
    }, nil
}

func testsomethingunary(t *testing.t) {
    conn := newserver(t, func(s grpc.serviceregistrar) {
        pb.registersomethingunaryserver(s, &somethingunary{})
    })

    client := pb.newsomethingunaryclient(conn)

    response, err := client.getsomething(
        context.background(),
        &pb.somethingrequest{
            name: "test",
        },
    )
    if err != nil {
        t.fatalf("failed to get something: %v", err)
    }

    if response.message != "hello test" {
        t.errorf("unexpected response: %v", response.message)
    }
}

登录后复制

流式一元请求：

type somethingstream struct {
    pb.unimplementedsomethingstreamserver
}

func (s *somethingstream) getsomething(req *pb.somethingrequest, stream pb.somethingstream_getsomethingserver) error {
    if err := stream.send(&pb.somethingresponse{
        message: "hello " + req.name,
    }); err != nil {
        return err
    }

    return nil
}

func testsomethingstream(t *testing.t) {
    conn := newserver(t, func(s grpc.serviceregistrar) {
        pb.registersomethingstreamserver(s, &somethingstream{})
    })

    client := pb.newsomethingstreamclient(conn)

    stream, err := client.getsomething(
        context.background(),
        &pb.somethingrequest{
            name: "test",
        },
    )
    if err != nil {
        t.fatalf("failed to get something stream: %v", err)
    }

    response, err := stream.recv()
    if err != nil {
        t.fatalf("failed to receive response: %v", err)
    }

    if response.message != "hello test" {
        t.errorf("unexpected response: %v", response.message)
    }
}

登录后复制

我们可以看到，一元请求的代码更简单，更容易理解
比流请求的代码。

如果可以的话，一次发送多个文档

让我们比较一下这两个服务定义：

service bookstore {
  rpc listbooks(listbooksrequest) returns (stream book) {}
}

service bookstorebatch {
  rpc listbooks(listbooksrequest) returns (stream listbooksresponse) {}
}

message listbooksresponse {
  repeated book books = 1;
}

登录后复制

bookstore 一次流式传输一本书，而 bookstorebatch 同时流式传输多本书。

如果客户端需要列出所有书籍，使用bookstorebatch 效率更高
因为它减少了客户端和服务器之间的往返次数。

让我们看看 bookstore 和 bookstorebatch 的 golang 代码示例：

书店：

type bookstore struct {
    pb.unimplementedbookstoreserver
}

func (s *bookstore) listbooks(req *pb.listbooksrequest, stream pb.bookstore_listbooksserver) error {
    for _, b := range bookstoredata {
        if b.author == req.author {
            if err := stream.send(&pb.book{
                title:           b.title,
                author:          b.author,
                publicationyear: int32(b.publicationyear),
                genre:           b.genre,
            }); err != nil {
                return err
            }
        }
    }
    return nil
}

func testbookstore_listbooks(t *testing.t) {
    conn := newserver(t, func(s grpc.serviceregistrar) {
        pb.registerbookstoreserver(s, &bookstore{})
    })

    client := pb.newbookstoreclient(conn)

    stream, err := client.listbooks(
        context.background(),
        &pb.listbooksrequest{
            author: charlesdickens,
        },
    )
    if err != nil {
        t.fatalf("failed to list books: %v", err)
    }

    books := []*pb.book{}
    for {
        book, err := stream.recv()
        if err != nil {
            break
        }
        books = append(books, book)
    }

    if len(books) != charlesdickensbooks {
        t.errorf("unexpected number of books: %d", len(books))
    }
}

登录后复制

书店批次：

type bookstorebatch struct {
    pb.unimplementedbookstorebatchserver
}

func (s *bookstorebatch) listbooks(req *pb.listbooksrequest, stream pb.bookstorebatch_listbooksserver) error {
    const batchsize = 10
    books := make([]*pb.book, 0, batchsize)
    for _, b := range bookstoredata {
        if b.author == req.author {
            books = append(books, &pb.book{
                title:           b.title,
                author:          b.author,
                publicationyear: int32(b.publicationyear),
                genre:           b.genre,
            })

            if len(books) == batchsize {
                if err := stream.send(&pb.listbooksresponse{
                    books: books,
                }); err != nil {
                    return err
                }
                books = books[:0]
            }
        }
    }

    if len(books) > 0 {
        if err := stream.send(&pb.listbooksresponse{
            books: books,
        }); err != nil {
            return nil
        }
    }

    return nil
}

func testbookstorebatch_listbooks(t *testing.t) {
    conn := newserver(t, func(s grpc.serviceregistrar) {
        pb.registerbookstorebatchserver(s, &bookstorebatch{})
    })

    client := pb.newbookstorebatchclient(conn)

    stream, err := client.listbooks(
        context.background(),
        &pb.listbooksrequest{
            author: charlesdickens,
        },
    )
    if err != nil {
        t.fatalf("failed to list books: %v", err)
    }

    books := []*pb.book{}
    for {
        response, err := stream.recv()
        if err != nil {
            break
        }
        books = append(books, response.books...)
    }

    if len(books) != charlesdickensbooks {
        t.errorf("unexpected number of books: %d", len(books))
    }
}

登录后复制

从上面的代码中，需要明确哪一个更好。
让我们运行一个基准测试来看看差异：

书店基准：

func benchmarkbookstore_listbooks(b *testing.b) {
    conn := newserver(b, func(s grpc.serviceregistrar) {
        pb.registerbookstoreserver(s, &bookstore{})
    })

    client := pb.newbookstoreclient(conn)

    var benchinnerbooks []*pb.book
    b.resettimer()
    for i := 0; i < b.n; i++ {
        stream, err := client.listbooks(
            context.background(),
            &pb.listbooksrequest{
                author: charlesdickens,
            },
        )
        if err != nil {
            b.fatalf("failed to list books: %v", err)
        }

        books := []*pb.book{}
        for {
            book, err := stream.recv()
            if err != nil {
                break
            }
            books = append(books, book)
        }

        benchinnerbooks = books
    }

    benchbooks = benchinnerbooks
}

登录后复制

bookstorebatch 基准：

先见AI

数据为基，先见未见

查看详情

func benchmarkbookstorebatch_listbooks(b *testing.b) {
    conn := newserver(b, func(s grpc.serviceregistrar) {
        pb.registerbookstorebatchserver(s, &bookstorebatch{})
    })

    client := pb.newbookstorebatchclient(conn)

    var benchinnerbooks []*pb.book
    b.resettimer()
    for i := 0; i < b.n; i++ {
        stream, err := client.listbooks(
            context.background(),
            &pb.listbooksrequest{
                author: charlesdickens,
            },
        )
        if err != nil {
            b.fatalf("failed to list books: %v", err)
        }

        books := []*pb.book{}
        for {
            response, err := stream.recv()
            if err != nil {
                break
            }
            books = append(books, response.books...)
        }

        benchinnerbooks = books
    }

    benchbooks = benchinnerbooks
}

登录后复制

基准测试结果：

benchmarkbookstore_listbooks
benchmarkbookstore_listbooks-12                      732           1647454 ns/op           85974 b/op       1989 allocs/op
benchmarkbookstorebatch_listbooks
benchmarkbookstorebatch_listbooks-12                1202            937491 ns/op           61098 b/op        853 allocs/op

登录后复制

多么大的进步啊！ bookstorebatch 比 bookstore 快 1.75 倍。

但是为什么 bookstorebatch 比 bookstore 快？

服务器每次向客户端发送消息流.send()，都需要
对消息进行编码并通过网络发送。通过发送多个文件
我们立即减少了服务器需要编码和发送的次数
消息，不仅提高了服务器的性能，还提高了
对于需要解码消息的客户端。

在上面的例子中，批量大小设置为10，但客户端可以根据网络情况和文档大小进行调整。

使用双向流来控制流量

书店示例返回所有书籍并完成流，但如果客户端
需要实时观察事件（例如传感器），使用双向
直播是正确的选择。

双向流有点棘手，因为客户端和服务器都
可以同时发送和接收消息。希望 golang 能让这一切变得简单
像这样处理并发。

如前所述，传感器是双向流的一个很好的例子。
监视功能允许客户端决定监视和请求哪些传感器
如果需要的话，当前值。

让我们看一下下面的protobuf定义：

service sensor {
  rpc watch(stream watchrequest) returns (stream watchresponse) {}
}

message watchrequest {
  oneof request {
    watchcreaterequest create_request = 1;
    watchcancelrequest cancel_request = 2;
    watchnowrequest now_request = 3;
  }
}

message watchcreaterequest {
  // sensor_id contains the sensor id to watch.
  string sensor_id = 1;
}

message watchcancelrequest {
  // sensor_id contains the sensor id to cancel.
  string sensor_id = 1;
}

message watchnowrequest {
  // sensor_id contains the sensor id to get the current value.
  string sensor_id = 1;
}

message watchresponse {
  // sensor_id contains the sensor id for the current response.
  string sensor_id = 1;
  // created is true if the watch was created successfully.
  bool created = 2;
  // canceleted is true if the watch was canceled successfully or if the creation failed.
  bool canceleted = 3;
  // error contains the error message if something went wrong.
  string error = 4;
  // timestamp contains the timestamp of the value.
  google.protobuf.timestamp timestamp = 5;
  // value contains the value of the sensor.
  int32 value = 6;
}

登录后复制

请求消息不仅仅是消息流，更是一条可以
包含不同类型的请求。 oneof 指令允许我们定义一个
只能包含指定类型之一的字段。

传感器的 golang 代码将被忽略，但您可以在这里找到它

serverstream 包装流和传感器数据，使其更易于使用。

type serverstream struct {
    s           *sensorservice         // service
    stream      pb.sensor_watchserver  // stream
    sendch      chan *pb.watchresponse // control channel
    sensorch    chan sensordata        // data channel
    sensorwatch map[string]int         // map of sensor id to watch id
}

登录后复制

如前所述，服务器可以同时发送和接收消息，一个
函数将处理传入的消息，另一个函数将处理
传出消息。

接收消息：

func (ss *serverstream) recvloop() error {
    defer ss.close()
    for {
        req, err := ss.stream.recv()
        if errors.is(err, io.eof) {
            return nil
        }
        if err != nil {
            return err
        }

        switch req := req.request.(type) {
        case *pb.watchrequest_createrequest:
            // ignore validation (check the full code)

            // create a channel to send data to the client
            id := sensor.watch(ss.sensorch)
            ss.sensorwatch[sensorid] = id

            // send created message
            ss.sendch <- &pb.watchresponse{
                sensorid: sensorid,
                created:  true,
            }

        case *pb.watchrequest_cancelrequest:
            // ignore validation (check the full code)

            // cancel the watch
            ss.s.sensors[sensorid].cancel(id)
            delete(ss.sensorwatch, sensorid)

            ss.sendch <- &pb.watchresponse{
                sensorid:   sensorid,
                canceleted: true,
            }

        case *pb.watchrequest_nowrequest:
            // ignore validation (check the full code)

            // send current value
            ss.sendch <- &pb.watchresponse{
                sensorid:  sensorid,
                timestamp: timestamppb.now(),
                value:     int32(sensor.read()),
            }
        }
    }
}

登录后复制

switch语句用于处理不同类型的请求并做出决定
如何处理每个请求。只保留recvloop 函数很重要
读取消息但不向客户端发送消息，因此我们有 sendloop
将从控制通道读取消息并将其发送到客户端。

发送消息：

func (ss *serverstream) sendloop() {
    for {
        select {
        case m, ok := <-ss.sendch:
            if !ok {
                return
            }

            // send message
            if err := ss.stream.send(m); err != nil {
                return
            }

        case data, ok := <-ss.sensorch:
            if !ok {
                return
            }

            // send data
            if err := ss.stream.send(&pb.watchresponse{
                sensorid:  data.id,
                timestamp: timestamppb.new(data.time),
                value:     int32(data.val),
            }); err != nil {
                return
            }

        case <-ss.stream.context().done():
            return
        }
    }
}

登录后复制

sendloop函数读取控制通道和数据通道并发送
发送给客户端的消息。如果流关闭，该函数将返回。

最后，传感器服务的快乐路径测试：

func TestSensor(t *testing.T) {
    conn := newServer(t, func(s grpc.ServiceRegistrar) {
        pb.RegisterSensorServer(s, &sensorService{
            sensors: newSensors(),
        })
    })

    client := pb.NewSensorClient(conn)

    stream, err := client.Watch(context.Background())
    if err != nil {
        t.Fatalf("failed to watch: %v", err)
    }

    response := make(chan *pb.WatchResponse)
    // Go routine to read from the stream
    go func() {
        defer close(response)
        for {
            resp, err := stream.Recv()
            if errors.Is(err, io.EOF) {
                return
            }
            if err != nil {
                return
            }
            response <- resp
        }
    }()

    createRequest(t, stream, "temp")
    waitUntilCreated(t, response, "temp")
    waitForSensorData(t, response, "temp")

    createRequest(t, stream, "pres")
    waitUntilCreated(t, response, "pres")
    waitForSensorData(t, response, "pres")

    waitForSensorData(t, response, "temp")
    waitForSensorData(t, response, "pres")

    // invalid sensor
    createRequest(t, stream, "invalid")
    waitUntilCanceled(t, response, "invalid")

    nowRequest(t, stream, "light")
    waitForSensorData(t, response, "light")
    // Wait for 2 seconds to make sure we don't receive any data for light
    waitForNoSensorData(t, response, "light", 2*time.Second)

    cancelRequest(t, stream, "temp")
    waitUntilCanceled(t, response, "temp")

    waitForSensorData(t, response, "pres")
    // Wait for 2 seconds to make sure we don't receive any data for temp
    waitForNoSensorData(t, response, "temp", 2*time.Second)

    err = stream.CloseSend()
    if err != nil {
        t.Fatalf("failed to close send: %v", err)
    }
}

登录后复制

从上面的测试中我们可以看到客户端可以创建、取消、获取当前
传感器的值。客户端还可以同时观看多个传感器。