书接上文，我们继续实现剩余的两种方式--客户端流式 RPC、双向流式 RPC。
Client-side streaming RPC：客户端流式 RPC、
客户端流式 RPC，单向流，客户端通过流式发起多次 RPC 请求到服务端，服务端发起一次响应给客户端
Proto :
代码语言：javascript复制syntax = "proto3";
package proto;
message String {
string value = 1;
}
service HelloService {
rpc Hello (stream String) returns (String){};
}
server:
代码语言：javascript复制package main
import (
"google.golang.org/grpc"
"io"
"log"
"net"
pb "rpc/proto" // 设置引用别名
)
// HelloServiceImpl 定义我们的服务
type HelloServiceImpl struct{}
//实现Hello方法
func (p *HelloServiceImpl) Hello(stream pb.HelloService_HelloServer) error {
for {
resp, err := stream.Recv()
if err == io.EOF {
return stream.SendAndClose(&pb.String{Value:"say.hello"})
}
if err != nil {
return err
}
log.Printf("resp: %v", resp)
}
return nil
}
func main() {
// 新建gRPC服务器实例
grpcServer := grpc.NewServer()
// 在gRPC服务器注册我们的服务
pb.RegisterHelloServiceServer(grpcServer, new(HelloServiceImpl))
lis, err := net.Listen("tcp", ":1234")
if err != nil {
log.Fatal(err)
}
log.Println(" net.Listing...")
//用服务器 Serve() 方法以及我们的端口信息区实现阻塞等待，直到进程被杀死或者 Stop() 被调用
err = grpcServer.Serve(lis)
if err != nil {
log.Fatalf("grpcServer.Serve err: %v", err)
}
}
如上，我们对每一个 Recv 都进行了处理，当发现 io.EOF (流关闭) 后，需要通过 stream.SendAndClose 方法将最终的响应结果发送给客户端，同时关闭正在另外一侧等待的 Recv。
client:
代码语言：javascript复制package main
import (
"context"
"google.golang.org/grpc"
"log"
pb "rpc/proto" // 设置引用别名
)
// SayHello 调用服务端的 Hello 方法
func SayHello(client pb.HelloServiceClient, r *pb.String) error {
stream, _ := client.Hello(context.Background())
for n := 0; n < 6; n {
_ = stream.Send(r)
}
resp, _ := stream.CloseAndRecv()
log.Printf("resp err: %v", resp)
return nil
}
func main() {
conn, err := grpc.Dial("localhost:1234", grpc.WithInsecure())
if err != nil {
log.Fatal("dialing err:", err)
}
defer conn.Close()
// 建立gRPC连接
client := pb.NewHelloServiceClient(conn)
// 创建发送结构体
req := pb.String{
Value: "stream server grpc ",
}
SayHello(client, &req)
}
在 Server 端的 stream.CloseAndRecv，与 Client 端 stream.SendAndClose 是配套使用的方法。
开启服务器端，开启客户端。执行结果如下：
代码语言：javascript复制$ go run server.go
2021/11/17 13:26:34 net.Listing...
2021/11/17 13:26:44 resp: value:"stream server grpc "
2021/11/17 13:26:44 resp: value:"stream server grpc "
2021/11/17 13:26:44 resp: value:"stream server grpc "
2021/11/17 13:26:44 resp: value:"stream server grpc "
2021/11/17 13:26:44 resp: value:"stream server grpc "
2021/11/17 13:26:44 resp: value:"stream server grpc "
$ go run client.go
2021/11/17 13:26:44 resp err: value:"say.hello"
Bidirectional streaming RPC：双向流式 RPC
双向流式 RPC，由客户端以流式的方式发起请求，服务端也以流式的方式响应请求。
首个请求一定是 Client 发起，但具体交互方式（谁先谁后、一次发多少、响应多少、什么时候关闭）根据程序编写的方式来确定（可以结合协程）。
Proto :
代码语言：javascript复制syntax = "proto3";
package proto;
message String {
string value = 1;
}
service HelloService {
rpc Hello (stream String) returns (stream String){};
}
server:
代码语言：javascript复制package main
import (
"google.golang.org/grpc"
"io"
"log"
"net"
pb "rpc/proto" // 设置引用别名
)
// HelloServiceImpl 定义我们的服务
type HelloServiceImpl struct{}
//实现Hello方法
func (p *HelloServiceImpl) Hello(stream pb.HelloService_HelloServer) error {
for {
_ = stream.Send(&pb.String{Value: "say.hello"})
resp, err := stream.Recv()
//接收完了返回
if err == io.EOF {
return nil
}
if err != nil {
return err
}
log.Printf("resp: %v", resp)
}
}
func main() {
// 新建gRPC服务器实例
grpcServer := grpc.NewServer()
// 在gRPC服务器注册我们的服务
pb.RegisterHelloServiceServer(grpcServer, new(HelloServiceImpl))
lis, err := net.Listen("tcp", ":1234")
if err != nil {
log.Fatal(err)
}
log.Println(" net.Listing...")
err = grpcServer.Serve(lis)
if err != nil {
log.Fatalf("grpcServer.Serve err: %v", err)
}
}
client:
代码语言：javascript复制package main
import (
"context"
"google.golang.org/grpc"
"io"
"log"
pb "rpc/proto" // 设置引用别名
)
// SayHello 调用服务端的 Hello 方法
func SayHello(client pb.HelloServiceClient, r *pb.String) error {
stream, _ := client.Hello(context.Background())
for n := 0; n <= 3; n {
_ = stream.Send(r)
resp, err := stream.Recv()
if err == io.EOF {
break
}
if err != nil {
return err
}
log.Printf("resp err: %v", resp)
}
_ = stream.CloseSend()
return nil
}
func main() {
conn, err := grpc.Dial("localhost:1234", grpc.WithInsecure())
if err != nil {
log.Fatal("dialing err:", err)
}
defer conn.Close()
// 建立gRPC连接
client := pb.NewHelloServiceClient(conn)
// 创建发送结构体
req := pb.String{
Value: "stream server grpc ",
}
SayHello(client, &req)
}
❝服务端在循环中接收客户端发来的数据，如果遇到io.EOF表示客户端流被关闭，如果函数退出表示服 务端流关闭。生成返回的数据通过流发送给客户端，双向流数据的发送和接收都是完全独立的行为。需 要注意的是，发送和接收的操作并不需要一一对应，用户可以根据真实场景进行组织代码。 ❞
开启服务器端，开启客户端。执行结果如下：
代码语言：javascript复制$ go run server.go
2021/11/17 15:46:10 net.Listing...
2021/11/17 15:46:19 resp: value:"stream server grpc "
2021/11/17 15:46:19 resp: value:"stream server grpc "
2021/11/17 15:46:19 resp: value:"stream server grpc "
2021/11/17 15:46:19 resp: value:"stream server grpc "
$ go run client.go
2021/11/17 15:46:19 resp err: value:"say.hello"
2021/11/17 15:46:19 resp err: value:"say.hello"
2021/11/17 15:46:19 resp err: value:"say.hello"
2021/11/17 15:46:19 resp err: value:"say.hello"
图片及部分相关技术知识点来源于网络搜索，侵权删！
参考资料：
https://golang2.eddycjy.com/posts/ch3/03-simple-grpc/
https://www.cnblogs.com/FireworksEasyCool/p/12693749.html
