桃圆 發表於 2020-4-13 20:30:00

Go gRPC教程-服务端流式RPC（三）

<h3 id="前言">前言</h3>
<p>上一篇介绍了<code>简单模式RPC</code>，当数据量大或者需要不断传输数据时候，我们应该使用流式RPC，它允许我们边处理边传输数据。本篇先介绍<code>服务端流式RPC</code>。</p>
<p><code>服务端流式RPC</code>：客户端发送请求到服务器，拿到一个流去读取返回的消息序列。 客户端读取返回的流，直到里面没有任何消息。</p>
<h6 id="情景模拟实时获取股票走势">情景模拟：实时获取股票走势。</h6>
<p>1.客户端要获取某原油股的实时走势，客户端发送一个请求</p>
<p>2.服务端实时返回该股票的走势</p>
<h3 id="新建proto文件">新建proto文件</h3>
<p>新建server_stream.proto文件</p>
<p>1.定义发送信息</p>
<pre><code class="language-protobuf">// 定义发送请求信息
message SimpleRequest{
    // 定义发送的参数，采用驼峰命名方式，小写加下划线，如：student_name
    // 请求参数
    string data = 1;
}
</code></pre>
<p>2.定义接收信息</p>
<pre><code class="language-protobuf">// 定义流式响应信息
message StreamResponse{
    // 流式响应数据
    string stream_value = 1;
}
</code></pre>
<p>3.定义服务方法ListValue</p>
<p>服务端流式rpc，只要在响应数据前添加stream即可</p>
<pre><code class="language-protobuf">// 定义我们的服务（可定义多个服务,每个服务可定义多个接口）
service StreamServer{
    // 服务端流式rpc，在响应数据前添加stream
    rpc ListValue(SimpleRequest)returns(stream StreamResponse){};
}
</code></pre>
<p>4.编译proto文件</p>
<p>进入server_stream.proto所在目录，运行指令:</p>
<p><code>protoc --go_out=plugins=grpc:./ ./server_stream.proto</code></p>
<h3 id="创建server端">创建Server端</h3>
<p>1.定义我们的服务，并实现ListValue方法</p>
<pre><code class="language-go">// SimpleService 定义我们的服务
type StreamService struct{}
// ListValue 实现ListValue方法
func (s *StreamService) ListValue(req *pb.SimpleRequest, srv pb.StreamServer_ListValueServer) error {
        for n := 0; n &lt; 5; n++ {
                // 向流中发送消息， 默认每次send送消息最大长度为`math.MaxInt32`bytes
                err := srv.Send(&amp;pb.StreamResponse{
                        StreamValue: req.Data + strconv.Itoa(n),
                })
                if err != nil {
                        return err
                }
        }
        return nil
}
</code></pre>
<p>初学者可能觉得比较迷惑，ListValue的参数和返回值是怎样确定的。其实这些都是编译proto时生成的.pb.go文件中有定义，我们只需要实现就可以了。</p>
<p><img src="https://img2020.cnblogs.com/blog/1508611/202004/1508611-20200413200011972-285642192.png" alt="" loading="lazy"></p>
<p>2.启动gRPC服务器</p>
<pre><code class="language-go">const (
        // Address 监听地址
        Address string = ":8000"
        // Network 网络通信协议
        Network string = "tcp"
)

func main() {
        // 监听本地端口
        listener, err := net.Listen(Network, Address)
        if err != nil {
                log.Fatalf("net.Listen err: %v", err)
        }
        log.Println(Address + " net.Listing...")
        // 新建gRPC服务器实例
        // 默认单次接收最大消息长度为`1024*1024*4`bytes(4M)，单次发送消息最大长度为`math.MaxInt32`bytes
        // grpcServer := grpc.NewServer(grpc.MaxRecvMsgSize(1024*1024*4), grpc.MaxSendMsgSize(math.MaxInt32))
        grpcServer := grpc.NewServer()
        // 在gRPC服务器注册我们的服务
        pb.RegisterStreamServerServer(grpcServer, &amp;StreamService{})

        //用服务器 Serve() 方法以及我们的端口信息区实现阻塞等待，直到进程被杀死或者 Stop() 被调用
        err = grpcServer.Serve(listener)
        if err != nil {
                log.Fatalf("grpcServer.Serve err: %v", err)
        }
}
</code></pre>
<p>运行服务端</p>
<pre><code class="language-powershell">go run server.go
:8000 net.Listing...
</code></pre>
<h3 id="创建client端">创建Client端</h3>
<p>1.创建调用服务端ListValue方法</p>
<pre><code class="language-go">// listValue 调用服务端的ListValue方法
func listValue() {
        // 创建发送结构体
        req := pb.SimpleRequest{
                Data: "stream server grpc ",
        }
        // 调用我们的服务(ListValue方法)
        stream, err := grpcClient.ListValue(context.Background(), &amp;req)
        if err != nil {
                log.Fatalf("Call ListStr err: %v", err)
        }
        for {
                //Recv() 方法接收服务端消息，默认每次Recv()最大消息长度为`1024*1024*4`bytes(4M)
                res, err := stream.Recv()
                // 判断消息流是否已经结束
                if err == io.EOF {
                        break
                }
                if err != nil {
                        log.Fatalf("ListStr get stream err: %v", err)
                }
                // 打印返回值
                log.Println(res.StreamValue)
        }
}
</code></pre>
<p>2.启动gRPC客户端</p>
<pre><code class="language-go">// Address 连接地址
const Address string = ":8000"

var grpcClient pb.StreamServerClient

func main() {
        // 连接服务器
        conn, err := grpc.Dial(Address, grpc.WithInsecure())
        if err != nil {
                log.Fatalf("net.Connect err: %v", err)
        }
        defer conn.Close()

        // 建立gRPC连接
        grpcClient = pb.NewStreamServerClient(conn)
        route()
        listValue()
}
</code></pre>
<p>运行客户端</p>
<pre><code class="language-powershell">go run client.go
stream server grpc 0
stream server grpc 1
stream server grpc 2
stream server grpc 3
stream server grpc 4
</code></pre>
<p>客户端不断从服务端获取数据</p>
<h3 id="思考">思考</h3>
<p>假如服务端不停发送数据，类似获取股票走势实时数据，客户端能自己停止获取数据吗？</p>
<p>答案：可以的</p>
<p>1.我们把服务端的ListValue方法稍微修改</p>
<pre><code class="language-go">// ListValue 实现ListValue方法
func (s *StreamService) ListValue(req *pb.SimpleRequest, srv pb.StreamServer_ListValueServer) error {
        for n := 0; n &lt; 15; n++ {
                // 向流中发送消息， 默认每次send送消息最大长度为`math.MaxInt32`bytes
                err := srv.Send(&amp;pb.StreamResponse{
                        StreamValue: req.Data + strconv.Itoa(n),
                })
                if err != nil {
                        return err
                }
                log.Println(n)
                time.Sleep(1 * time.Second)
        }
        return nil
}
</code></pre>
<p>2.再把客户端调用ListValue方法的实现稍作修改，就可以得到结果了</p>
<pre><code class="language-go">// listValue 调用服务端的ListValue方法
func listValue() {
        // 创建发送结构体
        req := pb.SimpleRequest{
                Data: "stream server grpc ",
        }
        // 调用我们的服务(Route方法)
        // 同时传入了一个 context.Context ，在有需要时可以让我们改变RPC的行为，比如超时/取消一个正在运行的RPC
        stream, err := grpcClient.ListValue(context.Background(), &amp;req)
        if err != nil {
                log.Fatalf("Call ListStr err: %v", err)
        }
        for {
                //Recv() 方法接收服务端消息，默认每次Recv()最大消息长度为`1024*1024*4`bytes(4M)
                res, err := stream.Recv()
                // 判断消息流是否已经结束
                if err == io.EOF {
                        break
                }
                if err != nil {
                        log.Fatalf("ListStr get stream err: %v", err)
                }
                // 打印返回值
                log.Println(res.StreamValue)
                break
        }
        //可以使用CloseSend()关闭stream，这样服务端就不会继续产生流消息
        //调用CloseSend()后，若继续调用Recv()，会重新激活stream，接着之前结果获取消息
        stream.CloseSend()
}
</code></pre>
<p>只需要调用<code>CloseSend()</code>方法，就可以关闭服务端的stream，让它停止发送数据。值得注意的是，调用<code>CloseSend()</code>后，若继续调用<code>Recv()</code>，会重新激活stream，接着当前的结果继续获取消息。</p>
<p>这能完美解决客户端<code>暂停</code>-&gt;<code>继续</code>获取数据的操作。</p>
<h3 id="总结">总结</h3>
<p>本篇介绍了<code>服务端流式RPC</code>的简单使用，客户端发起一个请求，服务端不停返回数据，直到服务端停止发送数据或客户端主动停止接收数据为止。下篇将介绍<code>客户端流式RPC</code>。</p>
<p>教程源码地址：https://github.com/Bingjian-Zhu/go-grpc-example<br>
参考：gRPC官方文档中文版</p>


</div>
<div id="MySignature" role="contentinfo">
    看完之后若觉得对自己有帮助，恳请点赞或评论。这是对我最大的鼓励！<br><br>
来源：https://www.cnblogs.com/FireworksEasyCool/p/12693749.html
頁: [1]
查看完整版本: Go gRPC教程-服务端流式RPC（三）