Go gRPC服务客户端流式RPC教程

Go gRPC服务客户端流式RPC教程

本教程将介绍如何在Go语言中实现gRPC客户端流式RPC。

客户端流式RPC允许客户端像流一样发送多个请求，然后服务器响应一个单独的消息。该方案通常用于需要客户端向服务器传输大量数据的场景。在本文中，我们将使用Go中的grpc功能库来实现该方案。

步骤1：安装和设置gRPC

首先，我们需要安装Go中的gRPC库。可以使用以下命令：

go get google.golang.org/grpc

然后，您可以在Go程序中导入grpc库。

接下来，我们需要创建一个proto文件来定义服务和消息。在这个例子中，文件名为streaming.proto

我们的proto文件包括消息的架构以及服务的规范。例如：

syntax = "proto3";

message StreamRequest {
    string message = 1;
}

message StreamResponse {
    string message = 1;
}

service StreamingService {
    rpc Stream(StreamRequest) returns (StreamResponse) {};
}

此代码块中，定义了StreamRequest和StreamResponse消息，以及StreamingService服务，其中包含一个名为Stream的RPC。

步骤2：实现服务器端代码

使用上面的proto文件生成Go代码，可以使用以下命令：

protoc -I streaming/ streaming/streaming.proto --go_out=plugins=grpc:streaming

代码生成完成后，我们就可以开始实现服务器端代码了。首先，我们需要创建StreamingService的实现：

// server.go

package main

import (
    "fmt"
    "io"
    "log"
    "net"
    "google.golang.org/grpc"
    pb "github.com/myusername/myproject/streaming"
)

type server struct{}

func (s *server) Stream(stream pb.StreamingService_StreamServer) error {
    for {
        msg, err := stream.Recv()
        if err == io.EOF {
            return stream.SendAndClose(&pb.StreamResponse{Message: "Response"})
        }
        if err != nil {
            return err
        }
        fmt.Println("Message received:", msg.Message)
    }
    return nil
}

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

    s := grpc.NewServer()
    pb.RegisterStreamingServiceServer(s, &server{})
    fmt.Println("Server started on port", port)
    if err := s.Serve(lis); err != nil {
        log.Fatalf("failed to serve: %v", err)
    }
}

然后，实现Stream方法。在此示例中，Stream方法将被不断地调用，直到客户端发送一个EOF消息，此时服务器将发送一个带有响应消息的stream.SendAndClose响应。

步骤3：实现客户端代码

客户端代码如下所示：

// client.go

package main

import (
    "io"
    "log"
    "time"
    "google.golang.org/grpc"
    pb "github.com/myusername/myproject/streaming"
)

const address = "localhost:50051"

func main() {
    conn, err := grpc.Dial(address, grpc.WithInsecure(), grpc.WithBlock())
    if err != nil {
        log.Fatalf("did not connect: %v", err)
    }

    defer conn.Close()

    c := pb.NewStreamingServiceClient(conn)

    stream, err := c.Stream(context.Background())
    if err != nil {
        log.Fatalf("could not stream: %v", err)
    }

    for i:=1; i<=10; i++ {
        if err := stream.Send(&pb.StreamRequest{Message: "Request"}); err != nil {
            log.Fatalf("Failed to send: %v", err)
        }
        time.Sleep(time.Second)
    }

    response, err := stream.CloseAndRecv()
    if err != nil {
        log.Fatalf("Failed to receive: %v", err)
    }
    log.Println("Response received:", response.Message)
}

该方法将创建一个到服务器的连接，然后向服务器发送10个StreamRequest消息。然后，它将调用stream.CloseAndRecv来等待响应。

步骤4：运行客户端和服务器代码

使用以下命令启动服务器：

go run server.go

然后，在另一个控制台窗口中，使用以下命令运行客户端：

go run client.go

输出如下：

Message received: Request
Message received: Request
Message received: Request
Message received: Request
Message received: Request
Message received: Request
Message received: Request
Message received: Request
Message received: Request
Message received: Request
Response received: Response

这表明：客户端发送了10个请求，服务器收到了请求并打印消息，而客户端接收到了带有响应消息的响应。

示例2：使用流式RPC推送数据到gRPC服务器

在第二个示例中，我们将客户端流式RPC用于推送一些数据到服务器，服务器将接收并处理这些数据。

代码如下：

// server.go

package main

import (
    "fmt"
    "io"
    "log"
    "net"
    "google.golang.org/grpc"
    pb "github.com/myusername/myproject/streaming"
)

type server struct{}

func (s *server) Stream(stream pb.StreamingService_StreamServer) error {
    for {
        _, err := stream.Recv()
        if err == io.EOF {
            return nil
        }
        if err != nil {
            return err
        }
    }
    return nil
}

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

    s := grpc.NewServer()
    pb.RegisterStreamingServiceServer(s, &server{})
    fmt.Println("Server started on port", port)
    if err := s.Serve(lis); err != nil {
        log.Fatalf("failed to serve: %v", err)
    }
}

在此示例中，我们只需要Receiver方法接收和处理来自客户端的流式消息，并返回空响应即可。而客户端需要发送一些数据，代码如下：

// client.go

package main

import (
    "io"
    "log"
    "time"
    "google.golang.org/grpc"
    pb "github.com/myusername/myproject/streaming"
)

const address = "localhost:50051"

func main() {
    conn, err := grpc.Dial(address, grpc.WithInsecure(), grpc.WithBlock())
    if err != nil {
        log.Fatalf("did not connect: %v", err)
    }

    defer conn.Close()

    c := pb.NewStreamingServiceClient(conn)

    stream, err := c.Stream(context.Background())
    if err != nil {
        log.Fatalf("could not stream: %v", err)
    }

    for i:=1; i<=10; i++ {
        if err := stream.Send(&pb.StreamRequest{Message: "Request"}); err != nil {
            log.Fatalf("Failed to send: %v", err)
        }
        time.Sleep(time.Second)
    }

    _, err = stream.CloseAndRecv()
    if err != nil {
        log.Fatalf("Failed to receive: %v", err)
    }
}

在此示例中，我们只需要在for循环中发送10个“Request”消息。然后，我们可以关闭stream并等待响应。

总结

在本教程中，我们介绍了如何使用gRPC的客户端流式RPC功能。我们实现了一个可以向服务器推送流消息的客户端，并演示了如何从服务器发送响应消息。通过这个例子，相信您已经知道如何实现和使用gRPC的流式RPC功能了。