单体应用体量越来越大，代码不好维护和管理，所以就产生了微服务架构，按照公共或功能模块拆分为一个个独立的服务，然后各独立的服务之间可以相互调用。
微服务之间相互调用，该如何实现？
首先要解决下面5个问题：
1、如何规定远程调用的语法？
2、如何传递参数？
3、如何表示数据？
4、如何知道一个服务端都实现了哪些远程调用？从哪个端口可以访问这个远程调用？
5、发生了错误、重传、丢包、性能等问题怎么办？
大家可能都写过socket或则http通信，简单的client访问server的模式，认为通过这个就可以解决服务之间的相互调用了，但是考虑下上面5个问题，处理起来就不是那么容易的事情了，非个人可以完成的工作。
于是就诞生了rpc框架，让我们不用管底层实现，简单好用：
当客户端的应用想发起一个远程调用时，它实际是调用客户端的 stub。它负责将调用的接口、方法和参数，通过约定的协议规范进行编码，并通过本地的 rpcruntime 进行传输，将调用网络包发送到服务器。服务器端的 rpcruntime 收到请求后，交给服务器端的 stub 进行解码，然后调用服务端的方法，服务端执行方法，返回结果，服务器端的 stub 将返回结果编码后，发送给客户端，客户端的 rpcruntime 收到结果，发给客户端的 stub 解码得到结果，返回给客户端。
1、对于客户端而言，这些过程是透明的，就像本地调用一样；对于服务端而言，专注于业务逻辑的处理就可以了。
2、对于 stub 层，处理双方约定好的语法、语义、封装、解封装。
3、对于 rpcruntime，主要处理高性能的传输，以及网络的错误和异常。
来看一下rpc框架是如何解决上面5个问题的：1、2、3的问题可以由stub层解决，4的问题可以由服务注册和发布解决，5的问题可以由rpcruntime解决。
grpc 是一个高性能、开源和通用的 rpc 框架，面向移动和 http/2 设计。目前提供 c、java 和 go 语言版本，分别是：grpc, grpc-java, grpc-go. 其中 c 版本支持 c, c++, node.js, python, ruby, objective-c, php 和 c# 支持。
本文以go语言版本讲解
1、golang安装grpc
2、protocol buffer原理文章
3、grpc-go github地址:
grpc-go 的stub层协议约定问题通过.proto文件约定好服务接口、参数等，通过工具protoc-gen-go生成客户端和服务端共用的对照表，想生成什么语言文件就用相应的插件，这样就实现了跨语言。
生成go语言文件使用命令如下：
protoc --go_out=plugins=grpc:. *.proto
grpc rpcruntime层基于http/2设计，带来诸如双向流、流控、头部压缩、单 tcp 连接上的多复用请求等特性。
我要评论func main() {
//解析运行参数
flag.parse()
//配置监听协议、地址、端口
lis, err := net.listen("tcp", fmt.sprintf("localhost:%d", *port))
if err != nil {
log.fatalf("failed to listen: %v", err)
}
//grpc额外的服务配置，这里主要是需不需要加密
var opts []grpc.serveroption
if *tls {
if *certfile == "" {
*certfile = testdata.path("server1.pem")
}
if *keyfile == "" {
*keyfile = testdata.path("server1.key")
}
creds, err := credentials.newservertlsfromfile(*certfile, *keyfile)
if err != nil {
log.fatalf("failed to generate credentials %v", err)
}
opts = []grpc.serveroption{grpc.creds(creds)}
}
//grpc服务初始化，绑定一些配置参数
grpcserver := grpc.newserver(opts...)
//把.proto文件中定义的接口api实现注册到grpc服务上，方便调用
pb.registerrouteguideserver(grpcserver, newserver())
//grpc服务启动，开始监听
grpcserver.serve(lis)
}
关键处理就是一个for循环。如果accept() 返回错误，并且错误是临时性的，那么会有重试，重试时间以5ms翻倍增长，直到1s。
for {
rawconn, err := lis.accept()
//错误处理
if err != nil {
if ne, ok := err.(interface {
temporary() bool
}); ok && ne.temporary() {
if tempdelay == 0 {
tempdelay = 5 * time.millisecond
} else {
tempdelay *= 2
}
if max := 1 * time.second; tempdelay > max {
tempdelay = max
}
s.mu.lock()
s.printf("accept error: %v; retrying in %v", err, tempdelay)
s.mu.unlock()
timer := time.newtimer(tempdelay)
select {
case <-timer.c:
case <-s.quit.done():
timer.stop()
return nil
}
continue
}
s.mu.lock()
s.printf("done serving; accept = %v", err)
s.mu.unlock()
if s.quit.hasfired() {
return nil
}
return err
}
tempdelay = 0
// start a new goroutine to deal with rawconn so we don't stall this accept
// loop goroutine.
//
// make sure we account for the goroutine so gracefulstop doesn't nil out
// s.conns before this conn can be added.
s.servewg.add(1)
//重新启动一个goroutine处理accept的连接
go func() {
s.handlerawconn(rawconn)
s.servewg.done()
}()
}
主要作用就是获取一个服务端的transport，并开一个goroutine等待处理stream，里面会涉及到调用注册的方法。
st := s.newhttp2transport(conn, authinfo)
if st == nil {
return
}
rawconn.setdeadline(time.time{})
if !s.addconn(st) {
return
}
go func() {
s.servestreams(st)
s.removeconn(st)
}()
1、建立连接和绑定实现的接口
//解析运行参数
flag.parse()
//连接的一些配置，主要是加密，安全、阻塞
var opts []grpc.dialoption
if *tls {
if *cafile == "" {
*cafile = testdata.path("ca.pem")
}
creds, err := credentials.newclienttlsfromfile(*cafile, *serverhostoverride)
if err != nil {
log.fatalf("failed to create tls credentials %v", err)
}
opts = append(opts, grpc.withtransportcredentials(creds))
} else {
opts = append(opts, grpc.withinsecure())
}
opts = append(opts, grpc.withblock())
//建立一个连接
conn, err := grpc.dial(*serveraddr, opts...)
if err != nil {
log.fatalf("fail to dial: %v", err)
}
defer conn.close()
//创建一个实现了.proto文件定义的接口api的client
client := pb.newrouteguideclient(conn)
2、client调用方式
func (c *routeguideclient) getfeature(ctx context.context, in *point, opts ...grpc.calloption) (*feature, error) {
out := new(feature)
err := c.cc.invoke(ctx, "/routeguide.routeguide/getfeature", in, out, opts...)
if err != nil {
return nil, err
}
return out, nil
}
// printfeature gets the feature for the given point.
func printfeature(client pb.routeguideclient, point *pb.point) {
log.printf("getting feature for point (%d, %d)", point.latitude, point.longitude)
ctx, cancel := context.withtimeout(context.background(), 10*time.second)
defer cancel()
feature, err := client.getfeature(ctx, point)
if err != nil {
log.fatalf("%v.getfeatures(_) = _, %v: ", client, err)
}
log.println(feature)
}
// getfeature returns the feature at the given point.
func (s *routeguideserver) getfeature(ctx context.context, point *pb.point) (*pb.feature, error) {
for _, feature := range s.savedfeatures {
if proto.equal(feature.location, point) {
return feature, nil
}
}
// no feature was found, return an unnamed feature
return &pb.feature{location: point}, nil
}
func (c *routeguideclient) listfeatures(ctx context.context, in *rectangle, opts ...grpc.calloption) (routeguide_listfeaturesclient, error) {
stream, err := c.cc.newstream(ctx, &_routeguide_servicedesc.streams[0], "/routeguide.routeguide/listfeatures", opts...)
if err != nil {
return nil, err
}
x := &routeguidelistfeaturesclient{stream}
if err := x.clientstream.sendmsg(in); err != nil {
return nil, err
}
if err := x.clientstream.closesend(); err != nil {
return nil, err
}
return x, nil
}
// printfeatures lists all the features within the given bounding rectangle.
func printfeatures(client pb.routeguideclient, rect *pb.rectangle) {
log.printf("looking for features within %v", rect)
ctx, cancel := context.withtimeout(context.background(), 10*time.second)
defer cancel()
stream, err := client.listfeatures(ctx, rect)
if err != nil {
log.fatalf("%v.listfeatures(_) = _, %v", client, err)
}
for {
feature, err := stream.recv()
if err == io.eof {
break
}
if err != nil {
log.fatalf("%v.listfeatures(_) = _, %v", client, err)
}
log.println(feature)
}
}
// listfeatures lists all features contained within the given bounding rectangle.
func (s *routeguideserver) listfeatures(rect *pb.rectangle, stream pb.routeguide_listfeaturesserver) error {
for _, feature := range s.savedfeatures {
if inrange(feature.location, rect) {
if err := stream.send(feature); err != nil {
return err
}
}
}
return nil
}
func (c *routeguideclient) recordroute(ctx context.context, opts ...grpc.calloption) (routeguide_recordrouteclient, error) {
stream, err := c.cc.newstream(ctx, &_routeguide_servicedesc.streams[1], "/routeguide.routeguide/recordroute", opts...)
if err != nil {
return nil, err
}
x := &routeguiderecordrouteclient{stream}
return x, nil
}
// runrecordroute sends a sequence of points to server and expects to get a routesummary from server.
func runrecordroute(client pb.routeguideclient) {
// create a random number of random points
r := rand.new(rand.newsource(time.now().unixnano()))
pointcount := int(r.int31n(100)) + 2 // traverse at least two points
var points []*pb.point
for i := 0; i < pointcount; i++ {
points = append(points, randompoint(r))
}
log.printf("traversing %d points.", len(points))
ctx, cancel := context.withtimeout(context.background(), 10*time.second)
defer cancel()
stream, err := client.recordroute(ctx)
if err != nil {
log.fatalf("%v.recordroute(_) = _, %v", client, err)
}
for _, point := range points {
if err := stream.send(point); err != nil {
log.fatalf("%v.send(%v) = %v", stream, point, err)
}
}
reply, err := stream.closeandrecv()
if err != nil {
log.fatalf("%v.closeandrecv() got error %v, want %v", stream, err, nil)
}
log.printf("route summary: %v", reply)
}
// recordroute records a route composited of a sequence of points.
//
// it gets a stream of points, and responds with statistics about the "trip":
// number of points, number of known features visited, total distance traveled, and
// total time spent.
func (s *routeguideserver) recordroute(stream pb.routeguide_recordrouteserver) error {
var pointcount, featurecount, distance int32
var lastpoint *pb.point
starttime := time.now()
for {
point, err := stream.recv()
if err == io.eof {
endtime := time.now()
return stream.sendandclose(&pb.routesummary{
pointcount: pointcount,
featurecount: featurecount,
distance: distance,
elapsedtime: int32(endtime.sub(starttime).seconds()),
})
}
if err != nil {
return err
}
pointcount++
for _, feature := range s.savedfeatures {
if proto.equal(feature.location, point) {
featurecount++
}
}
if lastpoint != nil {
distance += calcdistance(lastpoint, point)
}
lastpoint = point
}
}
func (c *routeguideclient) routechat(ctx context.context, opts ...grpc.calloption) (routeguide_routechatclient, error) {
stream, err := c.cc.newstream(ctx, &_routeguide_servicedesc.streams[2], "/routeguide.routeguide/routechat", opts...)
if err != nil {
return nil, err
}
x := &routeguideroutechatclient{stream}
return x, nil
}
// runroutechat receives a sequence of route notes, while sending notes for various locations.
func runroutechat(client pb.routeguideclient) {
notes := []*pb.routenote{
{location: &pb.point{latitude: 0, longitude: 1}, message: "first message"},
{location: &pb.point{latitude: 0, longitude: 2}, message: "second message"},
{location: &pb.point{latitude: 0, longitude: 3}, message: "third message"},
{location: &pb.point{latitude: 0, longitude: 1}, message: "fourth message"},
{location: &pb.point{latitude: 0, longitude: 2}, message: "fifth message"},
{location: &pb.point{latitude: 0, longitude: 3}, message: "sixth message"},
}
ctx, cancel := context.withtimeout(context.background(), 10*time.second)
defer cancel()
stream, err := client.routechat(ctx)
if err != nil {
log.fatalf("%v.routechat(_) = _, %v", client, err)
}
waitc := make(chan struct{})
go func() {
for {
in, err := stream.recv()
if err == io.eof {
// read done.
close(waitc)
return
}
if err != nil {
log.fatalf("failed to receive a note : %v", err)
}
log.printf("got message %s at point(%d, %d)", in.message, in.location.latitude, in.location.longitude)
}
}()
for _, note := range notes {
if err := stream.send(note); err != nil {
log.fatalf("failed to send a note: %v", err)
}
}
stream.closesend()
<-waitc
}
// routechat receives a stream of message/location pairs, and responds with a stream of all
// previous messages at each of those locations.
func (s *routeguideserver) routechat(stream pb.routeguide_routechatserver) error {
for {
in, err := stream.recv()
if err == io.eof {
return nil
}
if err != nil {
return err
}
key := serialize(in.location)
s.mu.lock()
s.routenotes[key] = append(s.routenotes[key], in)
// note: this copy prevents blocking other clients while serving this one.
// we don't need to do a deep copy, because elements in the slice are
// insert-only and never modified.
rn := make([]*pb.routenote, len(s.routenotes[key]))
copy(rn, s.routenotes[key])
s.mu.unlock()
for _, note := range rn {
if err := stream.send(note); err != nil {
return err
}
}
}
}
newclientstream：获取传输层 trasport 并组合封装到 clientstream 中返回，在这块会涉及负载均衡、超时控制、 encoding、 stream 的动作，与服务端基本一致的行为。
cs.sendmsg：发送 rpc 请求出去，但其并不承担等待响应的功能。
cs.recvmsg：阻塞等待接受到的 rpc 方法响应结果。
// invoke sends the rpc request on the wire and returns after response is
// received. this is typically called by generated code.
//
// all errors returned by invoke are compatible with the status package.
func (cc *clientconn) invoke(ctx context.context, method string, args, reply interface{}, opts ...calloption) error {
// allow interceptor to see all applicable call options, which means those
// configured as defaults from dial option as well as per-call options
opts = combine(cc.dopts.calloptions, opts)
if cc.dopts.unaryint != nil {
return cc.dopts.unaryint(ctx, method, args, reply, cc, invoke, opts...)
}
return invoke(ctx, method, args, reply, cc, opts...)
}
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)
}
// newstream creates a new stream for the client side. this is typically
// called by generated code. ctx is used for the lifetime of the stream.
//
// to ensure resources are not leaked due to the stream returned, one of the following
// actions must be performed:
//
// 1. call close on the clientconn.
// 2. cancel the context provided.
// 3. call recvmsg until a non-nil error is returned. a protobuf-generated
// client-streaming rpc, for instance, might use the helper function
// closeandrecv (note that closesend does not recv, therefore is not
// guaranteed to release all resources).
// 4. receive a non-nil, non-io.eof error from header or sendmsg.
//
// if none of the above happen, a goroutine and a context will be leaked, and grpc
// will not call the optionally-configured stats handler with a stats.end message.
func (cc *clientconn) newstream(ctx context.context, desc *streamdesc, method string, opts ...calloption) (clientstream, error) {
// allow interceptor to see all applicable call options, which means those
// configured as defaults from dial option as well as per-call options
opts = combine(cc.dopts.calloptions, opts)
if cc.dopts.streamint != nil {
return cc.dopts.streamint(ctx, desc, cc, method, newclientstream, opts...)
}
return newclientstream(ctx, desc, cc, method, opts...)
}
1.从实践到原理，带你参透grpc.
2..
如对本文有疑问， 点击进行留言回复！！
网友评论