Go-grpc实现

什么是grpc和protobuf

grpc

? grpc是一个Google开源的高性能、开源和通用的RPC框架，面向移动和HTTP/2设计。目前提供C、Java和Go语言版本, 分别是grpc, grpc-java 和 grpc-go, 其中C语言版本又支持C , C++,Node.js, Python, Ruby, Object-C, PHP, C#

grpc协议使用的序列化程序不是json 、xml 等， 而是使用的protobuf序列化及反序列化

protobuf

• 习惯使用json、xml交互数据的人，大多没有听说过 Protocol Buffer
• Protocol Buffer其实是Google出品的一种轻量&高效的结构化数据存储格式，性能要比json、xml强很多
• 目前主流使用的protobuf3

go get -d google.golang.org/protobuf/cmd/protoc-gen-go

安装好protoc执行程序后 写入.proto文件

syntax = "proto3";option go_package = ".;proto";service Greeter{  rpc SayHello (HelloRequest) returns (HelloReply);}message HelloRequest {  string name = 1;}message HelloReply {  string message = 1;}

写完之后 执行命令生成.go文件protoc -I . first.proto --go_out=plugins=grpc:.

• Server端代码

package mainimport (	"context"	"log"	"net"	"google.golang.org/grpc"	"google.golang.org/grpc/reflection"	pb "AwesomeMicroPro/MicroProject/proto")// server is used to implement pb.GreeterServer .type server struct {	pb.UnimplementedGreeterServer}// SayHello 方法接受远程调用方法func (s *server) SayHello(ctx context.Context, request *pb.HelloRequest) (*pb.HelloReply, error) {	log.Println(request.Name + "remote 1 first")	return &pb.HelloReply{Message: "hello " + request.Name}, nil}func main() {	g := grpc.NewServer()	reflection.Register(g)	pb.RegisterGreeterServer(g, &server{})	listen, err := net.Listen("tcp", "127.0.0.1:8080")	if err != nil {		panic("Listen Port Failed"+err.Error())	}	err = g.Serve(listen)	if err != nil {		panic("Server Enable Failed"+err.Error())	}}

• Client端代码

package mainimport (	"context"	"google.golang.org/grpc"	"log"	"os"	pb "AwesomeMicroPro/MicroProject/proto")const (	address = "127.0.0.1:8080")func main() {	// 拨号连接Server	conn, err := grpc.Dial(address, grpc.WithInsecure())	if err != nil {		panic("Connect grpc Server Failed" + err.Error())	}	// 最后关闭连接	defer conn.Close()	client := pb.NewGreeterClient(conn)	name := "testing ***"	if len(os.Args) > 1 {		name = os.Args[1]	}	// 远程调用SayHello方法传入对应的值	reply, err := client.SayHello(context.Background(), &pb.HelloRequest{Name: name})	if err != nil {		panic("fun remote execute failed" + err.Error())	}	log.Println(reply.Message)}

rpc的四种模式

• 简单模式
  • 简单模式最为传统 和熟悉的数据流模式没有任何区别，即客户端请求一次数据 服务端相应一个数据
• 服务端数据流模式
  • 客户端发起一次请求 服务端连续不断的返回数据流 stream， 典型的例子就是客户端发起一个股票代码 服务端实时的将股票的信息返回到客户端
• 客户端数据流模式
  • 客户端源源不断的向服务端发送数据流 ，发送结束后 服务端返回一个响应数据， 例子 ： 物联网终端向服务器报告数据
• 双向数据流模式
  • 客户端和服务端都可以发送数据流 、实时交互。 例子： 聊天

服务端数据流、客户端数据流、双向数据流 模式

stream.proto

syntax = "proto3";option go_package = "./;proto";service Greeter  {  rpc GetStream (StreamReqData) returns (stream StreamResData) {} // 服务端流模式  rpc PutStream (stream StreamReqData) returns (StreamResData) {} // 客户端流模式  rpc AllStream (stream StreamResData) returns (stream StreamReqData) {} // 双向流模式}message StreamReqData {  string data = 1;}message StreamResData {  string data = 2;}

protoc -I . stream.proto --go_out=plugins=grpc:.

server.go

package mainimport (	"AwesomeMicroPro/stream_grpc_test/proto"	"fmt"	"google.golang.org/grpc"	"log"	"net"	"sync"	"time")const PORT = ":50052"var wg sync.WaitGrouptype server struct {}// GetStream 服务端流模式 源源不断的发送数据给客户端func (s *server) GetStream(data *proto.StreamReqData, res proto.Greeter_GetStreamServer) error {	for i := 0; i < 10; i++ {		_ = res.Send(&proto.StreamResData{			Data: fmt.Sprintf("%v", time.Now().Unix()),		})		time.Sleep(time.Second)	}	return nil}// PutStream 客户端流模式  源源不断的接受客户端的请求func (s *server) PutStream(putStreamServer proto.Greeter_PutStreamServer) error {	for i:=0; i< 10; i ++{		recv, err := putStreamServer.Recv()		if err != nil {			log.Println("Recv Data Failed"+ err.Error())			break		}		fmt.Println(recv)	}	return nil}func (s *server) AllStream(allstream proto.Greeter_AllStreamServer) error {	wg.Add(2)	go func() {		for {			recv, err := allstream.Recv()			if err != nil {				log.Println("接受数据失败")				break			}			fmt.Println("服务端接收到的数据:"+recv.Data)		}		defer wg.Done()	}()	go func() {		for  {			err := allstream.Send(&proto.StreamReqData{				Data: fmt.Sprintf("我是服务端%v", time.Now().Unix()),			})			if err != nil {				log.Println("服务端发送数据失败" + err.Error())				break			}			time.Sleep(time.Second)		}		wg.Done()	}()	wg.Wait()	return nil}func main() {	listen, err := net.Listen("tcp", PORT)	if err != nil {		log.Println("Listen tcp port failed" + err.Error())	}	newServer := grpc.NewServer()	proto.RegisterGreeterServer(newServer, &server{})	err = newServer.Serve(listen)}

Client.go

package mainimport (	"AwesomeMicroPro/stream_grpc_test/proto"	"context"	"fmt"	"google.golang.org/grpc"	"log"	"sync"	"time")var wg sync.WaitGroupfunc main() {	conn, err := grpc.Dial("localhost:50052", grpc.WithInsecure())	if err != nil {		log.Println("connect grpc service failed" + err.Error())	}	defer conn.Close()	client := proto.NewGreeterClient(conn)	// GetStream 服务端流模式  适合订阅监控某些信息	stream, err := client.GetStream(context.Background(), &proto.StreamReqData{Data: "服务端数据流调用"})	for i:=0; i < 10; i++{		recv, err := stream.Recv()		if err != nil {			log.Println(err.Error())		}		log.Println(recv)	}	// 客户端流模式	putStream, err := client.PutStream(context.Background())	for i:=0; i < 10; i ++ {		err := putStream.Send(&proto.StreamReqData{Data: "**da **" + string(i)})		if err != nil {			log.Println("err" + err.Error())			break		}	}	// 双向流模式	allStream, err := client.AllStream(context.Background())	wg.Add(2)	go func() {		for {			recv, err := allStream.Recv()			if err != nil {				log.Println("接收数据失败")				break			}			fmt.Println("客户端接收到的数据: " + recv.Data)		}		defer wg.Done()			}()	go func() {		for {			err := allStream.Send(&proto.StreamResData{Data: fmt.Sprintf("我是客户端 %v", time.Now().Unix())})			if err != nil {				log.Println("客户端发送数据失败"+err.Error())				break			}			time.Sleep(time.Second)		}		wg.Done()	}()	wg.Wait()}