服务间使用 Http 相互调用时,经常会设置一些业务自定义 header 如时间戳、trace信息等,gRPC使用 HTTP/2
协议自然也是支持的,gRPC 通过 google.golang.org/grpc/metadata
包内的 MD
类型提供相关的功能接口。
// MD is a mapping from metadata keys to values. Users should use the following
// two convenience functions New and Pairs to generate MD.
type MD map[string][]string
metadata.MD
类型的定义非常简单,可以像一个普通的 map 一样直接操作,同时 metadata 包里封装了很多工
具方法供我们使用。
// 使用 New 方法创建
md := metadata.New(map[string]string{"k1":"v1", "k2", "v2"})
// 直接使用 make 创建
md := make(metadata.MD)
// 使用 Pairs 方法创建
md := metadata.Pairs("k1", "v1-1", "k1", "v1-2")
// 一些操作
md.Set("key", "v1", "v2")
md.Append("key", "v3")
md.Delete("key")
vals := md.Get("key")
客户端请求的 metadata 是通过设置 context 使用的,metadata 包提供了两个 context 相关的方法,设置好
context 后直接在调用 rpc 方法时传入即可:
md := metadata.New(map[string]string{"k1":"v1", "k2", "v2"})
// 使用 NewOutgoingContext 初始化一个新的 context
ctx := metadata.NewOutgoingContext(context.Background(), md)
// 使用 AppendToOutgoingContext 向 context 追加 metadata
ctx = metadata.AppendToOutgoingContext(ctx, "k3", "v3")
客户端接收响应中的 metadata 需要区分普通 rpc 和 stream rpc :
// 普通 rpc,使用 grpc.Header 方法包装为 CallOption
var md metadata.MD
res, err := client.Ping(ctx, &pb.PingRequest{Value: "ping"}, grpc.Header(&md))
// stream rpc
stream, err := client.MultiPong(context.Background(), &pb.PingRequest{Value: "ping"})
if err != nil {
log.Fatal(err)
}
// 通过 stream 对象的 Header 方法获取
md, err := stream.Header()
if err != nil {
log.Fatal(err)
}
对应客户端请求的 metadata 是使用 context 设置的,那么服务端在接收时自然也是从 context 中读取,
metadata 包中的 FromIncommingContext
方法就是用来读取 context 中的 metadata数据的:
// unary rpc
func (s *PingPongServer) Ping(ctx context.Context, req *pb.PingRequest) (*pb.PongResponse, error) {
// 读取请求metadata
md, ok := metadata.FromIncomingContext(ctx)
if ok {
log.Printf("Got md: %v", md)
}
// stream rpc
func (s *PingPongServer) MultiPingPong(stream pb.PingPong_MultiPingPongServer) error {
md, ok := metadata.FromIncomingContext(stream.Context())
if ok {
log.Printf("Got md: %v", md)
}
服务端设置响应的 metadata 也非常简单,只需要调用封装好的 SetHeader
或 SendHeader
方法即可:
// unary rpc
func (s *PingPongServer) Ping(ctx context.Context, req *pb.PingRequest) (*pb.PongResponse, error) {
// 读取请求metadata
md, ok := metadata.FromIncomingContext(ctx)
if ok {
log.Printf("Got md: %v", md)
}
// SetHeader设置响应 metadata
grpc.SetHeader(ctx, metadata.New(map[string]string{"rkey": "rval"}))
// 注意 SendHeader 只能调用一次
// grpc.SendHeader(ctx, metadata.New(map[string]string{"rkey": "rval"}))
// stream rpc, 调用 stream 的 SetHeader 方法
func (s *PingPongServer) MultiPong(req *pb.PingRequest, stream pb.PingPong_MultiPongServer) error {
stream.SetHeader(metadata.New(map[string]string{"rkey": "rval"}))
// 注意 SendHeader 只能调用一次
// stream.SendHeader(metadata.New(map[string]string{"rkey": "rval"}))
demo.proto
文件内容:
syntax="proto3";
package protos;
option go_package = "./protos;protos";
service Greeter {
rpc SayHello(HelloRequest) returns(HelloReply){}
}
message HelloRequest {
string name = 1;
}
message HelloReply {
string message = 1;
}
编译生成demo.pb.go 文件:
$ protoc --go_out=plugins=grpc:. demo.proto
server.go
文件的内容:
package main
import (
"context"
"flag"
"fmt"
"demo/protos"
"google.golang.org/grpc"
"google.golang.org/grpc/metadata"
"log"
"net"
)
var host = "127.0.0.1"
var (
ServiceName = flag.String("ServiceName", "hello_service", "service name")
Port = flag.Int("Port", 50001, "listening port")
)
type server struct {
}
func main() {
flag.Parse()
lis, err := net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", *Port))
if err != nil {
log.Fatalf("failed to listen:%s", err)
} else {
fmt.Printf("listen at :%d\n", *Port)
}
defer lis.Close()
s := grpc.NewServer()
defer s.GracefulStop()
protos.RegisterGreeterServer(s, &server{})
addr := fmt.Sprintf("%s:%d", host, *Port)
fmt.Printf("server add:%s\n", addr)
if err := s.Serve(lis); err != nil {
fmt.Printf("failed to server: %s", err)
}
}
func (s *server) SayHello(ctx context.Context, in *protos.HelloRequest) (*protos.HelloReply, error) {
md, ok := metadata.FromIncomingContext(ctx)
if !ok {
fmt.Printf("get metadata error")
}else{
fmt.Println("get metadata success: ",md)
}
if t, ok := md["timestamp"]; ok {
fmt.Printf("timestamp from metadata:\n")
for i, e := range t {
fmt.Printf(" %d. %s\n", i, e)
}
}
if t1, ok1 := md["key1"]; ok1 {
fmt.Printf("key1 from metadata:\n")
for i, e := range t1 {
fmt.Printf(" %d . %s\n", i, e)
}
}
if len(md) > 0 {
for k, v := range md {
fmt.Printf("%v:%v\n", k, v)
}
}
return &protos.HelloReply{Message: "server: " + in.Name}, nil
}
client.go
文件的内容:
package main
import (
"context"
"fmt"
"demo/protos"
"google.golang.org/grpc"
"google.golang.org/grpc/metadata"
"time"
)
const (
timestampFormat = time.StampNano
)
func main() {
conn, err := grpc.Dial("127.0.0.1:50001", grpc.WithInsecure())
if err != nil {
panic(err)
}
client := protos.NewGreeterClient(conn)
md := metadata.Pairs("timestamp", time.Now().Format(timestampFormat))
md = metadata.New(map[string]string{"key1": "val1", "key2": "val2"})
ctx := metadata.NewOutgoingContext(context.Background(), md)
resp, err := client.SayHello(ctx, &protos.HelloRequest{Name: "Hello"})
if err == nil {
fmt.Printf("Reply is : %s\n", resp.Message)
} else {
fmt.Printf("call server error:%s\n", err)
}
}
[root@zsx demo]# go run server.go
listen at :50001
server add:127.0.0.1:50001
get metadata success: map[:authority:[127.0.0.1:50001] content-type:[application/grpc] key1:[val1] key2:[val2] user-agent:[grpc-go/1.53.0]]
key1 from metadata:
0 . val1
:authority:[127.0.0.1:50001]
content-type:[application/grpc]
user-agent:[grpc-go/1.53.0]
key2:[val2]
key1:[val1]
[root@zsx demo]# go run client.go
Reply is : server: Hello
# 项目结构
[root@zsx protoc]# tree demo/
demo/
├── client.go
├── demo.proto
├── go.mod
├── go.sum
├── protos
│ └── demo.pb.go
└── server.go
1 directory, 6 files