前言#
上一篇介绍了服务端流式RPC，客户端发送请求到服务器，拿到一个流去读取返回的消息序列。 客户端读取返回的流的数据。本篇将介绍客户端流式RPC。
客户端流式RPC：与服务端流式RPC相反，客户端不断的向服务端发送数据流，而在发送结束后，由服务端返回一个响应。
情景模拟：客户端大量数据上传到服务端。
新建proto文件#
新建client_stream.proto文件
1.定义发送信息
Copy
// 定义流式请求信息 message StreamRequest{ //流式请求参数 string stream_data = 1; }
2.定义接收信息
Copy
// 定义响应信息 message SimpleResponse{ //响应码 int32 code = 1; //响应值 string value = 2; }
3.定义服务方法RouteList
客户端流式rpc，只要在请求的参数前添加stream即可
Copy
service StreamClient{ // 客户端流式rpc，在请求的参数前添加stream rpc RouteList (stream StreamRequest) returns (SimpleResponse){}; }
4.编译proto文件
进入client_stream.proto所在目录，运行指令:
protoc --go_out=plugins=grpc:./ ./client_stream.proto
创建Server端#
1.定义我们的服务，并实现RouteList方法
Copy
`type SimpleService struct{}
func (s *SimpleService) RouteList(srv pb.StreamClient_RouteListServer) error {
for {
res, err := srv.Recv()
if err == io.EOF {
return srv.SendAndClose(&pb.SimpleResponse{Value: "ok"})
}
if err != nil {
return err
}
log.Println(res.StreamData)
}
}`
2.启动gRPC服务器
Copy
`const (
Address string = ":8000"
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...")
grpcServer := grpc.NewServer()
pb.RegisterStreamClientServer(grpcServer, &SimpleService{})
err = grpcServer.Serve(listener)
if err != nil {
log.Fatalf("grpcServer.Serve err: %v", err)
}
}`
运行服务端
Copy
go run server.go :8000 net.Listing...
创建Client端#
1.创建调用服务端RouteList方法
Copy
`func routeList() {
stream, err := streamClient.RouteList(context.Background())
if err != nil {
log.Fatalf("Upload list err: %v", err)
}
for n := 0; n < 5; n++ {
err := stream.Send(&pb.StreamRequest{StreamData: "stream client rpc " + strconv.Itoa(n)})
if err != nil {
log.Fatalf("stream request err: %v", err)
}
}
res, err := stream.CloseAndRecv()
if err != nil {
log.Fatalf("RouteList get response err: %v", err)
}
log.Println(res)
}`
2.启动gRPC客户端
Copy
`const Address string = “:8000”
var streamClient pb.StreamClientClient
func main() {
conn, err := grpc.Dial(Address, grpc.WithInsecure())
if err != nil {
log.Fatalf("net.Connect err: %v", err)
}
defer conn.Close()
streamClient = pb.NewStreamClientClient(conn)
routeList()
}`
运行客户端
Copy
go run client.go code:200 value:"hello grpc" value:"ok"
服务端不断从客户端获取到数据
Copy
stream client rpc 0 stream client rpc 1 stream client rpc 2 stream client rpc 3 stream client rpc 4
思考#
服务端在没有接受完消息时候能主动停止接收数据吗（很少有这种场景）？
答案：可以的，但是客户端代码需要注意EOF判断
1.我们把服务端的RouteList方法实现稍微修改，当接收到一条数据后马上调用SendAndClose()关闭stream.
Copy
`func (s *SimpleService) RouteList(srv pb.StreamClient_RouteListServer) error {
for {
res, err := srv.Recv()
if err == io.EOF {
return srv.SendAndClose(&pb.SimpleResponse{Value: "ok"})
}
if err != nil {
return err
}
log.Println(res.StreamData)
return srv.SendAndClose(&pb.SimpleResponse{Value: "ok"})
}
}`
2.再把客户端调用RouteList方法的实现稍作修改
Copy
`func routeList() {
stream, err := streamClient.RouteList(context.Background())
if err != nil {
log.Fatalf("Upload list err: %v", err)
}
for n := 0; n < 5; n++ {
err := stream.Send(&pb.StreamRequest{StreamData: "stream client rpc " + strconv.Itoa(n)})
if err == io.EOF {
break
}
if err != nil {
log.Fatalf("stream request err: %v", err)
}
}
res, err := stream.CloseAndRecv()
if err != nil {
log.Fatalf("RouteList get response err: %v", err)
}
log.Println(res)
}`
客户端Send()需要检测err是否为EOF，因为当服务端在消息没接收完前主动调用SendAndClose()关闭stream，若此时客户端继续执行Send()，则会返回EOF错误。
总结#
本篇介绍了客户端流式RPC的简单使用，下篇将介绍双向流式RPC。
教程源码地址：https://github.com/Bingjian-Zhu/go-grpc-example
参考：gRPC官方文档中文版
