- 由于是基于http2实现的,http2又要求必须是https协议,所以与https大致相同,建立连接时除了三次握手外还存在一次tls握手
- 连接建立后会发起一个连接前言请求(Magic帧 Setting帧),
- 通过Win_up滑动窗口传输数据
- 基于Header/Data帧进行传播和设置操作
- grpc会将数据存放到Data中, Data通常采用protobuf,需要设置IDL结构体,编解码数据进行传输
- 请求执行完毕后通过四次挥手断开连接
HEADERS (flags = END_HEADERS)
:method : POST // 请求的方法
:scheme : http // http/https
:path : /Comment/AddComment // 微服务路径 /{服务名}/{服务方法}
:authority : baidu.com // 目标 URL 的虚拟主机名
te : trailers // gRPC 中 这个值必须为 trailers
grpc-timeout : 15 // 调用的超时时间
Content-Type : application/grpc // grpc 必须以 applaction/grpc 开头
grpc-encoding : gzip // 压缩类型
authorization : *****
HEADERS (flags = END_HEADERS)
:status : 200
grpc-encoding : gzip
Content-Type: application/grpc
END_STREAM 也 称作 EOS(end of stream)
HEADERS (flags = END_STREAM, END_HEADERS)
grpc-status : 0 // OK / grcp 状态码
grpc-message : **** // 对错误的描述
go mod export GO111MODULE=on
go mod export GOPROXY=https://goproxy.io
go get google.golang.org/grpc
go install google.golang.org/protobuf/cmd/protoc-gen-go
go install google.golang.org/grpc/cmd/protoc-gen-go-grpc
go get google.golang.org/grpc/cmd/protoc-gen-go-grpc
go get -u github.com/golang/prorobuf/protoc-gen-go
//todo 指定生成proto时用的语法版本
syntax = "proto3";
//当前文件所在包
package test;
//todo 指定生成的go文件所在目录包,"."表示在当前目录生成，"proto"表示生成的go文件的包名是proto
option go_package = ".;proto";
//todo 定义请求参数类型
message TestRequest {
//参数类型 变量名=标号
string message = 1;
}
//todo 定义响应参数类型
message TestResponse {
string message = 1;
int32 code = 2;
Corpus corpus = 4;
}
//todo 定义枚举
enum Corpus {
UNIVERSAL = 0;
WEB = 1;
IMAGES = 2;
LOCAL = 3;
NEWS = 4;
PRODUCTS = 5;
VIDEO = 6;
}
//todo 定义server
service MyGrpc {
//todo 定义接口
//通信协议 方法名(入参类型) returns (反参类型){}
rpc MethodOne(TestRequest) returns (TestResponse) {}
//定义以流式响应的接口
rpc ServerStreaming(TestRequest) returns (stream TestResponse) {}
//定义以流式接收数据的接口
rpc ClientStreaming(stream TestRequest) returns (TestResponse) {}
//定义双向数据流接口
rpc BidirectionalStreaming(stream TestRequest) returns (stream TestResponse) {}
}
protoc --go_out=. 文件名.proto
protoc --go-grpc_out=. 文件名.proto
protoc --go_out=plugins=grpc:. 文件名.proto
protoc --go-grpc_out=. my_test.proto
protoc --go_out=. my_test.proto
发现多生成了一个mustEmbedUnimplementedXxxx()接口,如果不需要该接口可以使用
“protoc --go-grpc_out=require_unimplemented_servers=false” 关闭,或者在grpc server实现结构体中匿名嵌入Unimplemented***Server结构体
func _MyGrpc_MethodOne_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(TestRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(MyGrpcServer).MethodOne(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: "/test.MyGrpc/MethodOne",
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(MyGrpcServer).MethodOne(ctx, req.(*TestRequest))
}
return interceptor(ctx, in, info, handler)
}
func _MyGrpc_ServerStreaming_Handler(srv interface{}, stream grpc.ServerStream) error {
m := new(TestRequest)
if err := stream.RecvMsg(m); err != nil {
return err
}
return srv.(MyGrpcServer).ServerStreaming(m, &myGrpcServerStreamingServer{stream})
}
type MyGrpc_ServerStreamingServer interface {
Send(*TestResponse) error
grpc.ServerStream
}
type myGrpcServerStreamingServer struct {
grpc.ServerStream
}
func (x *myGrpcServerStreamingServer) Send(m *TestResponse) error {
return x.ServerStream.SendMsg(m)
}
func _MyGrpc_ClientStreaming_Handler(srv interface{}, stream grpc.ServerStream) error {
return srv.(MyGrpcServer).ClientStreaming(&myGrpcClientStreamingServer{stream})
}
type MyGrpc_ClientStreamingServer interface {
SendAndClose(*TestResponse) error
Recv() (*TestRequest, error)
grpc.ServerStream
}
type myGrpcClientStreamingServer struct {
grpc.ServerStream
}
func (x *myGrpcClientStreamingServer) SendAndClose(m *TestResponse) error {
return x.ServerStream.SendMsg(m)
}
func (x *myGrpcClientStreamingServer) Recv() (*TestRequest, error) {
m := new(TestRequest)
if err := x.ServerStream.RecvMsg(m); err != nil {
return nil, err
}
return m, nil
}
func _MyGrpc_BidirectionalStreaming_Handler(srv interface{}, stream grpc.ServerStream) error {
return srv.(MyGrpcServer).BidirectionalStreaming(&myGrpcBidirectionalStreamingServer{stream})
}
type MyGrpc_BidirectionalStreamingServer interface {
Send(*TestResponse) error
Recv() (*TestRequest, error)
grpc.ServerStream
}
type myGrpcBidirectionalStreamingServer struct {
grpc.ServerStream
}
func (x *myGrpcBidirectionalStreamingServer) Send(m *TestResponse) error {
return x.ServerStream.SendMsg(m)
}
func (x *myGrpcBidirectionalStreamingServer) Recv() (*TestRequest, error) {
m := new(TestRequest)
if err := x.ServerStream.RecvMsg(m); err != nil {
return nil, err
}
return m, nil
}
// MyGrpc_ServiceDesc is the grpc.ServiceDesc for MyGrpc service.
// It's only intended for direct use with grpc.RegisterService,
// and not to be introspected or modified (even as a copy)
var MyGrpc_ServiceDesc = grpc.ServiceDesc{
ServiceName: "test.MyGrpc",
HandlerType: (*MyGrpcServer)(nil),
Methods: []grpc.MethodDesc{
{
MethodName: "MethodOne",
Handler: _MyGrpc_MethodOne_Handler,
},
},
Streams: []grpc.StreamDesc{
{
StreamName: "ServerStreaming",
Handler: _MyGrpc_ServerStreaming_Handler,
ServerStreams: true,
},
{
StreamName: "ClientStreaming",
Handler: _MyGrpc_ClientStreaming_Handler,
ClientStreams: true,
},
{
StreamName: "BidirectionalStreaming",
Handler: _MyGrpc_BidirectionalStreaming_Handler,
ServerStreams: true,
ClientStreams: true,
},
},
Metadata: "my_test.proto",
}
//客户端结构体
type myGrpcClient struct {
cc grpc.ClientConnInterface
}
//创建客户端结构体方法
func NewMyGrpcClient(cc grpc.ClientConnInterface) MyGrpcClient {
return &myGrpcClient{cc}
}
func (c *myGrpcClient) MethodOne(ctx context.Context, in *TestRequest, opts ...grpc.CallOption) (*TestResponse, error) {
out := new(TestResponse)
err := c.cc.Invoke(ctx, "/test.MyGrpc/MethodOne", in, out, opts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *myGrpcClient) ServerStreaming(ctx context.Context, in *TestRequest, opts ...grpc.CallOption) (MyGrpc_ServerStreamingClient, error) {
stream, err := c.cc.NewStream(ctx, &MyGrpc_ServiceDesc.Streams[0], "/test.MyGrpc/ServerStreaming", opts...)
if err != nil {
return nil, err
}
x := &myGrpcServerStreamingClient{stream}
if err := x.ClientStream.SendMsg(in); err != nil {
return nil, err
}
if err := x.ClientStream.CloseSend(); err != nil {
return nil, err
}
return x, nil
}
type MyGrpc_ServerStreamingClient interface {
Recv() (*TestResponse, error)
grpc.ClientStream
}
type myGrpcServerStreamingClient struct {
grpc.ClientStream
}
func (x *myGrpcServerStreamingClient) Recv() (*TestResponse, error) {
m := new(TestResponse)
if err := x.ClientStream.RecvMsg(m); err != nil {
return nil, err
}
return m, nil
}
func (c *myGrpcClient) ClientStreaming(ctx context.Context, opts ...grpc.CallOption) (MyGrpc_ClientStreamingClient, error) {
stream, err := c.cc.NewStream(ctx, &MyGrpc_ServiceDesc.Streams[1], "/test.MyGrpc/ClientStreaming", opts...)
if err != nil {
return nil, err
}
x := &myGrpcClientStreamingClient{stream}
return x, nil
}
type MyGrpc_ClientStreamingClient interface {
Send(*TestRequest) error
CloseAndRecv() (*TestResponse, error)
grpc.ClientStream
}
type myGrpcClientStreamingClient struct {
grpc.ClientStream
}
func (x *myGrpcClientStreamingClient) Send(m *TestRequest) error {
return x.ClientStream.SendMsg(m)
}
func (x *myGrpcClientStreamingClient) CloseAndRecv() (*TestResponse, error) {
if err := x.ClientStream.CloseSend(); err != nil {
return nil, err
}
m := new(TestResponse)
if err := x.ClientStream.RecvMsg(m); err != nil {
return nil, err
}
return m, nil
}
func (c *myGrpcClient) BidirectionalStreaming(ctx context.Context, opts ...grpc.CallOption) (MyGrpc_BidirectionalStreamingClient, error) {
stream, err := c.cc.NewStream(ctx, &MyGrpc_ServiceDesc.Streams[2], "/test.MyGrpc/BidirectionalStreaming", opts...)
if err != nil {
return nil, err
}
x := &myGrpcBidirectionalStreamingClient{stream}
return x, nil
}
type MyGrpc_BidirectionalStreamingClient interface {
Send(*TestRequest) error
Recv() (*TestResponse, error)
grpc.ClientStream
}
type myGrpcBidirectionalStreamingClient struct {
grpc.ClientStream
}
func (x *myGrpcBidirectionalStreamingClient) Send(m *TestRequest) error {
return x.ClientStream.SendMsg(m)
}
func (x *myGrpcBidirectionalStreamingClient) Recv() (*TestResponse, error) {
m := new(TestResponse)
if err := x.ClientStream.RecvMsg(m); err != nil {
return nil, err
}
return m, nil
}
import (
"context"
"fmt"
"google.golang.org/grpc"
"google.golang.org/grpc/metadata"
"io"
"log"
"net"
)
// 1.创建服务结构体
type Server struct {
}
// 2.实现所有接口
// 2.1实现一元接口(普通接口)示例
func (s *Server) MethodOne(ctx context.Context, in *TestRequest) (*TestResponse, error) {
fmt.Printf("--- UnaryEcho ---\n")
md, ok := metadata.FromIncomingContext(ctx)
if !ok {
log.Println("miss metadata from context")
}
fmt.Println("md", md)
fmt.Printf("request received: %v, sending echo\n", in)
return &TestResponse{Message: in.Message}, nil
}
// 2.2流式响应接口实现
func (s *Server) ServerStreaming(in *TestRequest, stream MyGrpc_ServerStreamingServer) error {
fmt.Printf("--- ServerStreamingEcho ---\n")
fmt.Printf("request received: %v\n", in)
// Read requests and send responses.
for i := 0; i < 10; i++ {
fmt.Printf("echo message %v\n", in.Message)
//stream.Send()发送数据
err := stream.Send(&TestResponse{Message: in.Message})
if err != nil {
return err
}
}
return nil
}
// 2.3流式接收接口实现
func (s *Server) ClientStreaming(stream MyGrpc_ClientStreamingServer) error {
fmt.Printf("--- ClientStreamingEcho ---\n")
// Read requests and send responses.
var message string
for {
//stream.Recv()接收数据
in, err := stream.Recv()
if err == io.EOF {
fmt.Printf("echo last received message\n")
return stream.SendAndClose(&TestResponse{Message: message})
}
message = in.Message
fmt.Printf("request received: %v, building echo\n", in)
if err != nil {
return err
}
}
}
// 2.4双向数据流接口实现
func (s *Server) BidirectionalStreaming(stream MyGrpc_BidirectionalStreamingServer) error {
fmt.Printf("--- BidirectionalStreamingEcho ---\n")
// Read requests and send responses.
for {
//stream.Recv()接收数据
in, err := stream.Recv()
if err == io.EOF {
return nil
}
if err != nil {
return err
}
fmt.Printf("request received %v, sending echo\n", in)
//stream.Send()发送数据
if err := stream.Send(&TestResponse{Message: in.Message}); err != nil {
return err
}
}
}
// 不太清除这个接口是做什么的,假设不想生成这个接口可以执行 "protoc --go-grpc_out=require_unimplemented_servers=false" 关闭,
// 或者实现grpcServer的结构体中匿名嵌入UnimplementedXXXServer这个base结构体
func (s *Server) mustEmbedUnimplementedMyGrpcServer() {}
// 3.main方法启动服务
func main() {
//1.监听tcp,指定端口
lis, err := net.Listen("tcp", ":8080")
if nil != err {
log.Fatalf("listen: %v", err)
}
//2.创建grpc服务器
s := grpc.NewServer()
//3.将服务实现注册到grpc服务器
RegisterMyGrpcServer(s, &Server{})
//4.监听启动服务
s.Serve(lis)
}
import (
"context"
"fmt"
"io"
"log"
"sync"
"time"
"google.golang.org/grpc"
"google.golang.org/grpc/metadata"
)
// 请求服务端普通接口示例
func MethodOneCallWithMetadata(c MyGrpcClient, message string) {
//1.设置请求头
md := metadata.Pairs("timestamp", time.Now().Format(time.StampNano))
md.Append("authorization", "Bearer AccessToken")
//2.创建ctx
ctx := metadata.NewOutgoingContext(context.Background(), md)
//3.设置请求参数,并调用生成的指定方法MethodOne()请求服务端
r, err := c.MethodOne(ctx, &TestRequest{Message: message})
if err != nil {
log.Fatalf("failed to call UnaryEcho: %v", err)
}
fmt.Printf("response:%v\n", r.Message)
}
// 请求服务端流式响应接口示例
func serverStreamingWithMetadata(c MyGrpcClient, message string) {
//1.设置请求头
md := metadata.Pairs("timestamp", time.Now().Format(time.StampNano))
md.Append("authorization", "Bearer AccessToken")
//2.创建ctx
ctx := metadata.NewOutgoingContext(context.Background(), md)
//3.服务端流式响应,调用生成的流式响应方法,构建流式响应stream
stream, err := c.ServerStreaming(ctx, &TestRequest{Message: message})
if err != nil {
log.Fatalf("failed to call ServerStreamingEcho: %v", err)
}
//4.遍历接收服务端响应
// Read all the responses.
var rpcStatus error
for {
//stream.Recv()接收流中的数据
r, err := stream.Recv()
if err != nil {
rpcStatus = err
break
}
fmt.Printf(" - %s\n", r.Message)
}
if rpcStatus != io.EOF {
log.Fatalf("failed to finish server streaming: %v", rpcStatus)
}
}
// 流式请求服务端接口示例
func clientStreamWithMetadata(c MyGrpcClient, message string) {
//1.设置请求头
md := metadata.Pairs("timestamp", time.Now().Format(time.StampNano))
md.Append("authorization", "Bearer AccessToken")
//2.创建ctx
ctx := metadata.NewOutgoingContext(context.Background(), md)
//3.客户端流式发送数据到服务端,调用生成的流式方法构建流式数据stream
stream, err := c.ClientStreaming(ctx)
if err != nil {
log.Fatalf("failed to call ClientStreamingEcho: %v\n", err)
}
//4.以流形式多次向服务端发送数据
for i := 0; i < 10; i++ {
//5.基于流式结构发送数据到服务端
if err := stream.Send(&TestRequest{Message: message}); err != nil {
log.Fatalf("failed to send streaming: %v\n", err)
}
}
//6.接收服务端响应并关闭连接
r, err := stream.CloseAndRecv()
if err != nil {
log.Fatalf("failed to CloseAndRecv: %v\n", err)
}
fmt.Printf("response:%v\n", r.Message)
}
// 双向数据流接口示例
func bidirectionalWithMetadata(c MyGrpcClient, message string) {
//1.设置请求头
md := metadata.Pairs("timestamp", time.Now().Format(time.StampNano))
md.Append("authorization", "Bearer AccessToken")
//2.创建ctx
ctx := metadata.NewOutgoingContext(context.Background(), md)
//3.客户端服务端双写流式通信,调用生成的方法创建双向流stream
stream, err := c.BidirectionalStreaming(ctx)
if err != nil {
log.Fatalf("failed to call BidirectionalStreamingEcho: %v\n", err)
}
//4.协程内部多次向服务端以流的形式发送数据
go func() {
// Send all requests to the server.
for i := 0; i < 10; i++ {
if err := stream.Send(&TestRequest{Message: message}); err != nil {
log.Fatalf("failed to send streaming: %v\n", err)
}
}
stream.CloseSend()
}()
//5..以流的形式接收服务端响应数据
var rpcStatus error
fmt.Printf("response:\n")
for {
r, err := stream.Recv()
if err != nil {
rpcStatus = err
break
}
fmt.Printf(" - %s\n", r.Message)
}
if rpcStatus != io.EOF {
log.Fatalf("failed to finish server streaming: %v", rpcStatus)
}
}
func main() {
//遍历执行模拟并发
wg := sync.WaitGroup{}
for i := 0; i < 1; i++ {
wg.Add(1)
go func() {
defer wg.Done()
//1.与服务端建立连接
conn, err := grpc.Dial("localhost:8080", grpc.WithInsecure())
if err != nil {
log.Fatalf("did not connect: %v", err)
}
//退出时的关闭
defer conn.Close()
//2.调用生成的创建client方法,
//基于生成的Client创建
//grpcClient连接,并绑定连接
c := NewMyGrpcClient(conn)
//3.请求服务端一元方法(普通方法)
//for i := 0; i < 100; i++ {
MethodOneCallWithMetadata(c, message)
time.Sleep(400 * time.Millisecond)
//}
//
//4.请求服务端,流式响应方法
serverStreamingWithMetadata(c, message)
time.Sleep(1 * time.Second)
//5.流式请求服务端接口
clientStreamWithMetadata(c, message)
time.Sleep(1 * time.Second)
//6.双向流式
bidirectionalWithMetadata(c, message)
}()
}
wg.Wait()
time.Sleep(1 * time.Second)
}
- 基于https建立连接时除了三次握手外还存在一次tls握手
- 连接建立后会发起一个连接前言请求(Magic帧 Setting帧),
- 通过Win_up滑动窗口传输数据
- 基于Header/Data帧进行传播和设置操作
- grpc会将数据存放到Data中, Data通常采用protobuf,需要设置IDL结构体,编解码数据进行传输
- 请求执行完毕后通过四次挥手断开连接
- 服务器端在接收到客户端的请求消息后，会发回一个响应的序列。这种多个响应所组成的序列也被称为“流”。在将所有的服务器端响应发送完毕之后，服务器端会以 trailer 元数据的形式将其状态发送给客户端，从而标记流的结束
- 客户端的 Go 语言实现中使用 Recv 方法从客户端流中检索消息，并且持续检索，直到流结束为止
- 客户端会发送多个请求给服务器端，而不再是单个请求,注意服务器端不一定要等到从客户端接收到所有消息后才发送响应,可以在接收到流中的一条消息或几条消息之后就发送响应，也可以在读取完流中的所有消息之后再发送响应