上一篇介绍了客户端流式 RPC，客户端不断地向服务端发送数据流，在发送结束或流关闭后，由服务端返回一个响应。本篇将介绍双向流式 RPC。
双向流式 RPC：客户端和服务端双方使用读写流去发送一个消息序列，两个流独立操作，双方可以同时发送和同时接收。
情景模拟：双方对话（可以一问一答、一问多答、多问一答，形式灵活）。
新建 proto 文件
新建 both_stream.proto 文件。
1. 定义发送信息
// 定义流式请求信息
message StreamRequest{
//流请求参数
string question = 1;
}2. 定义接收信息
// 定义流式响应信息
message StreamResponse{
//流响应数据
string answer = 1;
}3. 定义服务方法 Conversations
双向流式 rpc，只要在请求的参数前和响应参数前都添加 stream 即可。
service Stream{
// 双向流式rpc，同时在请求参数前和响应参数前加上stream
rpc Conversations(stream StreamRequest) returns(stream StreamResponse){};
}4. 编译 proto 文件
进入 both_stream.proto 所在目录，运行指令：
protoc --go_out=plugins=grpc:./ ./both_stream.proto创建 Server 端
1. 定义我们的服务，并实现 Conversations 方法
这里简单实现对话中一问一答的形式。
// 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(&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)
}
}2. 启动 gRPC 服务器
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, &StreamService{})
//通过服务器 Serve() 方法监听该端口，阻塞等待直到进程被杀死或者 Stop() 被调用
err = grpcServer.Serve(listener)
if err != nil {
log.Fatalf("grpcServer.Serve err: %v", err)
}
}运行服务端：
go run server.go:8000 net.Listing...创建 Client 端
1. 创建调用服务端 Conversations 方法
// 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 < 5; n++ {
err := stream.Send(&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)
}
}2. 启动 gRPC 客户端
// 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()
}运行客户端，获取到服务端的应答：
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服务端获取到来自客户端的提问：
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总结
本篇介绍了双向流式 RPC 的简单使用。
