快捷導(dǎo)航

Golang處理gRPC請(qǐng)求/響應(yīng)元數(shù)據(jù)的示例代碼

更新時(shí)間：2024年03月01日 09:30:04 作者：ghosind

前段時(shí)間實(shí)現(xiàn)內(nèi)部gRPC框架時(shí),為了實(shí)現(xiàn)在服務(wù)端攔截器中打印請(qǐng)求及響應(yīng)的頭部信息,便查閱了部分關(guān)于元數(shù)據(jù)的資料,因?yàn)橹形木W(wǎng)絡(luò)上對(duì)于該領(lǐng)域的信息較少,于是在這做了一些簡單的總結(jié),需要的朋友可以參考下

元數(shù)據(jù)

gRPC的元數(shù)據(jù)（metadata）是基于HTTP/2頭部實(shí)現(xiàn)的鍵值對(duì)數(shù)據(jù)，它通常用來實(shí)現(xiàn)gRPC的鑒權(quán)、鏈路跟蹤以及自定義頭部數(shù)據(jù)等功能。

gRPC的元數(shù)據(jù)分為兩種類型，分別是Header及Trailer。Header可以由客戶端或服務(wù)端發(fā)送，它在客戶端請(qǐng)求數(shù)據(jù)或服務(wù)器響應(yīng)數(shù)據(jù)前發(fā)送。Trailer是一種特殊的頭部信息，它僅可由服務(wù)端發(fā)送，且位于發(fā)送的數(shù)據(jù)之后。

客戶端處理

在gRPC客戶端中，無論是一元調(diào)用還是流調(diào)用，可以比較簡單地通過google.golang.org/grpc/metadata包提供的AppendToOutgoingContext或NewOutgoingContext方法向請(qǐng)求中加入頭部元數(shù)據(jù)，例如以下幾種方式：

// 通過metadata創(chuàng)建新的context
md := metadata.Pairs("k1", "v1", "k2", "v2")
ctx := metadata.NewOutgoingContext(ctx, md)

// 或是向context中添加元數(shù)據(jù)
ctx = metadata.AppendToOutgoingContext(ctx, "k3", "v3")

// ... 通過ctx進(jìn)行RPC調(diào)用

對(duì)于服務(wù)端返回的響應(yīng)中的元數(shù)據(jù)，一元調(diào)用與流調(diào)用的處理方式就較為不同。對(duì)于一元調(diào)用，需要提前定義好用于存儲(chǔ)元數(shù)據(jù)的變量，然后在調(diào)用時(shí)通過grpc.Header或grpc.Trailer增加調(diào)用的選項(xiàng)：

var header, trailer metadata.MD
resp, err := cli.UnaryCall(ctx, req, grpc.Header(&header), grpc.Trailer(&trailer))

// 處理header或trailer

而對(duì)于任意方式的流調(diào)用，都可以簡單地通過流調(diào)用返回流的Header或Trailer方法獲得元數(shù)據(jù)：

stream, err := cli.StreamCall(ctx)

header, err := stream.Header()
trailer, err := stream.Trailer()

服務(wù)端處理

對(duì)于服務(wù)端，請(qǐng)求的元數(shù)據(jù)需要通過metadata.FromIncomingContext從context中獲?。?/p>

// 一元調(diào)用
md, ok := metadata.FromIncomingContext(ctx)

// 流調(diào)用
ctx := stream.Context() // 需要先從流中得到context
md, ok := metadata.FromIncomingContext(ctx)

同樣，在服務(wù)端發(fā)送元數(shù)據(jù)需要根據(jù)一元調(diào)用與流調(diào)用使用不同的方式。對(duì)于一元調(diào)用，可以通過grpc.SendHeader、grpc.SetHeader以及grpc.SetTrailer方法設(shè)置發(fā)送的元數(shù)據(jù)，例如：

header := metadata.Pairs("header-key", "header-val")
grpc.SendHeader(ctx, header)
trailer := metadata.Pairs("trailer-key", "trailer-val")
grpc.SetTrailer(ctx, trailer)

對(duì)于上述的SendHeader及SetHeader方法，其區(qū)別為SendHeader方法只能調(diào)用一次，而SetHeader方法將會(huì)對(duì)所有調(diào)用的元數(shù)據(jù)進(jìn)行合并發(fā)送。

對(duì)于流調(diào)用，服務(wù)端發(fā)送元數(shù)據(jù)則是通過流對(duì)象中的上述方法：

header := metadata.Pairs("header-key", "header-val")
stream.SendHeader(,header)
trailer := metadata.Pairs("trailer-key", "trailer-val")
stream.SetTrailer(trailer)

服務(wù)器攔截器處理

對(duì)于gRPC服務(wù)端一元調(diào)用及流調(diào)用攔截器，請(qǐng)求元數(shù)據(jù)的讀取與響應(yīng)元數(shù)據(jù)的發(fā)送與上一節(jié)中的實(shí)現(xiàn)相同，便不再贅述。下面我們將討論一下在攔截器中更新請(qǐng)求元數(shù)據(jù)，以及讀取響應(yīng)的元數(shù)據(jù)。

一元調(diào)用攔截器更新請(qǐng)求元數(shù)據(jù)

在服務(wù)端攔截器中更新請(qǐng)求的元數(shù)據(jù)，其實(shí)現(xiàn)的方式與客戶端發(fā)送元數(shù)據(jù)類似，即需要通過更新后的元數(shù)據(jù)創(chuàng)建新的context。對(duì)于一元調(diào)用攔截器，其簡單實(shí)現(xiàn)如下所示：

md, ok := metadata.FromIncomingContext(ctx)

md.Append("new-key", "new-value")
ctx = metadata.NewIncomingContext(ctx, md)

resp, err := handler(ctx, req) // 傳遞context至handler中

一元調(diào)用攔截器讀取響應(yīng)元數(shù)據(jù)

對(duì)于一元調(diào)用響應(yīng)的元數(shù)據(jù)，gRPC未提供直接訪問的方法響應(yīng)的元數(shù)據(jù)。為了在攔截器中能讀取到響應(yīng)的元數(shù)據(jù)，我們可以通過覆蓋原始grpc.ServerTransportStream并對(duì)設(shè)置的元數(shù)據(jù)進(jìn)行備份的方式進(jìn)行實(shí)現(xiàn)。

type WrappedServerTransportStream struct {
  grpc.ServerTransportStream

  header  metadata.MD
  trailer metadata.MD
}

func (s *WrappedServerTransportStream) SendHeader(md metadata.MD) error {
  if err := s.ServerTransportStream.SendHeader(md); err != nil {
    return err
  }

  s.header = md

  return nil
}

// 在需要的情況下繼續(xù)實(shí)現(xiàn)下面的幾個(gè)方法：
// func (s *WrappedServerTransportStream) SetHeader(metadata.MD) error
// func (s *WrappedServerTransportStream) SetTrailer(metadata.MD) error

在定義帶有元數(shù)據(jù)副本的ServerTransportStream實(shí)現(xiàn)后，我們需要通過grpc.ServerTransportStreamFromContext獲取到一元調(diào)用的原始流，在對(duì)其進(jìn)行封裝后，調(diào)用grpc.NewContextWithServerTransportStream創(chuàng)建新的context。

stream := grpc.ServerTransportStreamFromContext(ctx)
wrappedStream := &WrappedServerTransportStream{
  ServerTransportStream: stream,
}
ctx = grpc.NewContextWithServerTransportStream(ctx, wrappedStream)

resp, err := handler(ctx, req)

// 通過wrappedStream.header、wrappedStream.trailer讀取響應(yīng)的元數(shù)據(jù)

需要注意，grpc.ServerTransportStream接口是一個(gè)實(shí)驗(yàn)性的接口，在后續(xù)版本中可能會(huì)被移除，所以本節(jié)中描述的方法在后續(xù)版本中可能不再可用。

流調(diào)用攔截器更新請(qǐng)求元數(shù)據(jù)

而對(duì)于流調(diào)用，gRPC沒有提供修改其context的方法，為了實(shí)現(xiàn)修改流調(diào)用請(qǐng)求元數(shù)據(jù)，就需要實(shí)現(xiàn)grpc.ServerStream接口并加入帶有修改后元數(shù)據(jù)的context。以下是一個(gè)簡單的實(shí)現(xiàn)：

type WrappedStream struct {
  grpc.ServerStream
  ctx context.Context
}

func (s *WrappedStream) Context() context.Context {
  return s.ctx
}

func ExampleStreamInterceptor(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
  md, ok := metadata.FromIncomingContext(ss.Context())
  md.append("new-key", "new-value")

  ctx := metadata.NewIncomingContext(ss.Context(), md)

  return handler(srv, &WrappedStream{ss, ctx})
}

流調(diào)用攔截器讀取響應(yīng)元數(shù)據(jù)

與在一元調(diào)用攔截器中相同，若需要在流調(diào)用攔截器中讀取響應(yīng)的元數(shù)據(jù)，我們可以實(shí)現(xiàn)grpc.ServerStream接口，并在其中保存元數(shù)據(jù)的副本。例如我們可以在上節(jié)的WrappedStream的基礎(chǔ)上，對(duì)其進(jìn)行一定修改：

type WrappedStream struct {
  grpc.ServerStream

  header  metadata.MD
  trailer metadata.MD
}

func (s *WrappedStream) SendHeader(md metadata.MD) error {
  if err := s.ServerStream.SendHeader(md); err != nil {
    return err
  }

  s.header = md

  return nil
}

// 繼續(xù)實(shí)現(xiàn)SetHeader、SetTrailer等方法

func ExampleStreamInterceptor(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
  stream := &WrappedStream{ServerStream: ss}
  err := handler(srv, stream)

  // 通過stream.header、stream.trailer讀取響應(yīng)元數(shù)據(jù)

  return err
}

以上就是Golang處理gRPC請(qǐng)求/響應(yīng)元數(shù)據(jù)的示例代碼的詳細(xì)內(nèi)容，更多關(guān)于Golang處理gRPC請(qǐng)求/響應(yīng)的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: