旅行从心开始 發表於 2020-4-14 15:13:00

Go gRPC教程-双向流式RPC（五）

<h3 id="前言">前言</h3>
<p>上一篇介绍了<code>客户端流式RPC</code>，客户端不断的向服务端发送数据流，在发送结束或流关闭后，由服务端返回一个响应。本篇将介绍<code>双向流式RPC</code>。</p>
<p><code>双向流式RPC</code>：客户端和服务端双方使用读写流去发送一个消息序列，两个流独立操作，双方可以同时发送和同时接收。</p>
<h6 id="情景模拟双方对话可以一问一答一问多答多问一答形式灵活">情景模拟：双方对话（可以一问一答、一问多答、多问一答，形式灵活）。</h6>
<h3 id="新建proto文件">新建proto文件</h3>
<p>新建both_stream.proto文件</p>
<p>1.定义发送信息</p>
<pre><code class="language-protobuf">// 定义流式请求信息
message StreamRequest{
    //流请求参数
    string question = 1;
}
</code></pre>
<p>2.定义接收信息</p>
<pre><code class="language-protobuf">// 定义流式响应信息
message StreamResponse{
    //流响应数据
    string answer = 1;
}
</code></pre>
<p>3.定义服务方法Conversations</p>
<p>双向流式rpc，只要在请求的参数前和响应参数前都添加stream即可</p>
<pre><code class="language-protobuf">service Stream{
    // 双向流式rpc，同时在请求参数前和响应参数前加上stream
    rpc Conversations(stream StreamRequest) returns(stream StreamResponse){};
}
</code></pre>
<p>4.编译proto文件</p>
<p>进入both_stream.proto所在目录，运行指令:</p>
<p><code>protoc --go_out=plugins=grpc:./ ./both_stream.proto</code></p>
<h3 id="创建server端">创建Server端</h3>
<p>1.定义我们的服务，并实现RouteList方法</p>
<p>这里简单实现对话中一问一答的形式</p>
<pre><code class="language-go">// StreamService 定义我们的服务
type StreamService struct{}
// Conversations 实现Conversations方法
func (s *StreamService) Conversations(srv pb.Stream_ConversationsServer) error {
        n := 1
        for {
                req, err := srv.Recv()
                if err == io.EOF {
                        return nil
                }
                if err != nil {
                        return err
                }
                err = srv.Send(&amp;pb.StreamResponse{
                        Answer: "from stream server answer: the " + strconv.Itoa(n) + " question is " + req.Question,
                })
                if err != nil {
                        return err
                }
                n++
                log.Printf("from stream client question: %s", req.Question)
        }
}
</code></pre>
<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服务器实例
        grpcServer := grpc.NewServer()
        // 在gRPC服务器注册我们的服务
        pb.RegisterStreamServer(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.创建调用服务端Conversations方法</p>
<pre><code class="language-go">// conversations 调用服务端的Conversations方法
func conversations() {
        //调用服务端的Conversations方法，获取流
        stream, err := streamClient.Conversations(context.Background())
        if err != nil {
                log.Fatalf("get conversations stream err: %v", err)
        }
        for n := 0; n &lt; 5; n++ {
                err := stream.Send(&amp;pb.StreamRequest{Question: "stream client rpc " + strconv.Itoa(n)})
                if err != nil {
                        log.Fatalf("stream request err: %v", err)
                }
                res, err := stream.Recv()
                if err == io.EOF {
                        break
                }
                if err != nil {
                        log.Fatalf("Conversations get stream err: %v", err)
                }
                // 打印返回值
                log.Println(res.Answer)
        }
        //最后关闭流
        err = stream.CloseSend()
        if err != nil {
                log.Fatalf("Conversations close stream err: %v", err)
        }
}
</code></pre>
<p>2.启动gRPC客户端</p>
<pre><code class="language-go">// Address 连接地址
const Address string = ":8000"

var streamClient pb.StreamClient

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

        // 建立gRPC连接
        streamClient = pb.NewStreamClient(conn)
        conversations()
}
</code></pre>
<p>运行客户端，获取到服务端的应答</p>
<pre><code class="language-powershell">go run client.go
from stream server answer: the 1 question is stream client rpc 0
from stream server answer: the 2 question is stream client rpc 1
from stream server answer: the 3 question is stream client rpc 2
from stream server answer: the 4 question is stream client rpc 3
from stream server answer: the 5 question is stream client rpc 4
</code></pre>
<p>服务端获取到来自客户端的提问</p>
<pre><code class="language-powershell">from stream client question: stream client rpc 0
from stream client question: stream client rpc 1
from stream client question: stream client rpc 2
from stream client question: stream client rpc 3
from stream client question: stream client rpc 4
</code></pre>
<h3 id="总结">总结</h3>
<p>本篇介绍了<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/12698194.html
頁: [1]
查看完整版本: Go gRPC教程-双向流式RPC（五）