// 定義發(fā)送請(qǐng)求信息
message SimpleRequest{
    // 定義發(fā)送的參數(shù)，采用駝峰命名方式，小寫加下劃線，如：student_name
    // 請(qǐng)求參數(shù)
    string data = 1;
}

// 定義流式響應(yīng)信息
message StreamResponse{
    // 流式響應(yīng)數(shù)據(jù)
    string stream_value = 1;
}

// 定義我們的服務(wù)（可定義多個(gè)服務(wù),每個(gè)服務(wù)可定義多個(gè)接口）
service StreamServer{
    // 服務(wù)端流式rpc，在響應(yīng)數(shù)據(jù)前添加stream
    rpc ListValue(SimpleRequest)returns(stream StreamResponse){};
}

protoc --go_out=plugins=grpc:./ ./server_stream.proto

// SimpleService 定義我們的服務(wù)
type StreamService struct{}
// ListValue 實(shí)現(xiàn)ListValue方法
func (s *StreamService) ListValue(req *pb.SimpleRequest, srv pb.StreamServer_ListValueServer) error {
	for n := 0; n < 5; n++ {
		// 向流中發(fā)送消息， 默認(rèn)每次send送消息最大長度為`math.MaxInt32`bytes
		err := srv.Send(&pb.StreamResponse{
			StreamValue: req.Data + strconv.Itoa(n),
		})
		if err != nil {
			return err
		}
	}
	return nil
}

const (
	// Address 監(jiān)聽地址
	Address string = ":8000"
	// Network 網(wǎng)絡(luò)通信協(xié)議
	Network string = "tcp"
)

func main() {
	// 監(jiān)聽本地端口
	listener, err := net.Listen(Network, Address)
	if err != nil {
		log.Fatalf("net.Listen err: %v", err)
	}
	log.Println(Address + " net.Listing...")
	// 新建gRPC服務(wù)器實(shí)例
	// 默認(rèn)單次接收最大消息長度為`1024*1024*4`bytes(4M)，單次發(fā)送消息最大長度為`math.MaxInt32`bytes
	// grpcServer := grpc.NewServer(grpc.MaxRecvMsgSize(1024*1024*4), grpc.MaxSendMsgSize(math.MaxInt32))
	grpcServer := grpc.NewServer()
	// 在gRPC服務(wù)器注冊(cè)我們的服務(wù)
	pb.RegisterStreamServerServer(grpcServer, &StreamService{})

	//用服務(wù)器 Serve() 方法以及我們的端口信息區(qū)實(shí)現(xiàn)阻塞等待，直到進(jìn)程被殺死或者 Stop() 被調(diào)用
	err = grpcServer.Serve(listener)
	if err != nil {
		log.Fatalf("grpcServer.Serve err: %v", err)
	}
}

go run server.go
:8000 net.Listing...

// listValue 調(diào)用服務(wù)端的ListValue方法
func listValue() {
	// 創(chuàng)建發(fā)送結(jié)構(gòu)體
	req := pb.SimpleRequest{
		Data: "stream server grpc ",
	}
	// 調(diào)用我們的服務(wù)(ListValue方法)
	stream, err := grpcClient.ListValue(context.Background(), &req)
	if err != nil {
		log.Fatalf("Call ListStr err: %v", err)
	}
	for {
		//Recv() 方法接收服務(wù)端消息，默認(rèn)每次Recv()最大消息長度為`1024*1024*4`bytes(4M)
		res, err := stream.Recv()
		// 判斷消息流是否已經(jīng)結(jié)束
		if err == io.EOF {
			break
		}
		if err != nil {
			log.Fatalf("ListStr get stream err: %v", err)
		}
		// 打印返回值
		log.Println(res.StreamValue)
	}
}

// Address 連接地址
const Address string = ":8000"

var grpcClient pb.StreamServerClient

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

	// 建立gRPC連接
	grpcClient = pb.NewStreamServerClient(conn)
	route()
	listValue()
}

// ListValue 實(shí)現(xiàn)ListValue方法
func (s *StreamService) ListValue(req *pb.SimpleRequest, srv pb.StreamServer_ListValueServer) error {
	for n := 0; n < 15; n++ {
		// 向流中發(fā)送消息， 默認(rèn)每次send送消息最大長度為`math.MaxInt32`bytes
		err := srv.Send(&pb.StreamResponse{
			StreamValue: req.Data + strconv.Itoa(n),
		})
		if err != nil {
			return err
		}
		log.Println(n)
		time.Sleep(1 * time.Second)
	}
	return nil
}

// listValue 調(diào)用服務(wù)端的ListValue方法
func listValue() {
	// 創(chuàng)建發(fā)送結(jié)構(gòu)體
	req := pb.SimpleRequest{
		Data: "stream server grpc ",
	}
	// 調(diào)用我們的服務(wù)(Route方法)
	// 同時(shí)傳入了一個(gè) context.Context ，在有需要時(shí)可以讓我們改變RPC的行為，比如超時(shí)/取消一個(gè)正在運(yùn)行的RPC
	stream, err := grpcClient.ListValue(context.Background(), &req)
	if err != nil {
		log.Fatalf("Call ListStr err: %v", err)
	}
	for {
		//Recv() 方法接收服務(wù)端消息，默認(rèn)每次Recv()最大消息長度為`1024*1024*4`bytes(4M)
		res, err := stream.Recv()
		// 判斷消息流是否已經(jīng)結(jié)束
		if err == io.EOF {
			break
		}
		if err != nil {
			log.Fatalf("ListStr get stream err: %v", err)
		}
		// 打印返回值
		log.Println(res.StreamValue)
		break
	}
	//可以使用CloseSend()關(guān)閉stream，這樣服務(wù)端就不會(huì)繼續(xù)產(chǎn)生流消息
	//調(diào)用CloseSend()后，若繼續(xù)調(diào)用Recv()，會(huì)重新激活stream，接著之前結(jié)果獲取消息
	stream.CloseSend()
}