当前位置: 首页 > news >正文

gRPC-Go基础(4)metadata和超时设置

文章目录

  • 0. 简介
  • 1. metadata
    • 1.1 metadata结构
    • 1.2 metadata创建
    • 1.3 客户端处理metadata
    • 1.4 服务端处理metadata
    • 1.5 metadata的传输
  • 2. 超时设置
    • 2.1 客户端输出超时信息
    • 2.2 服务端端接收超时信息
  • 3. 小结

0. 简介

Go在多个go routine之间传递数据使用的是Go SDK提供的context包,而context的作用范围在进程内,而gRPC使用的是跨进程的网络传输,那如何实现跨进程的元数据传输呢?

1. metadata

1.1 metadata结构

metadata的简单理解,就是http 的 Header 中的 key-value 对

gRPC使用metadata在服务之间传输全局数据,metadata形式为键值对(k-v)列表,如下:

type MD map[string][]string
  • k一般为字符串,v可以是字符串,也可以是二进制数据,当v是二进制数据时,k必须以-bin结尾,二进制数据会被base64编码然后传输;
  • 同时,k不能以grpc-开头,因为这是为gRPC所保留;
  • k中大写字符会被转化为小写;
  • 如上,一个k可以对应多个v。

1.2 metadata创建

New方法

md := metadata.New(map[string]string{"k1":"v1","k2":"v2"})

Pair方法(相同的key自动合并)

md := metadata.Pairs("k1", "v1","k1", "v1.2","k2-bin", string([]byte{1, 2}),
)

k1对应的值会被自动合并为map[string][]string{k1: {v1, v1.2}}。k2-bin对应的值会被进行base64编码。

1.3 客户端处理metadata

1.3.1 发送metadata

NewOutgoingContext

md := metadata.Pairs("k1", "v1","k1", "v1.2","k2-bin", string([]byte{1, 2}),
)
ctx := metadata.NewOutgoingContext(context.Background(), md)

将新创建的 Metadata 添加到 context 中,这样会 覆盖 掉原来已有的 metadata。其实际上就是调用了context.WithValue 方法,生成了一个子context而已,这个子context中包含了传入的metadata。

// NewOutgoingContext creates a new context with outgoing md attached. If used
// in conjunction with AppendToOutgoingContext, NewOutgoingContext will
// overwrite any previously-appended metadata.
func NewOutgoingContext(ctx context.Context, md MD) context.Context {return context.WithValue(ctx, mdOutgoingKey{}, rawMD{md: md})
}

AppendToOutgoingContext

ctx = metadata.AppendToOutgoingContext(ctx, "k1", "v1", "k1", "v1.2", "k2-bin", string([]byte{1, 2}))

AppendToOutgoingContext方法将k-v对添加到已有的context中。如果对应的context没有metadata,那么就会创建一个;如果已有metadata了,那么就将数据添加到原来的metadata(推荐使用 AppendToOutgoingContext)。

1.3.2 接收metadata
目前,gRPC的客户端支持接收的metadata包括header和trailer。
一元RPC
header和trailer可以通过Header和Trailer方法,在调用gRPC方法的CallOption时传入,在函数调用结束后取出。

var header, trailer metadata.MD
res, err := grpcClient.SimpleRoute(ctx, &simplepb.SimpleRequest{Data: "I am iguochan"}, grpc.Header(&header), grpc.Trailer(&trailer))

值得注意的是,虽然流式RPC的方法调用中也有CallOption,但是这两个方法**明确仅用于一元RPC。

// Header returns a CallOptions that retrieves the header metadata
// for a unary RPC.
func Header(md *metadata.MD) CallOption {return HeaderCallOption{HeaderAddr: md}
}// Trailer returns a CallOptions that retrieves the trailer metadata
// for a unary RPC.
func Trailer(md *metadata.MD) CallOption {return TrailerCallOption{TrailerAddr: md}
}

流式RPC
所有的流式RPC都将使用以下形式接收服务端发送的header和trailer。通过接口ClientStream的Header() (metadata.MD, error)和Trailer() metadata.MD方法取出。

stream, err := grpcClient.RouteList(context.Background())// 从流中取header
header, err := stream.Header()// 从流中取出trailer
trailer := stream.Trailer()

1.4 服务端处理metadata

1.4.1 发送metadata
和客户端接收metadata一样,服务端只能发送header和trailer这两类metadata。

一元RPC
有关这些方法的具体用法可以参考go文档。

func (r *Server) SimpleRoute(ctx context.Context, request *simplepb.SimpleRequest) (*simplepb.SimpleResponse, error) {// ...md := metadata.Pairs(/* ... */)err := grpc.SendHeader(ctx, md)// ...md := metadata.Pairs(/* ... */)err = grpc.SetTrailer(ctx, md)
}

流式RPC

func (r *Server) RouteList(server simplepb.Route_RouteListServer) error {// ...md := metadata.Pairs(/* ... */)err := server.SendHeader(md)// ...md := metadata.Pairs(/* ... */)server.SetTrailer(md)// ...
}

1.4.2 接收metadata
要读取客户端发送的元数据,服务器需要从 RPC 上下文中检索它。如果是一元调用,则可以使用 RPC 处理程序的上下文。对于流式调用,服务器需要从流中获取上下文。

一元RPC

func (r *Server) SimpleRoute(ctx context.Context, request *simplepb.SimpleRequest) (*simplepb.SimpleResponse, error) {md, ok := metadata.FromIncomingContext(ctx)if ok {// do something with metadata}// ...
}

流式RPC

func (r *Server) ListValue(request *simplepb.SimpleRequest, server simplepb.Route_ListValueServer) error {md, ok := metadata.FromIncomingContext(server.Context())if ok {// do something with metadata}// ...
}

1.5 metadata的传输

以上介绍了metadata的用法,本节将介绍一下metadata在gRPC各端之间的传输机制。上面说过,在Go中,metadata就是实际上就是调用了context.WithValue 方法,生成了一个子context,那其实不管是在客户端还是服务端,都属于context的上下文传递。
1.5.1 客户端输出metadata
根据前面的文章,我们知道,用户定义好protobuf后通过protoc生成服务端的interface和客户端的桩代码(stub),在桩代码中,已经包含了客户端的实现:

func (c *routeClient) SimpleRoute(ctx context.Context, in *SimpleRequest, opts ...grpc.CallOption) (*SimpleResponse, error) {out := new(SimpleResponse)err := c.cc.Invoke(ctx, "/simplepb.Route/SimpleRoute", in, out, opts...)if err != nil {return nil, err}return out, nil
}

当发起gRPC请求后,会调用Invoke方法,最终会调用invoke方法:

func invoke(ctx context.Context, method string, req, reply interface{}, cc *ClientConn, opts ...CallOption) error {cs, err := newClientStream(ctx, unaryStreamDesc, cc, method, opts...)if err != nil {return err}if err := cs.SendMsg(req); err != nil {return err}return cs.RecvMsg(reply)
}

invoke方法会调用newClientStream,newClientStream方法比较长,在里面会生成一个函数newStream,并调用newClientStreamWithParams:

func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, opts ...CallOption) (_ ClientStream, err error) {// ...var newStream = func(ctx context.Context, done func()) (iresolver.ClientStream, error) {return newClientStreamWithParams(ctx, desc, cc, method, mc, onCommit, done, opts...)}// ...
}

newClientStreamWithParams最红会选择一个transport去传输:

func newClientStreamWithParams(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, mc serviceconfig.MethodConfig, onCommit, doneFunc func(), opts ...CallOption) (_ iresolver.ClientStream, err error) {// ...// Pick the transport to use and create a new stream on the transport.// Assign cs.attempt upon success.op := func(a *csAttempt) error {if err := a.getTransport(); err != nil {return err}if err := a.newStream(); err != nil {return err}// Because this operation is always called either here (while creating// the clientStream) or by the retry code while locked when replaying// the operation, it is safe to access cs.attempt directly.cs.attempt = areturn nil}// ...
}

而gRPC选择的是HTTP2作为传输协议,所以最终a.newStream()会最终调用到HTTP2的NewStream:

// NewStream creates a stream and registers it into the transport as "active"
// streams.  All non-nil errors returned will be *NewStreamError.
func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (*Stream, error) {ctx = peer.NewContext(ctx, t.getPeer())headerFields, err := t.createHeaderFields(ctx, callHdr)if err != nil {return nil, &NewStreamError{Err: err, AllowTransparentRetry: false}}
}

createHeaderFields,顾名思义,就是构建HTTP请求时的头(HEADERS Frame),其有关metadata从context到HTTP Header的转换实现如下:

func (t *http2Client) createHeaderFields(ctx context.Context, callHdr *CallHdr) ([]hpack.HeaderField, error) {// ...if md, added, ok := metadata.FromOutgoingContextRaw(ctx); ok {var k stringfor k, vv := range md {// HTTP doesn't allow you to set pseudoheaders after non pseudoheaders were set.if isReservedHeader(k) {continue}for _, v := range vv {headerFields = append(headerFields, hpack.HeaderField{Name: k, Value: encodeMetadataHeader(k, v)})}}for _, vv := range added {for i, v := range vv {if i%2 == 0 {k = strings.ToLower(v)continue}// HTTP doesn't allow you to set pseudoheaders after non pseudoheaders were set.if isReservedHeader(k) {continue}headerFields = append(headerFields, hpack.HeaderField{Name: k, Value: encodeMetadataHeader(k, v)})}}}// ...
}

可以看到,所谓metadata,其实最终是通过HTTP2协议的头帧带入到网络的,主要存储在Binary-HeaderASCII-Header,感兴趣的同学可以参考gRPC over HTTP2。
1.5.2 服务端接收metadata
服务端通过Serve方法来启动,监听来自客户端的连接:

err = grpcServer.Serve(listener)
if err != nil {log.Fatalf("grpcServer.Serve err: %v", err)
}

Serve方法最终会调用到handleRawConn,handleRawConn会通过HTTP2去接收消息。

func (s *Server) handleRawConn(lisAddr string, rawConn net.Conn) {// ...st := s.newHTTP2Transport(rawConn)// ...go func() {s.serveStreams(st)s.removeConn(lisAddr, st)}()
}

接下来的调用流程涉及的方法是:serveStreams——>HandleStreams——>operateHeaders,最终在operateHeaders,会将所有的头文件中的传入的Header(除去默认的一些字段等)读入到mdata中,这就是服务端拿到的metadata。

func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(*Stream), traceCtx func(context.Context, string) context.Context) (fatal bool) {// ...for _, hf := range frame.Fields {switch hf.Name {// ...default:if isReservedHeader(hf.Name) && !isWhitelistedHeader(hf.Name) {break}v, err := decodeMetadataHeader(hf.Name, hf.Value)if err != nil {headerError = truelogger.Warningf("Failed to decode metadata header (%q, %q): %v", hf.Name, hf.Value, err)break}mdata[hf.Name] = append(mdata[hf.Name], v)}}// ...
}

2. 超时设置

除了一些元数据的传输,在gRPC中,我们还支持将context中的超时传递到接下来的每个服务中,甚至可以跨语言,而这是怎么做到的呢?首先我们想到是通过metadata传输,实际上并不是的。

2.1 客户端输出超时信息

其实从前面1.5.1 客户端输出metadata的最后一步,createHeaderFields方法中,就有如下处理,超时的数值会被作为key是grpc-timeout的值放在HTTP2的头中(正如备注中所言,其实在网络上的传输时间是没有被计入的,会有一定误差)。

func (t *http2Client) createHeaderFields(ctx context.Context, callHdr *CallHdr) ([]hpack.HeaderField, error) {// ...if dl, ok := ctx.Deadline(); ok {// Send out timeout regardless its value. The server can detect timeout context by itself.// TODO(mmukhi): Perhaps this field should be updated when actually writing out to the wire.timeout := time.Until(dl)headerFields = append(headerFields, hpack.HeaderField{Name: "grpc-timeout", Value: grpcutil.EncodeDuration(timeout)})}// ...
}

2.2 服务端端接收超时信息

同理,在HTTP2的server端的头处理函数operateHeaders中,也会对grpc-timeout对应的值进行处理,将其值读入到timeout中,并置上timeoutSet标志,用于后续处理。

func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(*Stream), traceCtx func(context.Context, string) context.Context) (fatal bool) {// ...var (// ...timeoutSet booltimeout    time.Duration)// ...for _, hf := range frame.Fields {switch hf.Name {// ...case "grpc-timeout":timeoutSet = truevar err errorif timeout, err = decodeTimeout(hf.Value); err != nil {headerError = true}}}// ...if timeoutSet {s.ctx, s.cancel = context.WithTimeout(t.ctx, timeout)} else {s.ctx, s.cancel = context.WithCancel(t.ctx)}// ...
}

3. 小结

根据以上分析,我们可以发现,不管是metadata还是超时,都是通过HTTP2的头帧传入到对侧的。

相关文章:

gRPC-Go基础(4)metadata和超时设置

文章目录 0. 简介1. metadata1.1 metadata结构1.2 metadata创建1.3 客户端处理metadata1.4 服务端处理metadata1.5 metadata的传输 2. 超时设置2.1 客户端输出超时信息2.2 服务端端接收超时信息 3. 小结 0. 简介 Go在多个go routine之间传递数据使用的是Go SDK提供的context包…...

语言模型:从n-gram到神经网络的演进

目录 1 前言2 语言模型的两个任务2.1 自然语言理解2.2 自然语言生成 3 n-gram模型4 神经网络语言模型5 结语 1 前言 语言模型是自然语言处理领域中的关键技术之一,它致力于理解和生成人类语言。从最初的n-gram模型到如今基于神经网络的深度学习模型,语言…...

docker compose 部署 grafana + loki + vector 监控kafka消息

Centos7 随笔记录记录 docker compose 统一管理 granfana loki vector 监控kafka 信息。 当然如果仅仅是想通过 Grafana 监控kafka,推荐使用 Grafana Prometheus 通过JMX监控kafka 目录 1. 目录结构 2. 前提已安装Docker-Compose 3. docker-compose 自定义服…...

kubeadm创建k8s集群

kubeadm来快速的搭建一个k8s集群: 二进制搭建适合大集群,50台以上。 kubeadm更适合中下企业的业务集群。 部署框架 master192.168.10.10dockerkubelet kubeadm kubectl flannelnode1192.168.10.20dockerkubelet kubeadm kubectl flannelnode2192.168.1…...

鸿蒙开发之android对比开发《基础知识》

基于华为鸿蒙未来可能不再兼容android应用,推出鸿蒙开发系列文档,帮助android开发人员快速上手鸿蒙应用开发。 1. 鸿蒙使用什么基础语言开发? ArkTS是鸿蒙生态的应用开发语言。它在保持TypeScript(简称TS)基本语法风…...

2702 高级打字机

因为Undo操作只能撤销Type操作,所以Undo x 实际上就是删除文章末尾x个字母。用一个栈即可解决(每个字母最多进出一次)。 这种情况下只需要设计一个合理的数据结构依次执行操作即可。 版本树:Undo x撤销最近的x次修改操作&#xf…...

yolov5旋转目标检测-遥感图像检测-无人机旋转目标检测-附代码和原理

综述 为了解决旋转目标检测问题,研究者们提出了多种方法和算法。以下是一些常见的旋转目标检测方法: 基于滑动窗口的方法:在图像上以不同的尺度和角度滑动窗口,通过分类器判断窗口中是否存在目标。这种方法简单直观,…...

Qt学习:Qt的意义安装Qt

Qt 的简介 QT 是一个跨平台的 C图形用户界面应用程序框架。它为程序开发者提供图形界面所需的所有功能。它是完全面向对象的,很容易扩展,并且允许真正地组件编程。 支持平台 xP 、 Vista、Win7、win8、win2008、win10Windows . Unix/Linux: Ubuntu 等…...

Anylogic Pro 8.8.x for Mac / for Linux Crack

Digital twins – a step towards a digital enterprise AnyLogic是唯一一个支持创建模拟模型的方法的模拟建模工具:面向过程(离散事件)、系统动态和代理,以及它们的任何组合。AnyLogic提供的建模语言的独特性、灵活性和强大性使…...

ROS无人机初始化GPS定位漂移误差,确保无人机稳定飞行

引言: 由于GPS在室外漂移的误差比较大,在长时间静止后启动,程序发布的位置可能已经和预期的位置相差较大,导致无法完成任务,尤其是气压计的数据不准,可能会导致无人机不能起飞或者一飞冲天。本文主要是在进…...

k8s网络类型

k8s中的通信模式: pod内部之间容器与容器之间的通信。 在同一个pod中的容器共享资源和网络,使用同一个网络命名空间。可以直接通信的。 同一个node节点之内,不同pod之间的通信。 每一个pod都有一个全局的真实的IP地址,同一个n…...

Seata 中封装了四种分布式事务模式,分别是: AT 模式, TCC 模式, Saga 模式, XA 模式,

文章目录 seata概述Seata 中封装了四种分布式事务模式,分别是:AT 模式,TCC 模式,Saga 模式,XA 模式, 今天我们来聊聊seata seata 概述 在微服务架构下,由于数据库和应用服务的拆分&#xff0c…...

为什么设计制造行业需要数据加密?

设计制造行业是一个涉及多种技术、工艺、材料和产品的广泛领域,它对经济和社会的发展有着重要的影响。然而,随着数字化、智能化和网络化的发展,设计制造行业也面临着越来越多的数据安全风险,如数据泄露、数据篡改、数据窃取等。这…...

查看ios app运行日志

摘要 本文介绍了一款名为克魔助手的iOS应用日志查看工具,该工具可以方便地查看iPhone设备上应用和系统运行时的实时日志和奔溃日志。同时还提供了奔溃日志分析查看模块,可以对苹果奔溃日志进行符号化、格式化和分析,极大地简化了开发者的调试…...

怎么卸载macOS上的爱思助手如何卸载macOS上的logitech g hub,如何卸载顽固macOS应用

1.在App Store里下载Cleaner One Pro (注意,不需要订阅付费!!!白嫖基础功能就完全够了!!!) 2.运行软件,在左侧目录中选择“应用程序管理”,然后点…...

侦探IP“去推理化”:《名侦探柯南》剧场版走过26年

2023年贺岁档,柯南剧场版的第26部《黑铁的鱼影》如期上映。 这部在日本狂卷票房128亿日元的作品,被誉为有史以来柯南剧场版在商业成绩上最好的一部。 但该作在4月份日本还未上映前,就于国内陷入了巨大的争议。 试映内容里,灰原…...

图论 经典例题

1 拓扑排序 对有向图的节点排序,使得对于每一条有向边 U-->V U都出现在V之前 *有环无法拓扑排序 indegree[], nxs[];//前者表示节点 i 的入度,后者表示节点 i 指向的节点 queue [] for i in range(n):if indege[i] 0: queue.add(i)// 入度为0的节…...

Oracle数据updater如何回滚

1.查询update语句执行的时间节点 ; select t.FIRST_LOAD_TIME, t.SQL_TEXT from v$sqlarea t where to_char(t.FIRST_LOAD_TIME) > 2023-03-19/17:00:00 order by t.FIRST_LOAD_TIME desc;开启表的行迁移 alter table test enable row movement;3.回滚表数据到…...

redis开启密码验证

开启密码验证 (1)配置文件中设置 redis.conf文件里面配置requirepass参数,redis认证密码:foobared,然后重启redis服务 ./redis-cli 127.0.0.1:6379> 127.0.0.1:6379> 127.0.0.1:6379> CONFIG SET requi…...

一种删除 KubeSphere 中一直卡在 Terminating 的 Namespace--KubeSphere Logging System的简单方法

文章目录 一、问题提出二、删除方法1,获取kubesphere-logging-syste的详细信息json文件2,编辑kubesphere-logging-system.json3,执行清理命令 三、检查结果 一、问题提出 在使用 KubeSphere 的时候发现有一个日志服务KubeSphere Logging Sys…...

<6>-MySQL表的增删查改

目录 一,create(创建表) 二,retrieve(查询表) 1,select列 2,where条件 三,update(更新表) 四,delete(删除表&#xf…...

渲染学进阶内容——模型

最近在写模组的时候发现渲染器里面离不开模型的定义,在渲染的第二篇文章中简单的讲解了一下关于模型部分的内容,其实不管是方块还是方块实体,都离不开模型的内容 🧱 一、CubeListBuilder 功能解析 CubeListBuilder 是 Minecraft Java 版模型系统的核心构建器,用于动态创…...

镜像里切换为普通用户

如果你登录远程虚拟机默认就是 root 用户,但你不希望用 root 权限运行 ns-3(这是对的,ns3 工具会拒绝 root),你可以按以下方法创建一个 非 root 用户账号 并切换到它运行 ns-3。 一次性解决方案:创建非 roo…...

新能源汽车智慧充电桩管理方案:新能源充电桩散热问题及消防安全监管方案

随着新能源汽车的快速普及,充电桩作为核心配套设施,其安全性与可靠性备受关注。然而,在高温、高负荷运行环境下,充电桩的散热问题与消防安全隐患日益凸显,成为制约行业发展的关键瓶颈。 如何通过智慧化管理手段优化散…...

让AI看见世界:MCP协议与服务器的工作原理

让AI看见世界:MCP协议与服务器的工作原理 MCP(Model Context Protocol)是一种创新的通信协议,旨在让大型语言模型能够安全、高效地与外部资源进行交互。在AI技术快速发展的今天,MCP正成为连接AI与现实世界的重要桥梁。…...

html-<abbr> 缩写或首字母缩略词

定义与作用 <abbr> 标签用于表示缩写或首字母缩略词&#xff0c;它可以帮助用户更好地理解缩写的含义&#xff0c;尤其是对于那些不熟悉该缩写的用户。 title 属性的内容提供了缩写的详细说明。当用户将鼠标悬停在缩写上时&#xff0c;会显示一个提示框。 示例&#x…...

GruntJS-前端自动化任务运行器从入门到实战

Grunt 完全指南&#xff1a;从入门到实战 一、Grunt 是什么&#xff1f; Grunt是一个基于 Node.js 的前端自动化任务运行器&#xff0c;主要用于自动化执行项目开发中重复性高的任务&#xff0c;例如文件压缩、代码编译、语法检查、单元测试、文件合并等。通过配置简洁的任务…...

Netty从入门到进阶(二)

二、Netty入门 1. 概述 1.1 Netty是什么 Netty is an asynchronous event-driven network application framework for rapid development of maintainable high performance protocol servers & clients. Netty是一个异步的、基于事件驱动的网络应用框架&#xff0c;用于…...

uniapp 字符包含的相关方法

在uniapp中&#xff0c;如果你想检查一个字符串是否包含另一个子字符串&#xff0c;你可以使用JavaScript中的includes()方法或者indexOf()方法。这两种方法都可以达到目的&#xff0c;但它们在处理方式和返回值上有所不同。 使用includes()方法 includes()方法用于判断一个字…...

R 语言科研绘图第 55 期 --- 网络图-聚类

在发表科研论文的过程中&#xff0c;科研绘图是必不可少的&#xff0c;一张好看的图形会是文章很大的加分项。 为了便于使用&#xff0c;本系列文章介绍的所有绘图都已收录到了 sciRplot 项目中&#xff0c;获取方式&#xff1a; R 语言科研绘图模板 --- sciRplothttps://mp.…...