上一篇写了一下rpc调用过程的实现方式，简单来说就是服务端把实现了接口的结构体对象进行反射，抽取方法，签名，保存，客户端调用的时候go-micro封请求数据，服务端接收到请求时，找到需要调用调用的对象和对应的方法，利用反射进行调用，返回数据。 但是没有说stream的实现方式，感觉单独写一篇帖子来说这个更好一些。上一篇帖子是基础，理解了上一篇，stream实现原理一点即破。先说一下使用方式，再说原理。
当前go-micro对 rpc 调用的方式大概如下：
普通的rpc调用 是这样：
1.连接服务器或者从缓存池得到连接
2.客户端 ->发送数据 -> 服务端接收
3.服务端 ->返回数据 -> 客户端处理数据
4.关闭连接或者把连接返回到缓存池
当前 rps stream的实现方式 是这样子:
1. 连接服务器
2. 客户端多次发送请求-> 服务端接收
3. 服务端多次返回数据-> 客户端处理数据
4. 关闭连接
当数据量比较大的时候我们可以用stream方式分批次传输数据。对于客户端还是服务端没有限制，我们可以根据自己的需要使用stream方式，使用方式也非常的简单，在定义接口的时候在参数或者返回值前面加上stream然后就可以多次进行传输了，使用的代码还是之前写的例子,代码都在github上：
比如我的例子中定义了两个使用stream的接口，一个只在返回值使用stream，另一个是在参数和返回值前都加上了stream，最终的使用方式没有区别
rpc Stream(model.SRequest) returns (stream model.SResponse) {}
rpc BidirectionalStream(stream model.SRequest) returns (stream model.SResponse) {}
看一下go-micro为我们生成的代码rpcapi.micro.go里，不要被吓到，生成了很多代码，但是没啥理解不了的
Server端
// Server API for Say service
type SayHandler interface {
// .... others
Stream(context.Context, *model.SRequest, Say_StreamStream) error
BidirectionalStream(context.Context, Say_BidirectionalStreamStream) error
}
type Say_StreamStream interface {
SendMsg(interface{}) error
RecvMsg(interface{}) error
Close() error
Send(*model.SResponse) error
}
type Say_BidirectionalStreamStream interface {
SendMsg(interface{}) error
RecvMsg(interface{}) error
Close() error
Send(*model.SResponse) error
Recv() (*model.SRequest, error)
}
// .... others
Client端
// Client API for Say service
type SayService interface {
//... others
Stream(ctx context.Context, in *model.SRequest, opts ...client.CallOption) (Say_StreamService, error)
BidirectionalStream(ctx context.Context, opts ...client.CallOption) (Say_BidirectionalStreamService, error)
}
type Say_StreamService interface {
SendMsg(interface{}) error
RecvMsg(interface{}) error
Close() error
Recv() (*model.SResponse, error)
}
type Say_BidirectionalStreamService interface {
SendMsg(interface{}) error
RecvMsg(interface{}) error
Close() error
Send(*model.SRequest) error
Recv() (*model.SResponse, error)
}
你会发现参数前面加了 Stream后，生成的代码会把你的参数变成一个接口，这个接口主要要的方法是
SendMsg(interface{}) error
RecvMsg(interface{}) error
Close() error
剩下的两个接口方法是根据你是发送还是接收生成的，如果有发送就会有Send(你的参数)，如果有接收会生成Rev() (你的参数， error),但这两个方法只是为了让你使用时方便，里面调用的还是SendMsg(interface)和RecvMsg(interface)方法，但是他们是怎么工作的，如何多次发送和接收传输的数据，是不是感觉很神奇。
我就以TsBidirectionalStream 方法为例开始分析，上一篇和再早之前的帖子已经说了服务端启动的时候都做了哪些操作，这里就不再赘述，
服务端的实现,很简单，不断的获取客户端发过来的数据，再给客户端一次一次的返回一些数据。
/*
模拟数据
*/
func (s *Say) BidirectionalStream(ctx context.Context, stream rpcapi.Say_BidirectionalStreamStream) error {
for {
req, err := stream.Recv()
if err == io.EOF {
break
}
if err != nil {
return err
}
for i := int64(0); i < req.Count; i++ {
if err := stream.Send(&model.SResponse{Value: []string {lib.RandomStr(lib.Random(3, 6))}}); err != nil {
return err
}
}
}
return nil
}
启动服务，服务开始监听客户端传过来的数据.....
客户端调用服务端方法：
// 调用
func TsBidirectionalStream(client rpcapi.SayService) {
rspStream, err := client.BidirectionalStream(context.Background())
if err != nil {
panic(err)
}
// send
go func() {
rspStream.Send(&model.SRequest{Count: 2})
rspStream.Send(&model.SRequest{Count: 5})
// close the stream
if err := rspStream.Close(); err != nil {
fmt.Println("stream close err:", err)
}
}()
// recv
idx := 1
for {
rsp, err := rspStream.Recv()
if err == io.EOF {
break
} else if err != nil {
panic(err)
}
fmt.Printf("test stream get idx %d data %v\n", idx, rsp)
idx++
}
fmt.Println("Read Value End")
}
当客户端在调用rpc的stream方法是要很得到stream
rspStream, err := client.BidirectionalStream(context.Background())
//
func (c *sayService) BidirectionalStream(ctx context.Context, opts ...client.CallOption) (Say_BidirectionalStreamService, error) {
req := c.c.NewRequest(c.name, "Say.BidirectionalStream", &model.SRequest{})
stream, err := c.c.Stream(ctx, req, opts...)
if err != nil {
return nil, err
}
return &sayServiceBidirectionalStream{stream}, nil
}
这个调用c.c.Stream(ctx, req, opts...)是关键，他的内部实现就是和服务器进行连接，然后返回一个stream，进行操作。
客户端：和服务端建立连接，返回Stream，进行接收和发送数据
服务端：接收客户端连接请求，利用反射找到相应的方法，组织Strem，传给方法，进行数据的发送和接收
建立连接的时候就是一次rpc调用，服务端接受连接，然后客户端发送一次调用，但是传输的是空数据，服务端利用反射找到具体的方法，组织stream，调用具体方法，利用这个连接，客户端和服务端进行多次通信。
![]()
