go gRPC 服务端、客户端双向流demo

syntax = "proto3";

option go_package = "./;pb";

package pb;

service Greeter{
    rpc AllStream (stream StreamReqData) returns (stream StreamResData) {}
}

// 请求数据格式
message StreamReqData{
    string data = 1;
}

// 返回数据格式
message StreamResData{
    string data = 1;
}

server/main.go:

package main

import (
	pb "demo01/pb"
	"log"
	"net"
	"sync"
	"time"

	"google.golang.org/grpc"
)

const (
	port = ":50051"
)

// 服务对象
type server struct {
	pb.UnimplementedGreeterServer
}

func (s *server) AllStream(allStr pb.Greeter_AllStreamServer) error{
	wg := sync.WaitGroup{}
	wg.Add(2)
	go func(){
		for {
			data, _ := allStr.Recv()
			log.Println(data)
		}
	}()

	go func(){
		for {
			allStr.Send(&pb.StreamResData{Data: "from server"})
			time.Sleep((time.Second))
		}
	}()
	
	wg.Wait()
	return nil

}

func main(){
	lis, err := net.Listen("tcp", port)	
	if err != nil {
		return 
	}
	// 创建一个grpc 服务器
	s := grpc.NewServer()
	// 注册事件
	pb.RegisterGreeterServer(s, &server{})
	// 处理链接
	s.Serve(lis)
}

client/main.go:

package main

import (
	"context"
	pb "demo01/pb"
	"log"
	"time"

	"google.golang.org/grpc"
	_ "google.golang.org/grpc/balancer/grpclb"
)
const (
	ADDRESS = "localhost:50051"
)

func main(){
	// 建立连接
	conn, err := grpc.Dial(ADDRESS, grpc.WithInsecure())
	if err != nil {
		return
	}
	defer conn.Close()
	// 通过刚才的链接 生成一个client对象
	c := pb.NewGreeterClient(conn)
	allStr, _ := c.AllStream(context.Background())
	go func(){
		for {
			data, _ := allStr.Recv()
			log.Println(data)
		}
	}()

	go func(){
		for {
			allStr.Send(&pb.StreamReqData{Data:"form client"})
			time.Sleep(time.Second)
		}
	}()

	select{}
	
}