从零开始实现一个RPC框架（五）

前言

这是系列最后一篇文章了，最后我们来为我们的rpc框架实现一个http gateway。这个功能实际上受到了rpcx的启发，基于这种方式实现一个简单的类似service mesh中的sidecar。

原理

http gateway可以接收来自客户端的http请求并将其转换为rpc请求然后交给服务端处理，再将服务端处理过后的结果通过http响应返回给客户端。
http gateway的大致原理就是将我们的RPC协议中header部分放到http header中，然后RPC协议中的body部分放到http body即可。

实现

首先我们需要定义http header中各个字段的名称：

const (HEADER_SEQ            = "rpc-header-seq"            //序号, 用来唯一标识请求或响应HEADER_MESSAGE_TYPE   = "rpc-header-message_type"   //消息类型，用来标识一个消息是请求还是响应HEADER_COMPRESS_TYPE  = "rpc-header-compress_type"  //压缩类型，用来标识一个消息的压缩方式HEADER_SERIALIZE_TYPE = "rpc-header-serialize_type" //序列化类型，用来标识消息体采用的编码方式HEADER_STATUS_CODE    = "rpc-header-status_code"    //状态类型，用来标识一个请求是正常还是异常HEADER_SERVICE_NAME   = "rpc-header-service_name"   //服务名HEADER_METHOD_NAME    = "rpc-header-method_name"    //方法名HEADER_ERROR          = "rpc-header-error"          //方法调用发生的异常HEADER_META_DATA      = "rpc-header-meta_data"      //其他元数据
)

然后我们需要启动一个http server，用来接收http请求。这里我们使用go自带的api，默认使用5080端口，如果发现端口已经被占用了，就递增端口。

func (s *SGServer) startGateway() {port := 5080ln, err := net.Listen("tcp", ":" + strconv.Itoa(port))for err != nil && strings.Contains(err.Error(), "address already in use") {port++ln, err = net.Listen("tcp", ":" + strconv.Itoa(port))}if err != nil {log.Printf("error listening gateway: %s", err.Error())}log.Printf("gateway listenning on " + strconv.Itoa(port))//避免阻塞，使用新的goroutine来执行http servergo func() {err := http.Serve(ln, s)if err != nil {log.Printf("error serving http %s", err.Error())}}()
}

接下来我们需要实现ServeHTTP函数：

func (s *SGServer) ServeHTTP(rw http.ResponseWriter, r *http.Request) {//如果url不对则直接返回if r.URL.Path != "/invoke" { rw.WriteHeader(404)return}//如果method不对则直接返回if r.Method != "POST" {rw.WriteHeader(405)return}//构造新的请求request := protocol.NewMessage(s.Option.ProtocolType)//根据http header填充request的headerrequest, err := parseHeader(request, r)if err != nil {rw.WriteHeader(400)}//根据http body填充request的datarequest, err = parseBody(request, r)if err != nil {rw.WriteHeader(400)}//构造contextctx := metadata.WithMeta(context.Background(), request.MetaData)response := request.Clone()response.MessageType = protocol.MessageTypeResponse//处理请求response = s.process(ctx, request, response)//返回相应s.writeHttpResponse(response, rw, r)
}func parseBody(message *protocol.Message, request *http.Request) (*protocol.Message, error) {data, err := ioutil.ReadAll(request.Body)if err != nil {return nil, err}message.Data = datareturn message, nil
}func parseHeader(message *protocol.Message, request *http.Request) (*protocol.Message, error) {headerSeq := request.Header.Get(HEADER_SEQ)seq, err := strconv.ParseUint(headerSeq, 10, 64)if err != nil {return nil, err}message.Seq = seqheaderMsgType := request.Header.Get(HEADER_MESSAGE_TYPE)msgType, err := protocol.ParseMessageType(headerMsgType)if err != nil {return nil, err}message.MessageType = msgTypeheaderCompressType := request.Header.Get(HEADER_COMPRESS_TYPE)compressType, err := protocol.ParseCompressType(headerCompressType)if err != nil {return nil, err}message.CompressType = compressTypeheaderSerializeType := request.Header.Get(HEADER_SERIALIZE_TYPE)serializeType, err := codec.ParseSerializeType(headerSerializeType)if err != nil {return nil, err}message.SerializeType = serializeTypeheaderStatusCode := request.Header.Get(HEADER_STATUS_CODE)statusCode, err := protocol.ParseStatusCode(headerStatusCode)if err != nil {return nil, err}message.StatusCode = statusCodeserviceName := request.Header.Get(HEADER_SERVICE_NAME)message.ServiceName = serviceNamemethodName := request.Header.Get(HEADER_METHOD_NAME)message.MethodName = methodNameerrorMsg := request.Header.Get(HEADER_ERROR)message.Error = errorMsgheaderMeta := request.Header.Get(HEADER_META_DATA)meta := make(map[string]interface{})err = json.Unmarshal([]byte(headerMeta), &meta)if err != nil {return nil, err}message.MetaData = metareturn message, nil
}func (s *SGServer) writeHttpResponse(message *protocol.Message, rw http.ResponseWriter, r *http.Request) {header := rw.Header()header.Set(HEADER_SEQ, string(message.Seq))header.Set(HEADER_MESSAGE_TYPE, message.MessageType.String())header.Set(HEADER_COMPRESS_TYPE, message.CompressType.String())header.Set(HEADER_SERIALIZE_TYPE, message.SerializeType.String())header.Set(HEADER_STATUS_CODE, message.StatusCode.String())header.Set(HEADER_SERVICE_NAME, message.ServiceName)header.Set(HEADER_METHOD_NAME, message.MethodName)header.Set(HEADER_ERROR, message.Error)metaDataJson, _ := json.Marshal(message.MetaData)header.Set(HEADER_META_DATA, string(metaDataJson))_, _ = rw.Write(message.Data)
}

最后我们只需要在wrapper中启动http server即可。

func (w *DefaultServerWrapper) WrapServe(s *SGServer, serveFunc ServeFunc) ServeFunc {return func(network string, addr string, meta map[string]interface{}) error {//省略前面的部分...//启动gateways.startGateway()return serveFunc(network, addr, meta)}
}

客户端测试代码：

func MakeHttpCall() {//声明参数并序列化，放到http请求的body中arg := service.Args{A: rand.Intn(200), B: rand.Intn(100)}data, _ := msgpack.Marshal(arg)body := bytes.NewBuffer(data)req, err := http.NewRequest("POST", "http://localhost:5080/invoke", body)if err != nil {log.Println(err)return}req.Header.Set(server.HEADER_SEQ, "1")req.Header.Set(server.HEADER_MESSAGE_TYPE, protocol.MessageTypeRequest.String())req.Header.Set(server.HEADER_COMPRESS_TYPE,protocol.CompressTypeNone.String())req.Header.Set(server.HEADER_SERIALIZE_TYPE,codec.MessagePack.String())req.Header.Set(server.HEADER_STATUS_CODE,protocol.StatusOK.String())req.Header.Set(server.HEADER_SERVICE_NAME,"Arith")req.Header.Set(server.HEADER_METHOD_NAME,"Add")req.Header.Set(server.HEADER_ERROR,"")meta := map[string]interface{}{"key":"value"}metaJson, _ := json.Marshal(meta)req.Header.Set(server.HEADER_META_DATA,string(metaJson))response, err := http.DefaultClient.Do(req)if err != nil {log.Println(err)return}if response.StatusCode != 200 {log.Println(response)} else if response.Header.Get(server.HEADER_ERROR) != "" {log.Println(response.Header.Get(server.HEADER_ERROR))} else {data, err = ioutil.ReadAll(response.Body)result := service.Reply{}msgpack.Unmarshal(data, &result)fmt.Println(result.C)}
}

结语

这个系列到此就告一段落了，但是还有很多需要改进和丰富的地方甚至是错误，后续再以单独文章的形式更新。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/800100.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

PicGo日志报错 image not found in clipboard

PicGo日志报错 image not found in clipboard

PicGo: image not found in clipboard 文章目录 PicGo: image not found in clipboard问题描述问题尝试解决方案 问题描述 背景：在剪切板中的图片无法通过 PicGo 的剪切板图片进行上传。 读取PicGo 日志报错，显示图片没有在剪切板中找到。 $ ------Erro…
阅读更多...
express接口请求的几种方式分析总结

express接口请求的几种方式分析总结

导语 在用express做接口开发的时候&#xff0c;我们要处理post,get,put,delete等请求&#xff0c;以及jsonp的方式&#xff0c;这篇文章记录下结合ajax&#xff0c;实现处理这些请求方式的过程 实现过程 上代码&#xff0c;主要演示post,get及jsonp的请求 <!DOCTYPE htm…
阅读更多...
（学习日记）2024.04.05：UCOSIII第三十三节：互斥量

（学习日记）2024.04.05：UCOSIII第三十三节：互斥量

写在前面&#xff1a; 由于时间的不足与学习的碎片化&#xff0c;写博客变得有些奢侈。 但是对于记录学习&#xff08;忘了以后能快速复习&#xff09;的渴望一天天变得强烈。 既然如此 不如以天为单位&#xff0c;以时间为顺序&#xff0c;仅仅将博客当做一个知识学习的目录&a…
阅读更多...
51单片机入门_江协科技_17~18_OB记录的笔记

51单片机入门_江协科技_17~18_OB记录的笔记

17. 定时器 17.1. 定时器介绍&#xff1a;51单片机的定时器属于单片机的内部资源&#xff0c;其电路的连接和运转均在单片机内部完成&#xff0c;无需占用CPU外围IO接口&#xff1b; 定时器作用&#xff1a; &#xff08;1&#xff09;用于计时系统&#xff0c;可实现软件计时&…
阅读更多...
数据库基础(面试常见题)

数据库基础(面试常见题)

一、数据库基础 1. 数据抽象&#xff1a;物理抽象、概念抽象、视图级抽象,内模式、模式、外模式2. SQL语言包括数据定义、数据操纵(Data Manipulation),数据控制(Data Control)数据定义&#xff1a;Create Table,Alter Table,Drop Table, Craete/Drop Index等数据操纵&#xf…
阅读更多...
C语言题目：分段函数练习

C语言题目：分段函数练习

题目描述 输入x &#xff0c;计算并输出下列分段函数 f(x) 的值。可以调用数学库函数&#xff1a;平方根函数sqrt(),绝对值函数fabs() 和幂函数 pow()。 保留2位小数 输入格式 x 输出格式 f(x) 样例输入 5 样例输出 15.00 代码分析 包含标准输入输出库和数学库: #incl…
阅读更多...
Linux -- 字符设备驱动--LED的驱动开发（初级框架）

Linux -- 字符设备驱动--LED的驱动开发（初级框架）

驱动框架一阶段 我们怎样去点亮一个 LED 呢&#xff1f;分为三步&#xff1a; 看原理图确定引脚&#xff0c;确定引脚输出什么电平才能点亮/熄灭 LED 看主芯片手册&#xff0c;确定寄存器操作方法&#xff1a;哪些寄存器&#xff1f;哪些位&#xff1f;地址是&#xff1f; 编…
阅读更多...
webpack5如何关闭全屏错误

webpack5如何关闭全屏错误

1、找到vue.config.js 2、在上面的devServer里面添加如下&#xff1a; client: {overlay: false, // 禁用全局错误提示},
阅读更多...
mynet开源库

mynet开源库

1.介绍 个人实现的c开源网络库&#xff0e; 2.软件架构 1.结构图 2.基于event的自动分发机制 3.多优先级分发队列&#xff0c;延迟分发队列 内部event服务于通知机制的优先级为0&#xff0c;外部event优先级为1&#xff0e; 当集中处理分发的event_callback时&#xff0c…
阅读更多...
更新！谷歌倾斜摄影转换生成OSGB瓦片V0.2版

更新！谷歌倾斜摄影转换生成OSGB瓦片V0.2版

半个月前发表了一篇文章(首发&#xff01;谷歌倾斜摄影转换生成OSGB格式),首次将谷歌原始倾斜摄影瓦片转成OSGB瓦片&#xff0c;有读者对文章里的内容有诸多疑问&#xff0c;可能我在文章中描述的不够清楚&#xff0c;这里再解释一下。 1.谷歌倾斜摄影3DTiles瓦片生成方案 之…
阅读更多...
若依ts版本（vue3+element plus+ts）

若依ts版本（vue3+element plus+ts）

1、项目简介 本项目参考若依前后端分离版&#xff0c;前端由[若依vue3]改写为ts版本[ruoyi-web-vue3-ts]&#xff0c;后端对[若依V3.8.7]进行了修改[后端版本分支vue3.ts.3.8.7]&#xff0c;具体文档参见[若依官方文档]。本项目对部分代码做了优化&#xff0c;增加了activiti7…
阅读更多...
langchain 学习笔记-FunctionCalling三种方式

langchain 学习笔记-FunctionCalling三种方式

ChatGPT 基于海量的训练数据生成答案&#xff0c;所以它无法回答训练数据中没有的信息或搜索信息 。人们希望 ChatGPT 具有对话以外的各种功能&#xff0c;例如“我想管理我的待办事项列表”。 函数调用是对此类请求的响应。 通过使用函数调用&#xff0c;ChatGPT 现在可以在生…
阅读更多...
LeetCode-215. 数组中的第K个最大元素【数组 分治 快速选择 排序 堆（优先队列）】

LeetCode-215. 数组中的第K个最大元素【数组 分治 快速选择 排序 堆（优先队列）】

LeetCode-215. 数组中的第K个最大元素【数组 分治 快速选择 排序 堆&#xff08;优先队列&#xff09;】 题目描述&#xff1a;解题思路一&#xff1a;排序解题思路二&#xff1a;快速选择解题思路三&#xff1a;0 题目描述&#xff1a; 给定整数数组 nums 和整数 k&#xff0…
阅读更多...
AI技术创业机会之智慧城市与智慧交通

AI技术创业机会之智慧城市与智慧交通

人工智能（AI）技术的创新与发展为智慧城市与智慧交通领域带来了革命性的变革，为创业者创造了大量创新与创业机会。以下详述了智慧城市与智慧交通背景下AI技术的创业机会及其具体细节与内容，以5000字篇幅深入剖析各细分领域，为有志于投身这一领域的创业者提供全面、深入的商…
阅读更多...
Vue-ts项目导入Js文件方法-书写提示无法找到模块“XXXX”的声明文件-配置说明

Vue-ts项目导入Js文件方法-书写提示无法找到模块“XXXX”的声明文件-配置说明

提示 无法找到模块“/filter/filter”的声明文件。“d:/code/byWork/mixmarvel/SendAssetsClient/src/filter/filter.js”隐式拥有 "any" 类型。 解决方法 修改 tsconfig.json {"compilerOptions": {"target": "esnext","modu…
阅读更多...
OneFlow深度学习框架介绍

OneFlow深度学习框架介绍

OneFlow是一个开源的深度学习框架&#xff0c;旨在为用户提供高效、易用的工具来进行深度学习模型的开发、训练和部署。以下是一些OneFlow框架的特点和优势&#xff1a; 高效性能&#xff1a; OneFlow针对大规模深度学习模型进行了优化&#xff0c;具有优秀的计算性能和内存管理…
阅读更多...
Linux Shell 比较运算有哪些，以及怎么用

Linux Shell 比较运算有哪些，以及怎么用

Linux Shell 比较运算有哪些&#xff0c;以及怎么用 在 Linux Shell 脚本编程中&#xff0c;进行算术比较是常见的需求&#xff0c;特别是在编写条件判断语句时。Shell 提供了一系列的算术比较操作符&#xff0c;这些操作符通常用在 if、while 等语句中&#xff0c;以便进行数…
阅读更多...
程序员如何搞副业？

程序员如何搞副业？

程序员不仅拥有将抽象概念转化为实际应用的能力&#xff0c;还通常具备强大的逻辑思维和问题解决能力。然而&#xff0c;许多程序员并不满足于仅仅在一家公司工作&#xff0c;他们渴望通过副业来实现个人价值的最大化&#xff0c;增加收入&#xff0c;甚至探索自己的创业梦想。…
阅读更多...
前端及nodejs后端面试经验

前端及nodejs后端面试经验

目录 前端双向绑定如何实现自定义指令如何实现promise的执行顺序父子组件的信息传递cookies&#xff0c;session&#xff0c;webStoragevuex回流和重绘指令watch和watchEffect的区别watch和computed的区别vue3新功能Composition APIproxy和definePropertyTeleport 请求post和pu…
阅读更多...
RIP配置不求人：手把手教你配置RIP路由

RIP配置不求人：手把手教你配置RIP路由

#教育优质作者发文挑战赛# 大家好&#xff0c;今天给同学们介绍一下RIP基本功能相关配置 01、基本概念 RIP是一种基于距离矢量&#xff08;Distance-Vector&#xff09;算法的协议&#xff0c;它使用跳数&#xff08;Hop Count&#xff09;作为度量值来衡量到达目的地址的距离…
阅读更多...
最新文章

Copyright @ 2022~2023