protocol 和 grpc 的基本使用

开发运维 2023-07-14 Escape 手机阅读

protocol

标量数值类型

protocol buffer 类型对应各语言的类型：Scalar Value Types

int32 对于负值的效率很低，应该使用 sint32

默认值

对于 string 默认是空字符串

对于 bytes 默认是空切片

对于 bool 默认是 false

对于数值类型默认是 0

对于枚举类型默认是第一个定义的枚举值，必须为 0

 enum Corpus {
  UNIVERSAL = 0;
  WEB = 1;
  IMAGES = 2;
  LOCAL = 3;
  NEWS = 4;
  PRODUCTS = 5;
  VIDEO = 6;
}

如果如果要将不同的枚举常量定义相同的值，需要设置 allow_alias 为 true

 enum EnumAllowingAlias {
  option allow_alias = true;
  UNKNOWN = 0;
  STARTED = 1;
  RUNNING = 1;
}
enum EnumNotAllowingAlias {
  UNKNOWN = 0;
  STARTED = 1;
  // RUNNING = 1;  // Uncommenting this line will cause a compile error inside Google and a warning message outside.
}

option go_package 作用

option go_package 用来指明生成的包名，语法为：option go_package = ";";

比如：

"./;proto"，当前目录下，包名为 proto
"common/stream/proto/v1"，如果这些路径不存在会创建这些路径，包名为 v1
"common/stream/proto/v1;proto"，路径为：common/stream/proto/v1，包名为 proto

在一个 proto 文件中引用另一个 proto 文件

使用 import 关键词，语法为：import ""; 可以引入第三方的 proto 文件，也可以引入自己的 proto 文件

base.proto 文件内容如下：

 syntax = "proto3";
 
option go_package = "./;proto";
 
message Empty {}
 
message Pong {
  string id = 1;
}

在自己的 proto 文件中引入 base.proto 文件：

 import "base.proto";
service Greeter {
  rpc Ping(Empty) returns(Pong);
}

引入第三方的 proto 文件：

要注意的一点是：在使用 protoc 生成代码时，需要指定 proto_path 参数

 import "google/protobuf/empty.proto";
service Greeter {
  rpc Ping(google.protobuf.Empty) returns(Pong);
}
 
message Pong {
  string id = 1;
}

对于第三方在 client/server 中如何使用呢？

比如上面使用的 google.protobuf.Empty，在生成的 go 文件中找到 Empty，然后再看引入它的路径，将它复制到你的 client/server 中即可。

嵌套 message 对象

嵌套使用，Address 是 Person 的嵌套对象，Address 只能在 Person 中使用，不能在 Person 外使用

 message Person {
  string name = 1;
  int32 age = 2;
  Address address = 3;
 
  message Address {
    string country = 1;
    string city = 2;
  }
}

在 client/server 中怎么使用呢？

还是去源码中找 address，我们会看到生成的 address 是 Person_Address，使用就可以了

ps：通过 import 引入自己写的 proto 文件，要自己单独生成 go 文件，然后再使用，否则会找不到

map 类型

 message Person {
  string name = 1;
  int32 age = 2;
  map address = 3;
}

在 client/server 中使用

 c.Ping(context.Background(), &proto.Person{
  Name: "uccs",
  Age:  18,
  Address: map[string]string{
    "City": "深圳",
  },
})

使用 Timestamp

 import "google/protobuf/timestamp.proto";
 
message Person {
  string name = 1;
  int32 age = 2;
  map address = 3;
  google.protobuf.Timestamp birthday = 4;
}

在 client/server 中如何使用 timestamp 呢？

也是一样去生成的 go 文件中找，如图所示:

找到它后，如何实例化 timestamppb 呢？

还是一步步去追溯源码，找到 New 方法，然后再看它的参数，就可以了

 import "google.golang.org/protobuf/types/known/timestamppb"
c.Ping(context.Background(), &proto.Person{
  Name: "uccs",
  Age:  18,
  Address: map[string]string{
    "City": "深圳",
  },
  Birthday: timestamppb.New(time.Now()),
})

grpc

grpc 的 metadata

对于每次请求，需要 grpc 带上一些 metadata，用来传递一些额外的信息，比如 token，trace id 等等。

metadata 是以 key-value 的形式存在的，key 为 string 类型，value 为 []string 类型。

metadata 文档和源码

metadata 类型定义如下：

type MD map[string][]string

实例化 metadata

实例化 metadata 有两种方式：

第一种：

md := metadata.New(map[string][]string{"key1": "value1", "key2": "value2"})

第二种：这种写法不区分大小写，会统一转成小写；key 和 value 之间也是用逗号分隔

 md := metadata.Pairs(
  "key1", "value1",
  "key1", "value1-1",  // 会自动组合成 { key1: 【value1, value1-1}
  "key2", "value2",
)

发送 metadata

 md := metadata.Pairs("key1", "value1")
 
// 新建一个有 metadata 的 context
ctx := metadata.NewOutgoingContext(context.Background(), md)
 
// 单向 rpc
res, err := client.SomeMethod(ctx, SomeRequest)

接收 metadata

 SomeMethod(ctx context.Context, in *pb.SomeRequest)(*pb.SomeResponse, error){
  md, ok := metadata.FromIncomingContext(ctx)
}

拦截器

请请求或者响应之前，统一对他们进行一些验证或处理，以实现一些通用的功能

在 server 中使用拦截器

server 中使用 UnaryInterceptor，它返回一个 grpc.ServerOption，可以在 grpc.NewServer 中使用

opt := grpc.UnaryInterceptor(interceptor)

我们只需要实现 interceptor 函数即可，它的签名通过查看 grpc.UnaryInterceptor 可以知道：

type UnaryServerInterceptor func(ctx context.Context, req interface{}, info *UnaryServerInfo, handler UnaryHandler) (resp interface{}, err error)

具体代码如下：

 interceptor := func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
  fmt.Println("server 端被拦截了...")
  return handler(ctx, req)
}
opt := grpc.UnaryInterceptor(interceptor)
g := grpc.NewServer(opt)

在 client 中使用拦截器

client 中使用 WithUnaryInterceptor，它返回一个 grpc.DialOption，可以在 grpc.Dial 中使用

opt := grpc.WithUnaryInterceptor(interceptor)

我们只需要实现 interceptor 函数即可，它的签名通过查看 grpc.WithUnaryInterceptor 可以知道：

type UnaryClientInterceptor func(ctx context.Context, method string, req, reply interface{}, cc *ClientConn, invoker UnaryInvoker, opts ...CallOption) error

具体代码如下：

 interceptor := func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
  fmt.Println("client 端被拦截了...")
  return invoker(ctx, method, req, reply, cc, opts...)
}
opt := grpc.WithUnaryInterceptor(interceptor)
conn, err := grpc.Dial("127.0.0.1:8080", grpc.WithInsecure(), opt)

应用

以验证 token 为例：

在 client 端修改 ctx 即可：

 interceptor := func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
  md := metadata.New(map[string]string{
    "token": "uccs",
  })
  ctx = metadata.NewOutgoingContext(ctx, md)
  return invoker(ctx, method, req, reply, cc, opts...)
}
opt := grpc.WithUnaryInterceptor(interceptor)
conn, err := grpc.Dial("127.0.0.1:8080", grpc.WithInsecure(), opt)

server 端验证 token：

 interceptor := func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
  md, ok := metadata.FromIncomingContext(ctx)
  if !ok {
    return resp, status.Errorf(codes.Unauthenticated, "无 token")
  }
 
  var (
    token  string
  )
 
  if v1, ok := md["token"]; ok {
    token = key1[0]
  }
 
  if token != "10101" {
    return resp, status.Errorf(codes.Unauthenticated, "无效的 token")
  }
 
  return handler(ctx, req)
}
opt := grpc.UnaryInterceptor(interceptor)
g := grpc.NewServer(opt)

使用 gprc 改写这个方法

grpc 提供了一个 WithPerRPCCredentials 拦截器，可以在 client 端使用，它的签名如下：

 type PerRPCCredentials interface {
  GetRequestMetadata(ctx context.Context, uri ...string) (map[string]string, error)
  RequireTransportSecurity() bool // 返回 `true` 表示开启 `TLS`，返回 `false` 表示不开启 `TLS`
}

也就是说我们实现 GetRequestMetadata 和 RequireTransportSecurity 方法即可实现拦截器

 func (c customCredential) GetRequestMetadata(ctx context.Context, uri ...string) (map[string]string, error) {
  return map[string]string{
    "token":  "101011",
  }, nil
}
func (c customCredential) RequireTransportSecurity() bool {
  return false
}
 
opt := grpc.WithPerRPCCredentials(customCredential{})
conn, err := grpc.Dial("127.0.0.1:8080", grpc.WithInsecure(), opt)

异常处理

状态码文档

server 端

server 抛出错误：

return status.Error(codes.InvalidArgument, "invalid username")

client 处理错误

 st, ok := status.FromError(err)
if ok {
  st.Message()
  st.Code()
}

往期文章

go 项目ORM、测试、api文档搭建

go 开发短网址服务笔记

go 实现统一加载资源的入口

go 语言编写简单的分布式系统

go 中 rpc 和 grpc 的使用

	enum Corpus {
	UNIVERSAL = 0;
	WEB = 1;
	IMAGES = 2;
	LOCAL = 3;
	NEWS = 4;
	PRODUCTS = 5;
	VIDEO = 6;
	}

	enum EnumAllowingAlias {
	option allow_alias = true;
	UNKNOWN = 0;
	STARTED = 1;
	RUNNING = 1;
	}
	enum EnumNotAllowingAlias {
	UNKNOWN = 0;
	STARTED = 1;
	// RUNNING = 1; // Uncommenting this line will cause a compile error inside Google and a warning message outside.
	}

	syntax = "proto3";

	option go_package = "./;proto";

	message Empty {}

	message Pong {
	string id = 1;
	}

	import "base.proto";
	service Greeter {
	rpc Ping(Empty) returns(Pong);
	}

	import "google/protobuf/empty.proto";
	service Greeter {
	rpc Ping(google.protobuf.Empty) returns(Pong);
	}

	message Pong {
	string id = 1;
	}

	message Person {
	string name = 1;
	int32 age = 2;
	Address address = 3;

	message Address {
	string country = 1;
	string city = 2;
	}
	}

	c.Ping(context.Background(), &proto.Person{
	Name: "uccs",
	Age: 18,
	Address: map[string]string{
	"City": "深圳",
	},
	})

	import "google/protobuf/timestamp.proto";

	message Person {
	string name = 1;
	int32 age = 2;
	map address = 3;
	google.protobuf.Timestamp birthday = 4;
	}

	import "google.golang.org/protobuf/types/known/timestamppb"
	c.Ping(context.Background(), &proto.Person{
	Name: "uccs",
	Age: 18,
	Address: map[string]string{
	"City": "深圳",
	},
	Birthday: timestamppb.New(time.Now()),
	})

	md := metadata.Pairs(
	"key1", "value1",
	"key1", "value1-1", // 会自动组合成 { key1: 【value1, value1-1}
	"key2", "value2",
	)

	md := metadata.Pairs("key1", "value1")

	// 新建一个有 metadata 的 context
	ctx := metadata.NewOutgoingContext(context.Background(), md)

	// 单向 rpc
	res, err := client.SomeMethod(ctx, SomeRequest)

	SomeMethod(ctx context.Context, in pb.SomeRequest)(pb.SomeResponse, error){
	md, ok := metadata.FromIncomingContext(ctx)
	}

	interceptor := func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
	fmt.Println("server 端被拦截了...")
	return handler(ctx, req)
	}
	opt := grpc.UnaryInterceptor(interceptor)
	g := grpc.NewServer(opt)

	interceptor := func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
	fmt.Println("client 端被拦截了...")
	return invoker(ctx, method, req, reply, cc, opts...)
	}
	opt := grpc.WithUnaryInterceptor(interceptor)
	conn, err := grpc.Dial("127.0.0.1:8080", grpc.WithInsecure(), opt)

	interceptor := func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
	md := metadata.New(map[string]string{
	"token": "uccs",
	})
	ctx = metadata.NewOutgoingContext(ctx, md)
	return invoker(ctx, method, req, reply, cc, opts...)
	}
	opt := grpc.WithUnaryInterceptor(interceptor)
	conn, err := grpc.Dial("127.0.0.1:8080", grpc.WithInsecure(), opt)

	interceptor := func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
	md, ok := metadata.FromIncomingContext(ctx)
	if !ok {
	return resp, status.Errorf(codes.Unauthenticated, "无 token")
	}

	var (
	token string
	)

	if v1, ok := md["token"]; ok {
	token = key1[0]
	}

	if token != "10101" {
	return resp, status.Errorf(codes.Unauthenticated, "无效的 token")
	}

	return handler(ctx, req)
	}
	opt := grpc.UnaryInterceptor(interceptor)
	g := grpc.NewServer(opt)

	type PerRPCCredentials interface {
	GetRequestMetadata(ctx context.Context, uri ...string) (map[string]string, error)
	RequireTransportSecurity() bool // 返回 `true` 表示开启 `TLS`，返回 `false` 表示不开启 `TLS`
	}

	func (c customCredential) GetRequestMetadata(ctx context.Context, uri ...string) (map[string]string, error) {
	return map[string]string{
	"token": "101011",
	}, nil
	}
	func (c customCredential) RequireTransportSecurity() bool {
	return false
	}

	opt := grpc.WithPerRPCCredentials(customCredential{})
	conn, err := grpc.Dial("127.0.0.1:8080", grpc.WithInsecure(), opt)

	st, ok := status.FromError(err)
	if ok {
	st.Message()
	st.Code()
	}