// 定義流式請(qǐng)求信息
message StreamRequest{
    //流請(qǐng)求參數(shù)
    string question = 1;
}

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

service Stream{
    // 雙向流式rpc，同時(shí)在請(qǐng)求參數(shù)前和響應(yīng)參數(shù)前加上stream
    rpc Conversations(stream StreamRequest) returns(stream StreamResponse){};
}

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

// StreamService 定義我們的服務(wù)
type StreamService struct{}
// Conversations 實(shí)現(xiàn)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)
	}
}

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í)例
	grpcServer := grpc.NewServer()
	// 在gRPC服務(wù)器注冊(cè)我們的服務(wù)
	pb.RegisterStreamServer(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...

// conversations 調(diào)用服務(wù)端的Conversations方法
func conversations() {
	//調(diào)用服務(wù)端的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)
	}
	//最后關(guān)閉流
	err = stream.CloseSend()
	if err != nil {
		log.Fatalf("Conversations close stream err: %v", err)
	}
}

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

var streamClient pb.StreamClient

func main() {
	// 連接服務(wù)器
	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()
}