• rpc server

type HelloWorld struct {
}

func (H *HelloWorld) HelloWorld(req string, resp *string) error {
	*resp = req + "你好,Sakura"
	return nil
}

func RpcServer() {
	// 1. 创建服务
	err := rpc.RegisterName("Hello", &HelloWorld{})
	if err != nil {
		log.Println("注册RPC服务失败:", err)
		return
	}

	// 2.设置监听
	listener, err := net.Listen("tcp", "127.0.0.1:8004")
	if err != nil {
		log.Println("监听失败:", err)
	}
	log.Println("监听端口 127.0.0.1:8004")

	// 3.建立连接
	conn, err := listener.Accept()
	if err != nil {
		log.Println("Accept err:", err)
	}
	fmt.Println("连接已建立")
	// 4.绑定服务,将连接和服务进行绑定
	rpc.ServeConn(conn)
}

• rpc client

func RpcClient() {
	// 1.用rpc链接服务器 --Dial(拨号)
	conn, err := rpc.Dial("tcp", "127.0.0.1:8004")
	if err != nil {
		log.Println("Dial err:", err)
	}
	defer conn.Close()

	// 2.调用远程函数
	var response string // 接收返回值
	// 指定调用的远程函数
	err = conn.Call("Hello.HelloWorld", "Sakura", &response)
	if err != nil {
		log.Fatalln("Call err:", err)
	}
	fmt.Println(response)
}

• Server

func GRPCSever() {
	// 1.注册服务,启动监听
	grpcserver := grpc.NewServer()
	pb.RegisterHelloServer(grpcserver, &HelloService{})

	// 2.监听
	listen, err := net.Listen("tcp", "127.0.0.1:8005")
	if err != nil {
		log.Println("监听失败,", err)
	}
	defer listen.Close()

	// 3.启动服务
	grpcserver.Serve(listen)
}

// 定义结构实现具体的业务
// 需要嵌入这个结构体
type HelloService struct {
	pb.UnimplementedHelloServer
}

// 重写未定义的方法
func (HelloService) Hello(context.Context, *pb.Person) (*pb.Person, error) {
	person := &pb.Person{
		Name: "Sakura",
		Age:  22,
	}
	return person, nil
}

• Client

func GRPCClient() {
	// 1.连接gRPC服务
	//grpcConn, err := grpc.Dial("127.0.0.1:8004")
	// 抑制安全策略
	grpcConn, err := grpc.Dial("127.0.0.1:8005", grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		fmt.Println("grpc Dial error:", err)
		return
	}
	defer grpcConn.Close()

	// 2.初始化客户端
	grpcClient := pb.NewHelloClient(grpcConn)

	// 3.调用远程服务（函数）
	reply, err := grpcClient.Hello(context.Background(), &pb.Person{Name: "李四", Age: 18})
	if err != nil {
		fmt.Println("reply error:", err)
		return
	}
	fmt.Println("reply:", reply)
}

• Server

var port = flag.Int("port", 8006, "Grpc server port")

func GrpcServer() {
	flag.Parse()
	// 监听端口
	listen, err := net.Listen("tcp", fmt.Sprintf(":%d", *port))
	if err != nil {
		log.Println("监听失败,", err)
	}
	server := grpc.NewServer()

	proto.RegisterEchoServer(server, &Servers{})

	server.Serve(listen)
}

type Servers struct {
	proto.UnimplementedEchoServer
}

// UnaryEcho 一元RPC服务的实现
// 元数据,map[string][string]
// token,timestamp,授权
func (s *Servers) UnaryEcho(ctx context.Context, req *proto.EchoRequest) (*proto.EchoResponse, error) {
	fmt.Println("----------UnaryEcho-----------")
	incomingContext, ok := metadata.FromIncomingContext(ctx)
	if !ok {
		log.Println("miss metadata from context")
	}
	fmt.Println("metadata:", incomingContext)
	return &proto.EchoResponse{Message: req.Message}, nil
}

• Client

var msg = "Client Data"

func GrpcClient() {
	conn, err := grpc.Dial("127.0.0.1:8006", grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		log.Println(err)
	}
	defer conn.Close()

	client := proto.NewEchoClient(conn)

	// 1.一元RPC
	unaryEchoWithMetadata(client, msg)
}

// 调用一元RPC方法
func unaryEchoWithMetadata(client proto.EchoClient, msg string) {
	fmt.Println("---------UnaryEcho---------")

	// Pairs封装一个metadata
	md := metadata.Pairs("timestamp", time.Now().Format(time.StampNano))
	//md.Append("authorization", "token....")
	ctx := metadata.NewOutgoingContext(context.Background(), md) // 即将输出的请求

	resp, err := client.UnaryEcho(ctx, &proto.EchoRequest{Message: msg})
	if err != nil {
		fmt.Println(err)
	} else {
		fmt.Println("Grpc Server Response:", resp)
	}
}

• Server

var port = flag.Int("port", 8006, "Grpc server port")

func GrpcServer() {
	flag.Parse()
	// 监听端口
	listen, err := net.Listen("tcp", fmt.Sprintf(":%d", *port))
	if err != nil {
		log.Println("监听失败,", err)
	}
	server := grpc.NewServer()

	proto.RegisterEchoServer(server, &Servers{})

	server.Serve(listen)
}

type Servers struct {
	proto.UnimplementedEchoServer
}

// 服务端流式处理RPC
func (s *Servers) ServerStreamingEcho(req *proto.EchoRequest, stream proto.Echo_ServerStreamingEchoServer) error {
	fmt.Println("----------ServerStreamingEcho-----------")

	// 服务端向客户单响应使用send函数
	for i := 0; i < 5; i++ { // 以流的形式发送多次
		err := stream.Send(&proto.EchoResponse{Message: req.Message})
		if err != nil {
			return err
		}
	}

	return nil
}

• Client

var msg = "Client Data"

func GrpcClient() {
	conn, err := grpc.Dial("127.0.0.1:8006", grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		log.Println(err)
	}
	defer conn.Close()

	client := proto.NewEchoClient(conn)

	// 1.调用一元RPC方法
	unaryEchoWithMetadata(client, msg)

	// 2.调用服务端流式处理
	serverStreaming(client, msg)
}

// 调用服务端流式处理
func serverStreaming(client proto.EchoClient, msg string) {
	fmt.Println("---------serverStreaming Client---------")

	// Pairs封装一个metadata
	md := metadata.Pairs("timestamp", time.Now().Format(time.StampNano))
	//md.Append("authorization", "token....")
	ctx := metadata.NewOutgoingContext(context.Background(), md) // 即将输出的请求

	stream, err := client.ServerStreamingEcho(ctx, &proto.EchoRequest{Message: msg})
	if err != nil {
		fmt.Println(err)
	}
	// 从流中接收数据,每次读一个消息
	for {
		// err 读取到末尾会返回 EOF
		recv, err := stream.Recv()
		if err != nil {
			log.Println(err)
			break
		} else if err == io.EOF {
			fmt.Println("finish receive")
			return
		}
		fmt.Println("response is :,", recv.Message)
	}
}

• Server

func GrpcServer() {
	flag.Parse()
	// 监听端口
	listen, err := net.Listen("tcp", fmt.Sprintf(":%d", *port))
	if err != nil {
		log.Println("监听失败,", err)
	}
	server := grpc.NewServer()

	proto.RegisterEchoServer(server, &Servers{})

	server.Serve(listen)
}

type Servers struct {
	proto.UnimplementedEchoServer
}

// ClientStreamingEcho 客户端流式处理
func (s *Servers) ClientStreamingEcho(stream proto.Echo_ClientStreamingEchoServer) error {
	fmt.Println("--------------ClientStreamingEcho------------------------")
	// 接收客户端消息
	for {
		recv, err := stream.Recv()
		if err == io.EOF {
			fmt.Println("receive finish")
			// 发送并关闭
			return stream.SendAndClose(&proto.EchoResponse{Message: "Receive Finish"})
		} else if err != nil {
			log.Println(err)
			return err
		}
		// 打印接收到的消息
		fmt.Println("request reveived: ", recv.Message)
	}
}

• Client

func GrpcClient() {
	conn, err := grpc.Dial("127.0.0.1:8006", grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		log.Println(err)
	}
	defer conn.Close()

	client := proto.NewEchoClient(conn)

	// 1.调用一元RPC方法
	unaryEchoWithMetadata(client, msg)

	// 2.调用服务端流式处理
	serverStreaming(client, msg)

	// 3.调用客户端流式处理
	clientStreaming(client, msg)
}

// 客户端流式处理
func clientStreaming(client proto.EchoClient, msg string) {
	fmt.Println("---------clientStreaming---------")

	// Pairs封装一个metadata
	md := metadata.Pairs("timestamp", time.Now().Format(time.StampNano))
	//md.Append("authorization", "token....")
	ctx := metadata.NewOutgoingContext(context.Background(), md) // 即将输出的请求

	stream, err := client.ClientStreamingEcho(ctx)
	if err != nil {
		fmt.Println(err)
	}
	// 向服务端循环发送消息
	for i := 0; i < 5; i++ {
		err = stream.Send(&proto.EchoRequest{Message: msg})
		if err != nil {
			log.Println("Fialed to send", err)
		}
	}
	// 关闭并接受
	recv, err := stream.CloseAndRecv()
	if err != nil {
		log.Println(err)
	}
	log.Println("Servere response:", recv)
}

• Server

func GrpcServer() {
	flag.Parse()
	// 监听端口
	listen, err := net.Listen("tcp", fmt.Sprintf(":%d", *port))
	if err != nil {
		log.Println("监听失败,", err)
	}
	server := grpc.NewServer()

	proto.RegisterEchoServer(server, &Servers{})

	server.Serve(listen)
}

type Servers struct {
	proto.UnimplementedEchoServer
}

// 双向流式处理
func (s *Servers) BidirectionalStreamingEcho(stream proto.Echo_BidirectionalStreamingEchoServer) error {
	fmt.Println("--------------BidirectionalStreamingEcho------------------------")
	// 每接收到客户端的消息,就向客户端发送响应
	for {
		recv, err := stream.Recv()
		if err == io.EOF {
			fmt.Println("消息发送完毕")
			stream.Send(&proto.EchoResponse{Message: "Receive over"})
			return nil
		} else if err != nil {
			log.Println("Receive Failed", err)
			return err
		}
		// 打印接收到消息
		log.Println("Client Message:", recv)
		// 回复响应
		err = stream.Send(&proto.EchoResponse{Message: "Receive success!"})
		if err != nil {
			log.Println("Send Fail,", err)
		}
	}
}

• Client

func GrpcClient() {
	conn, err := grpc.Dial("127.0.0.1:8006", grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		log.Println(err)
	}
	defer conn.Close()

	client := proto.NewEchoClient(conn)

	// 4.双向流式处理
	serverclientStreaming(client, msg)
}

// 双向流式处理
func serverclientStreaming(client proto.EchoClient, msg string) {
	fmt.Println("---------serverclientStreaming---------")

	// Pairs封装一个metadata
	md := metadata.Pairs("timestamp", time.Now().Format(time.StampNano))
	//md.Append("authorization", "token....")
	ctx := metadata.NewOutgoingContext(context.Background(), md) // 即将输出的请求

	stream, err := client.BidirectionalStreamingEcho(ctx)
	if err != nil {
		fmt.Println(err)
	}

	// 开启一个协程向服务端发送消息
	go func() {
		for i := 0; i < 5; i++ {
			err := stream.Send(&proto.EchoRequest{Message: msg})
			if err != nil {
				log.Println(err)
			}
		}
		stream.CloseSend()
	}()
	//time.Sleep(time.Second * 10)

	//循环接收服务器的消息
	for {
		recv, err1 := stream.Recv()
		if err1 == io.EOF {
			log.Println("----- Receive Finish ------ ")
			break
		} else if err != nil {
			log.Println("Receive Fail,", err)
		}
		log.Println("Server Response:", recv)
	}
}