gRPC是谷歌设计的一个开源RPC(Remote Process
Call)框架,其基于谷歌开发的Protocol
Buffer (也支持其他数据结构如JSON、XML等),提供了一种分布式系统内部各个微服务之间互相调用的方法,具有语言无关、平台无关、高效(HTTP/2)、安全(TLS)、可扩展性强的特点,已被广泛应用于诸多公司如:NetFlex、Square、Cisco等。
RPC or HTTP?
正式学习之前讨论一个有意思的话题,即RPC技术和HTTP协议在分布式系统中,有何区别呢?
目的及区别
一个自然的问题就是,RPC和HTTP都可以实现C/S之间的沟通,比如现行的微服务架构中,提倡RESTful风格,服务与服务之间都是通过暴露HTTP
endpoint并通过HTTP协议、JSON数据格式进行通信的。而RPC也广泛应用于分布式系统内部各个服务之间的互相调用,比如Java
RMI技术以及今天学习的gRPC框架。
那区别和要解决的问题是什么呢?
总的来说,个人理解区别如下:
RPC多用于分布式系统中,HTTP多用于B/S架构。
RPC关注点是网络通信的本地透明化,HTTP关注点是WWW上资源的访问
下面具体讨论一下二者出现要解决的问题和区别。从出现时间上来讲,RPC出现的时间是要比HTTP早的 。
根据wiki描述,1960年代就出现了分布式计算中的Request-Response
协议,1970年代出现了RPC的模型,如ARPANET(早期互联网)文档中就有采用,1980年代有了一些实用的实现。而Remote
Process Call(RPC) 一词是由Bruce Jay Nelson于1981年提出的。
a remote procedure call (RPC ) is
when a computer program causes a procedure (subroutine) to execute in a
different address space (commonly on another computer on a shared
network), which is written as if it were a normal (local) procedure
call, without the programmer explicitly writing the details for the
remote interaction.
而HTTP——Hypertext Transfer
Protocol 超文本传输协议则于1989年由Tim Berners-Lee
提出,第一个版本HTTP/1.0于1996年提出,现在已经到了HTTP/3.0版本(2022年),是现代互联网数据通信的基石。
The Hypertext Transfer Protocol
(HTTP ) is an application layer protocol in the Internet
protocol suite model for distributed, collaborative, hypermedia
information systems.
所以这两个协议设计之初的场景就有所区别,RPC技术(我理解RPC是技术而不是协议)是为了提供分布式计算场景下、不同实体上进程的通信所设计的技术,且要实现本地透明化调用的效果 ,像调用本地方法一样调用另一个机器上的方法,不对上层业务逻辑代码产生影响。
而HTTP协议是为了让客户端能能够访问WWW上的资源(文本、图像、视频)而设计的协议,并设计了大量的状态码来标识状态。因此从这个角度,HTTP协议更多用于B/S架构,而RPC更多用于C/S 。
RESTful为什么不用RPC呢?
当然,也不是说HTTP不能用于C/S。
我们回到开始我提到的例子,RESTful风格就提倡使用HTTP,那为什么不用RPC呢?这里就需要稍微了解下RESTful风格(于2000年由Roy
Fielding博士论文中提出)了。
RESTful——Resource Representational State
Transfer(表示层状态转移) 之所以RESTful风格选择HTTP的原因在于,RESTful的关注点在Representational ,即资源的表示 ,提倡将服务的资源以可读的方式表示出来,如JSON、XML等,并通过HTTP提供的方法GET、POST、PUT、DELETE执行状态,使得服务端的服务发生状态变化(State
Transfer )。比如Spring
Boot应用中,大家可能用过HATEOAS组件,实现向客户端返回相关资源链接的效果。比如我们访问api.github.com,从相应的json中就可以得到所有的资源及其对应的链接。
单纯使用HTTP替换掉RPC技术是可行的,但是意义不大。 RESTful中定义的动作(GET、POST、PUT、DELETE),HTTP的一些状态码、特别是传输的格式(JSON、XML)没有太大的意义,个人理解原因如下:
服务之间的调用不需要动作 的概念,只是简单的调用
HTTP大量的状态码对于分布式系统中需要考虑的三态(超时、成功、失败)来说是冗余的
进程之间传输的数据不需要可读 这个属性
因此,Leonard
Richardson也提出了REST成熟度模型,上述提到的、单纯使用HTTP替换掉RPC的方式就属于成熟度最低的Level0,有兴趣可以进一步阅读:Richardson
Maturity Model (martinfowler.com)
Protocol Buffer基本概念
Protocol
Buffer有两个版本v2和v3,前者是后者子集。详细概念阐述、代码例子可以参考:Core concepts,
architecture and lifecycle | gRPC 、Basics tutorial |
Java | gRPC
Protocol
Buffer在谷歌内部有广泛的应用,包括不限于服务器内部通信、存档数据的存储等。
正如名字所言,protobuf中的核心就是protocol ,即协议 。个人理解,协议即是定义实体之间交互的方式方法流程,用于实现某个特定目的,就像函数一样 。因此我们也可以将这个过程抽象一下,得到协议方法 (services
)和所用到的特定消息 (message
),而这两个元素也正是使用protobuf时我们需要定义的内容,通常写在文件.proto
文件中。
message
一个简单的消息 定义如下:
1 2 3 4 5 6 7 8 9 syntax = "proto3" message SearchRequest { string query = 1 ; int32 page_number = 2 ; int32 results_per_page = 3 ; }
如上我们定义了一个查询请求的消息结构 ,syntax
关键字的值代表我们所使用的protobuf的版本,下面的message
关键字开始描述了一个名为SearchRequest
的消息的结构,包含三个属性,每个属性由类型和数值组成。这样就完成了一条消息格式的编写了。
这里需要注意,上面例子中的数值并非是默认值 的含义,而是类似序号的含义,他们唯一标识了消息中的字段,官方也指出了序号的规则:
对于一个message,每个字段的序号必须是独一无二的
序号19000
~19999
属于protobuf的保留序号
我们不能使用保留序号,且使用序号的范围是1
~536870911
通常情况下1-15的序号就够我们用了,一则没那么多变量需要定义,二则15-2047之间就需要两个字节来记录了,会产生更大的数据包。
services
在定义好消息之后,我们就可以定义使用消息进行交互的协议了。假设我们定义一个协议,发送方发送一个搜索请求给接收方,接收方回复一个搜索结果,那么我们可以将该过程定义出来:
1 2 3 4 5 syntax = "proto3" service SearchService { rpc Search(SearchRequest) returns (SearchResponse) ; }
工作流程
当我们定义好了消息和服务方法后,我们就可以使用官方提供的编译器进行编译了:Downloads | Protocol Buffers
Documentation (protobuf.dev)
该编译器支持输出Python、Java、C++等语言的代码,我们可以利用这些代码提供的API实现RPC调用。向pom依赖中添加如下依赖和插件,具体可见官方的README :
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 <dependencies > <dependency > <groupId > io.grpc</groupId > <artifactId > grpc-netty-shaded</artifactId > <version > 1.56.0</version > <scope > runtime</scope > </dependency > <dependency > <groupId > io.grpc</groupId > <artifactId > grpc-protobuf</artifactId > <version > 1.56.0</version > </dependency > <dependency > <groupId > io.grpc</groupId > <artifactId > grpc-stub</artifactId > <version > 1.56.0</version > </dependency > <dependency > <groupId > org.apache.tomcat</groupId > <artifactId > annotations-api</artifactId > <version > 6.0.53</version > <scope > provided</scope > </dependency > </dependencies > <build > <extensions > <extension > <groupId > kr.motd.maven</groupId > <artifactId > os-maven-plugin</artifactId > <version > 1.7.1</version > </extension > </extensions > <plugins > <plugin > <groupId > org.xolstice.maven.plugins</groupId > <artifactId > protobuf-maven-plugin</artifactId > <version > 0.6.1</version > <configuration > <protocArtifact > com.google.protobuf:protoc:3.22.3:exe:${os.detected.classifier}</protocArtifact > <pluginId > grpc-java</pluginId > <pluginArtifact > io.grpc:protoc-gen-grpc-java:1.56.0:exe:${os.detected.classifier}</pluginArtifact > </configuration > <executions > <execution > <goals > <goal > compile</goal > <goal > compile-custom</goal > </goals > </execution > </executions > </plugin > </plugins > </build >
如果是idea,可以在右侧的mvn中看到protobuf的插件:
这样在maven进行compile时就会自动生成代码了。
一个Demo
下面写一个简单的Demo,首先定义消息和协议方法,实现查看我一个小板子的一些状态信息。定义枚举、消息、协议如下:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 syntax = "proto3" ; package status;option java_package = "TinyServer" ;option java_outer_classname = "StatusServiceProtos" ;option java_multiple_files = true ;enum StatusOptions { CPU_USAGE = 0 ; MEM_USAGE = 1 ; KERNEL = 2 ; TIME = 3 ; } message StatusRequest { optional int32 requestOpt = 1 ; } message StatusResponse { optional string statusReport = 1 ; } service ServerStatus { rpc QueryIndex(StatusRequest) returns (StatusResponse) {} }
使用maven
compile对项目进行编译,在target目录下我们可以得到这protobuf为我们生成的代码:
在这个生成的xxxGrpc
的类中包含了我们服务端和客户端代码需要依赖的类:
其中xxxImplBase
是我们服务端,编写服务时需要继承并覆写方法的类;而xxxStub
则是客户端与服务端沟通使用的类,又分为三种:
xxxStub:异步IO模式
xxxBlockingStub:即同步的,等待服务器响应期间保持阻塞状态
xxxFutureStub:既可以当异步也可以当同步,Future机制
服务端
下面我们编写服务端代码:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 import io.grpc.Grpc;import io.grpc.InsecureServerCredentials;import io.grpc.Server;import io.grpc.stub.StreamObserver;import java.io.IOException;import java.text.SimpleDateFormat;import java.util.Date;import java.util.concurrent.TimeUnit;import java.util.logging.Logger;import java.lang.management.ManagementFactory;import com.sun.management.OperatingSystemMXBean;import TinyServer.ServerStatusGrpc;import TinyServer.StatusOptions;import TinyServer.StatusRequest;import TinyServer.StatusResponse;public class TinyStatusServer { private static final Logger logger = Logger.getLogger(TinyStatusServer.class.getName()); private final String port; private final Server server; public TinyStatusServer (String port) { this .port = port; this .server = Grpc.newServerBuilderForPort(Integer.parseInt(port), InsecureServerCredentials.create()) .addService(new StatusServiceImpl ()) .build(); } public void startServer () { try { server.start(); logger.info("状态服务启动成功!监听端口:" + port); Runtime.getRuntime().addShutdownHook(new Thread (() -> { System.err.println("*** JVM已停止" ); try { TinyStatusServer.this .stopServer(); } catch (InterruptedException e) { throw new RuntimeException (e); } System.err.println("*** 服务关闭" ); })); server.awaitTermination(); } catch (IOException e) { throw new RuntimeException ("端口已被占用!" ); } catch (InterruptedException e) { throw new RuntimeException (e); } } public void stopServer () throws InterruptedException { if (server != null ) { server.shutdown().awaitTermination(5 , TimeUnit.SECONDS); } } private static class StatusServiceImpl extends ServerStatusGrpc .ServerStatusImplBase { private StatusResponse getResult (StatusRequest request) { OperatingSystemMXBean operatingSystemMXBean = (OperatingSystemMXBean) ManagementFactory.getOperatingSystemMXBean(); String result; switch (request.getRequestOpt()) { case StatusOptions.CPU_USAGE_VALUE: result = String.valueOf(operatingSystemMXBean.getSystemLoadAverage()); break ; case StatusOptions.MEM_USAGE_VALUE: result = (operatingSystemMXBean.getTotalPhysicalMemorySize() - operatingSystemMXBean.getFreePhysicalMemorySize()) / (1024 * 1024 ) + "MB" ; break ; case StatusOptions.KERNEL_VALUE: result = operatingSystemMXBean.getArch(); break ; case StatusOptions.TIME_VALUE: SimpleDateFormat format = new SimpleDateFormat ("yyyy-MM-dd HH:mm:ss" ); ; result = format.format(new Date ()); break ; default : result = "未知的操作数!" ; } return StatusResponse.newBuilder() .setStatusReport(result) .build(); } @Override public void queryIndex (StatusRequest request, StreamObserver<StatusResponse> responseObserver) { responseObserver.onNext(getResult(request)); responseObserver.onCompleted(); } } public static void main (String[] args) { TinyStatusServer statusServer = new TinyStatusServer (args[0 ]); statusServer.startServer(); } }
客户端
对应的客户端代码如下,我们使用的阻塞的stub:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 import TinyServer.ServerStatusGrpc;import TinyServer.ServerStatusGrpc.ServerStatusBlockingStub;import TinyServer.StatusRequest;import TinyServer.StatusResponse;import io.grpc.Grpc;import io.grpc.InsecureChannelCredentials;import io.grpc.ManagedChannel;import java.util.logging.Logger;public class TinyStatusClient { private static final Logger logger = Logger.getLogger(TinyStatusServer.class.getName()); private final ServerStatusBlockingStub blockingStub; public TinyStatusClient (ManagedChannel channel) { this .blockingStub = ServerStatusGrpc.newBlockingStub(channel); } public static void main (String[] args) { String target = args[0 ]; String requestOpt = args[1 ]; ManagedChannel channel = Grpc.newChannelBuilder(target, InsecureChannelCredentials.create()).build(); TinyStatusClient client = new TinyStatusClient (channel); StatusRequest request = StatusRequest.newBuilder() .setRequestOpt(Integer.parseInt(requestOpt)) .build(); StatusResponse response = client.blockingStub.queryIndex(request); logger.info(response.toString()); } }
效果如下:
参考学习