GoLong的学习之路,进阶,微服务之使用,RPC包(包括源码分析)

今天这篇是接上上篇RPC原理之后这篇是讲如何使用go本身自带的标准库RPC。这篇篇幅会比较短。重点在于上一章对的补充。

文章目录

  • RPC包的概念
  • 使用RPC包
  • 服务器代码分析
    • 如何实现的?
      • 总结
      • Server还提供了两个注册服务的方法
  • 客户端代码分析
    • 如何实现的?
      • 如何异步编程同步?
      • 总结
  • codec/序列化框架
    • 使用JSON协议的RPC

RPC包的概念

回顾RPC原理

看完回顾后其实就可以继续需了解并使用go中所提供的包。

Go语言的 rpc 包提供对通过网络或其他i/o连接导出的对象方法的访问,服务器注册一个对象,并把它作为服务对外可见(服务名称就是类型名称)。

注册后,对象的导出方法将支持远程访问。服务器可以注册不同类型的多个对象(服务) ,但是不支持注册同一类型的多个对象。

Go官方提供了一个RPC库: net/rpc

rpc提供了通过网络访问一个对象的输出方法的能力。

服务器需要注册对象,通过对象的类型名暴露这个服务。

注册后这个对象的输出方法就可以远程调用,这个库封装了底层传输的细节,包括序列化(默认GOB序列化器)。

对象的方法要能远程访问,它们必须满足一定的条件,否则这个对象的方法会被忽略:

  • 方法的类型是可输出的
  • 方法本身是可输出的
  • 方法必须由两个参数,必须是输出类型或者是内建类型
  • 方法的第二个参数必须是指针类型
  • 方法返回类型为error

所以一个输出方法的格式如下:

func (t *T) MethodName(argType T1, replyType *T2) error

这里的TT1T2能够被encoding/gob序列化,即使使用其它的序列化框架,将来这个需求可能回被弱化。

  • 第一个参数(T1)代表调用者(client)提供的参数
  • 第二个参数(*T2)代表要返回给调用者的计算结果
  • 方法的返回值如果不为空, 那么它作为一个字符串返回给调用者(所以需要一个序列化框架)
  • 如果返回error,则reply参数不会返回给调用者

使用RPC包

简单例子,是一个非常简单的服务。

在这里插入图片描述
我们在这个里面就搞112就好:

在这个例子中定义了一个简单的RPC服务器和客户端,使用的方法是一个

第一步需要定义传入参数和返回参数的数据结构

type Args struct {A, B int
}
type Quotient struct {Quo, Rem int
}

第二步定义一个服务对象,这个服务对象可以很简单。

比如类型是int或者是interface{},重要的是它输出的方法。

type Arith int

第三步实现这个类型的两个方法, 乘法和除法:

func (t *Arith) Multiply(args *Args, reply *int) error {*reply = args.A * args.Breturn nil
}
func (t *Arith) Divide(args *Args, quo *Quotient) error {if args.B == 0 {return errors.New("divide by zero")}quo.Quo = args.A / args.Bquo.Rem = args.A % args.Breturn nil
}

第四步实现RPC服务器: 基于tcp实现

生成了一个Arith对象,并使用rpc.Register注册这个服务,然后通过HTTP暴露出来

arith := new(Arith)
rpc.Register(arith)
rpc.HandleHTTP()
l, e := net.Listen("tcp", ":9091")
if e != nil {log.Fatal("listen error:", e)
}
go http.Serve(l, nil)
select{
}

客户端可以看到服务Arith以及它的两个方法Arith.MultiplyArith.Divide

第五步创建一个客户端,建立客户端和服务器端的连接: 分为同步调用和异步调用(都是远程调用)

同步调用:

client, err := rpc.DialHTTP("tcp", "127.0.0.1:9091")
if err != nil {log.Fatal("dialing:", err)
}args := &server.Args{7,8}
var reply int
err = client.Call("Arith.Multiply", args, &reply)
if err != nil {log.Fatal("arith error:", err)
}
fmt.Printf("Arith: %d*%d=%d", args.A, args.B, reply)

异步调用:

client, err := rpc.DialHTTP("tcp", "127.0.0.1:9091")
if err != nil {log.Fatal("dialing:", err)
}
quotient := new(Quotient)
divCall := client.Go("Arith.Divide", args, quotient, nil)
replyCall := <-divCall.Done    // will be equal to divCall
// check errors, print, etc.

完整的例子:

  1. 创建一个service.go 的文件用来保存结构体对象以及方法
package mainimport "errors"type Args struct {A, B int
}type Quotient struct {Quo, Rem int
}type Arith intfunc (t *Arith) Multiply(args *Args, reply *int) error {*reply = args.A * args.Breturn nil
}
func (t *Arith) Divide(args *Args, quo *Quotient) error {if args.B == 0 {return errors.New("divide by zero")}quo.Quo = args.A / args.Bquo.Rem = args.A % args.Breturn nil
}
  1. 创建一个RPC服务端,server.go
package mainimport ("log""net""net/http""net/rpc"
)func main() {arith := new(Arith)rpc.Register(arith)rpc.HandleHTTP()l, e := net.Listen("tcp", ":9091")if e != nil {log.Fatal("listen error:", e)}go http.Serve(l, nil)select {}
}
  1. 创建一个客户端,client.go
package mainimport ("fmt""log""net/rpc"
)func main() {// 建立HTTP连接client, err := rpc.DialHTTP("tcp", "127.0.0.1:9091")if err != nil {log.Fatal("dialing:", err)}// 同步调用args := &Args{7, 8}var reply interr = client.Call("Arith.Multiply", args, &reply)if err != nil {log.Fatal("arith error:", err)}fmt.Printf("Arith: %d*%d=%d", args.A, args.B, reply)// 异步调用quotient := new(Quotient)divCall := client.Go("Arith.Divide", args, quotient, nil)replyCall := <-divCall.Done // will be equal to divCall// check errors, print, etc.fmt.Println(replyCall.Error)fmt.Println(quotient)
}

打开终端:

先启动服务器:

go run server.go service.go

在打开一个终端:

最后启动一个客户端:

go run client.go service.go

结果为:
在这里插入图片描述

服务器代码分析

Server很多方法你可以直接调用,这对于一个简单的Server的实现更方便,但是你如果需要配置不同的Server,

比如不同的监听地址或端口,就需要自己生成Server:

var DefaultServer = NewServer()

Server多种Socket监听的方式:

    func (server *Server) Accept(lis net.Listener)func (server *Server) HandleHTTP(rpcPath, debugPath string)func (server *Server) ServeCodec(codec ServerCodec)func (server *Server) ServeConn(conn io.ReadWriteCloser)func (server *Server) ServeHTTP(w http.ResponseWriter, req *http.Request)func (server *Server) ServeRequest(codec ServerCodec) error
  • ServeHTTP: 实现了处理 http请求的业务逻辑,它首先处理httpCONNECT请求, 接收后就Hijacker这个连接conn, 然后调用ServeConn在这个连接上处理这个客户端的请求。
    • 其实是实现了 http.Handler接口,我们一般不直接调用这个方法。
      • Server.HandleHTTP设置rpc的上下文路径
      • rpc.HandleHTTP使用默认的上下文路径`
      • DefaultRPCPath
      • DefaultDebugPath
    • 当你启动一个http server的时候 http.ListenAndServe,面设置的上下文将用作RPC传输,这个上下文的请求会教给ServeHTTP来处理
    • 以上是RPC over http的实现,可以看出 net/rpc只是利用 http CONNECT建立连接,这和普通的 RESTful api还是不一样的。
    • (源码)
func (server *Server) ServeHTTP(w http.ResponseWriter, req *http.Request) {if req.Method != "CONNECT" {w.Header().Set("Content-Type", "text/plain; charset=utf-8")w.WriteHeader(http.StatusMethodNotAllowed)io.WriteString(w, "405 must CONNECT\n")return}conn, _, err := w.(http.Hijacker).Hijack()if err != nil {log.Print("rpc hijacking ", req.RemoteAddr, ": ", err.Error())return}io.WriteString(conn, "HTTP/1.0 "+connected+"\n\n")server.ServeConn(conn)
}

如何实现的?

Accept用来处理一个监听器,一直在监听客户端的连接,一旦监听器接收了一个连接,则还是交给ServeConn在另外一个goroutine中去处理:(源码)

//Accept接受侦听器上的连接并提供请求
//每个传入连接。接受阻塞,直到监听器
//返回非nil错误。对象中调用Accept
//go语句
func (server *Server) Accept(lis net.Listener) {for {conn, err := lis.Accept()if err != nil {log.Print("rpc.Serve: accept:", err.Error())return}go server.ServeConn(conn)}
}

协程进入ServeConn可以看出,很重要的一个方法就是ServeConn

// ServeConn在单连接上运行服务器。
// ServeConn阻塞,服务连接,直到客户端挂起。
//调用者通常在go语句中调用ServeConn。
// ServeConn使用gob连接格式(参见包gob)
//连接。要使用备用编解码器,请使用ServeCodec。
//有关并发访问的信息,请参阅NewClient的注释。.
func (server *Server) ServeConn(conn io.ReadWriteCloser) {buf := bufio.NewWriter(conn)srv := &gobServerCodec{rwc:    conn,dec:    gob.NewDecoder(conn),enc:    gob.NewEncoder(buf),encBuf: buf,}server.ServeCodec(srv)
}

连接其实是交给一个ServerCodec去处理,这里默认使用gobServerCodec去处理,这是一个未输出默认的编解码器,可以使用其它的编解码器。

// ServeCodec类似于ServeConn,但使用指定的编解码器来
//解码请求和编码响应。
func (server *Server) ServeCodec(codec ServerCodec) {sending := new(sync.Mutex)wg := new(sync.WaitGroup)for {service, mtype, req, argv, replyv, keepReading, err := server.readRequest(codec)if err != nil {if debugLog && err != io.EOF {log.Println("rpc:", err)}if !keepReading {break}// send a response if we actually managed to read a header.if req != nil {server.sendResponse(sending, req, invalidRequest, codec, err.Error())server.freeRequest(req)}continue}wg.Add(1)go service.call(server, sending, wg, mtype, req, argv, replyv, codec)}//我们已经看到没有更多的请求。//在关闭编解码器之前等待响应。wg.Wait()codec.Close()
}

它其实一直从连接中读取请求,然后调用go service.call在另外的goroutine中处理服务调用。

总结

  1. 对象重用。 RequestResponse都是可重用的,通过Lock处理竞争。这在大并发的情况下很有效。

  2. 使用了大量的goroutine。如果使用一定数量的goroutine作为worker池去处理这个case,可能还会有些性能的提升,但是更复杂了。使用goroutine可以获得了非常好的性能。

  3. 业务处理是异步的,服务的执行不会阻塞其它消息的读取。

  4. 一个codec实例必然和一个connnection相关,因为它需要从connection中读取request和发送response

go的rpc官方库的消息(requestresponse)的定义很简单, 就是消息头(header)+内容体(body)。
消息体是reply类型的序列化后的值。

type Request struct {ServiceMethod string // format: "Service.Method"Seq           uint64 // 客户端选择的序列号// 包含过滤或未导出的字段
}

Server还提供了两个注册服务的方法

第二个方法为服务起一个别名,否则服务名已它的类型命名

    func (server *Server) Register(rcvr interface{}) errorfunc (server *Server) RegisterName(name string, rcvr interface{}) error

它们俩底层调用register进行服务的注册(这里的源码太多就不放了)

func (server *Server) register(rcvr interface{}, name string, useName bool) error

受限于Go语言的特点,我们不可能在接到客户端的请求的时候,根据反射动态的创建一个对象,就是Java那样。

因此在Go语言中,我们需要预先创建一个服务map这是在编译的时候完成的
说白了这里需要建立一个注册名与服务之间的映射关系

server.serviceMap = make(map[string]*service)

同时每个服务还有一个方法map: map[string]*methodType,通过suitableMethods建立映射:

func suitableMethods(typ reflect.Type, reportErr bool) map[string]*methodType

这样rpc在读取请求header,通过查找这两个map,就可以得到要调用的服务及它的对应方法了。

func (s *service) call(server *Server, sending *sync.Mutex, wg *sync.WaitGroup, mtype *methodType, req *Request, argv, replyv reflect.Value, codec ServerCodec) {if wg != nil {defer wg.Done()}mtype.Lock()mtype.numCalls++mtype.Unlock()function := mtype.method.Func// 调用该方法,为应答提供一个新值。returnValues := function.Call([]reflect.Value{s.rcvr, argv, replyv})// 该方法的返回值是一个错误。.errInter := returnValues[0].Interface()errmsg := ""if errInter != nil {errmsg = errInter.(error).Error()}server.sendResponse(sending, req, replyv.Interface(), codec, errmsg)server.freeRequest(req)
}

客户端代码分析

客户端要建立和服务器的连接

 	func Dial(network, address string) (*Client, error)func DialHTTP(network, address string) (*Client, error)func DialHTTPPath(network, address, path string) (*Client, error)func NewClient(conn io.ReadWriteCloser) *Clientfunc NewClientWithCodec(codec ClientCodec) *Client

如何实现的?

DialHTTPDialHTTPPath是通过HTTP的方式和服务器建立连接,他俩的区别之在于是否设置上下文路径:

// DialHTTPPath连接HTTP RPC服务器在指定的网络地址和路径上
func DialHTTPPath(network, address, path string) (*Client, error) {conn, err := net.Dial(network, address)if err != nil {return nil, err}io.WriteString(conn, "CONNECT "+path+" HTTP/1.0\n\n")// 在切换到RPC协议之前,需要成功的HTTP响应resp, err := http.ReadResponse(bufio.NewReader(conn), &http.Request{Method: "CONNECT"})if err == nil && resp.Status == connected {return NewClient(conn), nil}if err == nil {err = errors.New("unexpected HTTP response: " + resp.Status)}conn.Close()return nil, &net.OpError{Op:   "dial-http",Net:  network + " " + address,Addr: nil,Err:  err,}
}

首先发送 CONNECT 请求,如果连接成功则通过NewClient(conn)创建client

Dial则通过TCP直接连接服务器

// Dial连接到指定网络地址的RPC服务器
func Dial(network, address string) (*Client, error) {conn, err := net.Dial(network, address)if err != nil {return nil, err}return NewClient(conn), nil
}

注意:根据服务是over HTTP还是 over TCP选择合适的连接方式

NewClient则创建一个缺省codecglob序列化库的客户端

// NewClient返回一个新的Client来处理到连接另一端的服务集合。
//在连接的写端添加一个缓冲区,报头和有效载荷作为一个单元发送。
//
//连接的读写部分是独立序列化的,不需要联锁。然而,每一半都可以访问并发,所以conn的实现应该防止,并发读或并发写。
func NewClient(conn io.ReadWriteCloser) *Client {encBuf := bufio.NewWriter(conn)client := &gobClientCodec{conn, gob.NewDecoder(conn), gob.NewEncoder(encBuf), encBuf}return NewClientWithCodec(client)
}

如果想用其它的序列化库,你可以调用NewClientWithCodec方法 一般用来做RPC框架的

// NewClientWithCodec类似于NewClient,但使用指定的编码请求和解码响应。
func NewClientWithCodec(codec ClientCodec) *Client {client := &Client{codec:   codec,pending: make(map[uint64]*Call),}go client.input()return client
}

重要的是input方法,它以一个死循环的方式不断地从连接中读取response,然后调用map中读取等待的Call.Done channel通知完成。(这个其实有点令牌扫描的作用,消息队列中有说)

消息的结构和服务器一致,都是Header+Body的方式

客户端的调用有两个方法: GoCall

  • Go方法是异步的,它返回一个 Call指针对象, 它的Done是一个channel,如果服务返回,Done就可以得到返回的对象(实际是Call对象,包含Replyerror信息)。
  • Call 方法是同步的,它实际是调用Go实现的。

如何异步编程同步?

func (client *Client) Call(serviceMethod string, args interface{}, reply interface{}) error {call := <-client.Go(serviceMethod, args, reply, make(chan *Call, 1)).Donereturn call.Error
}

从一个Channel中读取对象会被阻塞住,直到有对象可以读取,这种实现很简单,也很方便。

总结

从源码中:我们还可以学到锁Lock的一种实用方式,也就是尽快的释放锁,而不是defer mu.Unlock直到函数执行到最后才释放,那样占用的时间太长了。


codec/序列化框架

rpc框架默认使用gob序列化库,很多情况下我们追求更好的效率的情况下,或者追求更通用的序列化格式,我们可能采用其它的序列化方式, 比如protobuf, json, xml等。

市面上也有许多序列化框架。速度快而且非常好用。gRPC是互联网后台常用的RPC框架,其内部是由protobuf协议完成通讯。这个后面再讲。

JDK SerializableFSTKryoProtobufThriftHessionAvroFury
在这里插入图片描述

Fury是最新的序列化框架:号称比jdk 快170倍,后面会讲的 支持多种语言

Go官方库实现了JSON-RPC 1.0JSON-RPC是一个通过JSON格式进行消息传输的RPC规范,因此可以进行跨语言的调用。

Go的net/rpc/jsonrpc库可以将JSON-RPC的请求转换成自己内部的格式:

func (c *serverCodec) ReadRequestHeader(r *rpc.Request) error {c.req.reset()if err := c.dec.Decode(&c.req); err != nil {return err}r.ServiceMethod = c.req.Methodc.mutex.Lock()c.seq++c.pending[c.seq] = c.req.Idc.req.Id = nilr.Seq = c.seqc.mutex.Unlock()return nil
}

使用JSON协议的RPC

rpc 包默认使用的是 gob 协议对传输数据进行序列化/反序列化,比较有局限性

将例子进行修改:
服务器端:

package mainimport ("log""net""net/rpc""net/rpc/jsonrpc"
)func main() {arith := new(Arith)rpc.Register(arith)l, e := net.Listen("tcp", ":9091")if e != nil {log.Fatal("listen error:", e)}for {conn, _ := l.Accept()// 使用JSON协议rpc.ServeCodec(jsonrpc.NewServerCodec(conn))}
}

客户端:

package mainimport ("fmt""log""net""net/rpc""net/rpc/jsonrpc"
)func main() {// 建立HTTP连接conn, err := net.Dial("tcp", "127.0.0.1:9091")if err != nil {log.Fatal("dialing:", err)}client := rpc.NewClientWithCodec(jsonrpc.NewClientCodec(conn))// 同步调用args := &Args{7, 8}var reply interr = client.Call("Arith.Multiply", args, &reply)if err != nil {log.Fatal("arith error:", err)}fmt.Printf("Arith: %d*%d=%d", args.A, args.B, reply)// 异步调用quotient := new(Quotient)divCall := client.Go("Arith.Divide", args, quotient, nil)replyCall := <-divCall.Done // will be equal to divCall// check errors, print, etc.fmt.Println(replyCall.Error)fmt.Println(quotient)
}

如何使用与上面的例子一致。

社区中各式RPC框架(grpc、thrift等)就是为了让RPC调用更方便。

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

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

相关文章

Android Studio连接MYSQL数据库

首先导入mysql的jar包&#xff0c;这里连接的是8版本的。 这里之前到如果mysql的jar包了 首先跳到Project模式&#xff1a; 直接复制粘贴到这里&#xff1a; 这里之前到如果了。想删掉重新导入一次&#xff0c;但是报错,什么ioexception。这里将Project Structure中的Moudle中的…

【算法Hot100系列】两数之和

&#x1f49d;&#x1f49d;&#x1f49d;欢迎来到我的博客&#xff0c;很高兴能够在这里和您见面&#xff01;希望您在这里可以感受到一份轻松愉快的氛围&#xff0c;不仅可以获得有趣的内容和知识&#xff0c;也可以畅所欲言、分享您的想法和见解。 推荐:kwan 的首页,持续学…

【动手学深度学习】(十二)现代卷积神经网络

文章目录 一、深度卷积神经网络AlexNet1.理论知识 一、深度卷积神经网络AlexNet 1.理论知识 ImageNet(2010) 图片自然物体的彩色图片手写数字的黑色图片大小468 * 38728*28样本数1.2M60K类数100010 AlexNet AlexNet赢了2012ImageNet竞赛更深更大的LeNet主要改进&#xff…

漏刻有时百度地图API实战开发(9)Echarts使用bmap.js实现轨迹动画效果

Bmap.js是Echarts和百度地图相结合开发的一款JavaScript API&#xff0c;它可以帮助用户在web应用中获取包括地图中心点、地图缩放级别、地图当前视野范围、地图上标注点等在内的地图信息&#xff0c;并且支持在地图上添加控件&#xff0c;提供包括智能路线规划、智能导航(驾车…

web,Apache简述

一.HTTP请求访问的完整过程 1.建立连接 2.接收请求 3.处理请求 4.访问资源 服务器获取请求报文中请求的资源web服务器&#xff0c;即存放了web资源的服务器&#xff0c;负责向请求者提供对方请求的静态资源&#xff0c;或动态运行后生成的资源 静态资源&#xff1a;不需要…

Redis 环境搭建2

文章目录 第2关&#xff1a;使用 Redis 第2关&#xff1a;使用 Redis 本文是接着上篇文章写的第二关代码&#xff0c;部分人再进入第二关时不会保留第一关的配置的环境&#xff0c;可以通过下面一句代码进行检验。 redis-cli -p 7001 -c如果进入到了redis界面就是有环境&…

【9】Spring Boot 3 集成组件 : 基于spring security的认证权限开发【认证】

目录 【9】Spring Boot 3 集成组件 &#xff1a; 基于spring security的用户权限系统【认证】认证和权限流程【理论】认证授权鉴权权限控制认证、授权、鉴权、和权限控制关系认证和鉴权的关系 认证流程【理论】1. 直接携带认证令牌访问资源2. Session认证机制3. Jwt认证机制下的…

python+pytest接口自动化(12)-自动化用例编写思路 (使用pytest编写一个测试脚本)

经过之前的学习铺垫&#xff0c;我们尝试着利用pytest框架编写一条接口自动化测试用例&#xff0c;来厘清接口自动化用例编写的思路。 我们在百度搜索天气查询&#xff0c;会出现如下图所示结果&#xff1a; 接下来&#xff0c;我们以该天气查询接口为例&#xff0c;编写接口测…

Unity 实现单例模式

目录 基本概念 饿汉模式(推荐) 懒汉模式&#xff1a; 基本概念 单例模式&#xff1a;类只有一个实例&#xff0c;一般使用static来实现单例模式&#xff1b; 比如&#xff1a;有一个Test类,实现了单例&#xff0c;假设这个唯一的实例名为SingTonle,实例在类内被实现并被stat…

【pycharm】Pycharm中进行Git版本控制

本篇文章主要记录一下自己在pycharm上使用git的操作&#xff0c;一个新项目如何使用git进行版本控制。 文章使用的pycharm版本PyCharm Community Edition 2017.2.4&#xff0c;远程仓库为https://gitee.com/ 1.配置Git&#xff08;File>Settings&#xff09; 2.去Gitee创建…

微服务——服务保护Sentinel

雪崩问题 在单体项目里面&#xff0c;如果某一个模块出问题会导致整个项目都有问题。 在微服务项目里面&#xff0c;单独一个服务出问题理论上是不会影响别的服务的。 但是如果有别的业务需要调用这一个模块的话还是会有问题。 问题产生原因和解决思路 最初那只是一个小小…

持续集成交付CICD:通过API方式上传Nexus制品

目录 一、实验 1.通过API方式上传Nexus制品 二、问题 1.如何通过API方式上传PNG图片 2.如何通过API方式上传tar.gz 与 ZIP文件 3.如何通过API方式上传Jar file文件 4.如何通过API方式上传制品&#xff08;maven类型的制品&#xff09;文件 5.如何下载制品 一、实验 1.通…