1. 基本的RPC模型

主要介绍RPC是什么，基本的RPC代码，RPC与REST的区别，gRPC的使用

1.1 基本概念

• RPC（Remote Procedure Call）远程过程调用，简单的理解是一个节点请求另一个节点提供的服务
• 本地过程调用：如果需要将本地student对象的age+1，可以实现一个addAge()方法，将student对象传入，对年龄进行更新之后返回即可，本地方法调用的函数体通过函数指针来指定。
• 远程过程调用：上述操作的过程中，如果addAge()这个方法在服务端，执行函数的函数体在远程机器上，如何告诉机器需要调用这个方法呢？

1. 首先客户端需要告诉服务器，需要调用的函数，这里函数和进程ID存在一个映射，客户端远程调用时，需要查一下函数，找到对应的ID，然后执行函数的代码。
2. 客户端需要把本地参数传给远程函数，本地调用的过程中，直接压栈即可，但是在远程调用过程中不再同一个内存里，无法直接传递函数的参数，因此需要客户端把参数转换成字节流，传给服务端，然后服务端将字节流转换成自身能读取的格式，是一个序列化和反序列化的过程。 3.数据准备好了之后，如何进行传输？网络传输层需要把调用的ID和序列化后的参数传给服务端，然后把计算好的结果序列化传给客户端，因此TCP层即可完成上述过程，gRPC中采用的是HTTP2协议。 总结一下上述过程：

// Client端 // Student student = Call(ServerAddr, addAge, student)
1. 将这个调用映射为Call ID。
2. 将Call ID，student（params）序列化，以二进制形式打包
3. 把2中得到的数据包发送给ServerAddr，这需要使用网络传输层
4. 等待服务器返回结果
5. 如果服务器调用成功，那么就将结果反序列化，并赋给student，年龄更新
// Server端
1. 在本地维护一个Call ID到函数指针的映射call_id_map，可以用Map<String, Method> callIdMap
2. 等待服务端请求
3. 得到一个请求后，将其数据包反序列化，得到Call ID
4. 通过在callIdMap中查找，得到相应的函数指针
5. 将student（params）反序列化后，在本地调用addAge()函数，得到结果
6. 将student结果序列化后通过网络返回给Client

• 在微服务的设计中，一个服务A如果访问另一个Module下的服务B，可以采用HTTP REST传输数据，并在两个服务之间进行序列化和反序列化操作，服务B把执行结果返回过来。
• 由于HTTP在应用层中完成，整个通信的代价较高，远程过程调用中直接基于TCP进行远程调用，数据传输在传输层TCP层完成，更适合对效率要求比较高的场景，RPC主要依赖于客户端和服务端之间建立Socket链接进行，底层实现比REST更复杂。

1.2 rpc demo

客户端：

publicclassRPCClient<T> {
publicstatic <T> TgetRemoteProxyObj(finalClass<?> serviceInterface, finalInetSocketAddressaddr) {
// 1.将本地的接口调用转换成JDK的动态代理，在动态代理中实现接口的远程调用return (T) Proxy.newProxyInstance(serviceInterface.getClassLoader(), newClass<?>[]{serviceInterface},
newInvocationHandler() {
@OverridepublicObjectinvoke(Objectproxy, Methodmethod, Object[] args) throwsThrowable {
Socketsocket = null;
ObjectOutputStreamoutput = null;
ObjectInputStreaminput = null;
try{
// 2.创建Socket客户端，根据指定地址连接远程服务提供者socket = newSocket();
socket.connect(addr);
// 3.将远程服务调用所需的接口类、方法名、参数列表等编码后发送给服务提供者output = newObjectOutputStream(socket.getOutputStream());
output.writeUTF(serviceInterface.getName());
output.writeUTF(method.getName());
output.writeObject(method.getParameterTypes());
output.writeObject(args);
// 4.同步阻塞等待服务器返回应答，获取应答后返回input = newObjectInputStream(socket.getInputStream());
returninput.readObject();
}finally {
if (socket != null){
socket.close();
}
if (output != null){
output.close();
}
if (input != null){
input.close();
}
}
}
});
}
}

服务端：

publicclassServiceCenterimplementsServer {
privatestaticExecutorServiceexecutor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
privatestaticfinalHashMap<String, Class> serviceRegistry = newHashMap<String, Class>();
privatestaticbooleanisRunning = false;
privatestaticintport;
publicServiceCenter(intport){
ServiceCenter.port = port;
}
@Overridepublicvoidstart() throwsIOException {
ServerSocketserver = newServerSocket();
server.bind(newInetSocketAddress(port));
System.out.println("Server Start .....");
try{
while(true){
executor.execute(newServiceTask(server.accept()));
}
}finally {
server.close();
}
}
@Overridepublicvoidregister(ClassserviceInterface, Classimpl) {
serviceRegistry.put(serviceInterface.getName(), impl);
}
@OverridepublicbooleanisRunning() {
returnisRunning;
}
@OverridepublicintgetPort() {
returnport;
}
@Overridepublicvoidstop() {
isRunning = false;
executor.shutdown();
}
privatestaticclassServiceTaskimplementsRunnable {
Socketclient = null;
publicServiceTask(Socketclient) {
this.client = client;
}
@Overridepublicvoidrun() {
ObjectInputStreaminput = null;
ObjectOutputStreamoutput = null;
try{
input = newObjectInputStream(client.getInputStream());
StringserviceName = input.readUTF();
StringmethodName = input.readUTF();
Class<?>[] parameterTypes = (Class<?>[]) input.readObject();
Object[] arguments = (Object[]) input.readObject();
ClassserviceClass = serviceRegistry.get(serviceName);
if(serviceClass == null){
thrownewClassNotFoundException(serviceName + "not found!");
}
Methodmethod = serviceClass.getMethod(methodName, parameterTypes);
Objectresult = method.invoke(serviceClass.newInstance(), arguments);
output = newObjectOutputStream(client.getOutputStream());
output.writeObject(result);
}catch (Exceptione){
e.printStackTrace();
}finally {
if(output!=null){
try{
output.close();
}catch (IOExceptione){
e.printStackTrace();
}
}
if (input != null) {
try {
input.close();
} catch (IOExceptione) {
e.printStackTrace();
}
}
if (client != null) {
try {
client.close();
} catch (IOExceptione) {
e.printStackTrace();
}
}
}
}
}
}

publicclassServiceProducerImplimplementsServiceProducer{
@OverridepublicStringsendData(Stringdata) {
return"I am service producer!!!, the data is "+ data;
}
}

publicclassRPCTest {
publicstaticvoidmain(String[] args) throwsIOException {
newThread(newRunnable() {
@Overridepublicvoidrun() {
try {
ServerserviceServer = newServiceCenter(8088);
serviceServer.register(ServiceProducer.class, ServiceProducerImpl.class);
serviceServer.start();
} catch (IOExceptione) {
e.printStackTrace();
}
}
}).start();
ServiceProducerservice = RPCClient.getRemoteProxyObj(ServiceProducer.class, newInetSocketAddress("localhost", 8088));
System.out.println(service.sendData("test"));
}
}

1.3 完整源码

RPCdemo

1.4 分析

这里客户端只需要知道Server端的接口ServiceProducer即可，服务端在执行的时候，会根据具体实例调用实际的方法ServiceProducerImpl，符合面向对象过程中父类引用指向子类对象。

2. gRPC的使用

2.1. gRPC与REST

• REST通常以业务为导向，将业务对象上执行的操作映射到HTTP动词，格式非常简单，可以使用浏览器进行扩展和传输，通过JSON数据完成客户端和服务端之间的消息通信，直接支持请求/响应方式的通信。不需要中间的代理，简化了系统的架构，不同系统之间只需要对JSON进行解析和序列化即可完成数据的传递。
• 但是REST也存在一些弊端，比如只支持请求/响应这种单一的通信方式，对象和字符串之间的序列化操作也会影响消息传递速度，客户端需要通过服务发现的方式，知道服务实例的位置，在单个请求获取多个资源时存在着挑战，而且有时候很难将所有的动作都映射到HTTP动词。
• 正是因为REST面临一些问题，因此可以采用gRPC作为一种替代方案，gRPC 是一种基于二进制流的消息协议，可以采用基于Protocol Buffer的IDL定义grpc API,这是Google公司用于序列化结构化数据提供的一套语言中立的序列化机制，客户端和服务端使用HTTP/2以Protocol Buffer格式交换二进制消息。
• gRPC的优势是，设计复杂更新操作的API非常简单，具有高效紧凑的进程通信机制，在交换大量消息时效率高，远程过程调用和消息传递时可以采用双向的流式消息方式，同时客户端和服务端支持多种语言编写，互操作性强；不过gRPC的缺点是不方便与JavaScript集成，某些防火墙不支持该协议。
• 注册中心：当项目中有很多服务时，可以把所有的服务在启动的时候注册到一个注册中心里面，用于维护服务和服务器之间的列表，当注册中心接收到客户端请求时，去找到该服务是否远程可以调用，如果可以调用需要提供服务地址返回给客户端，客户端根据返回的地址和端口，去调用远程服务端的方法，执行完成之后将结果返回给客户端。这样在服务端加新功能的时候，客户端不需要直接感知服务端的方法，服务端将更新之后的结果在注册中心注册即可，而且当修改了服务端某些方法的时候，或者服务降级服务多机部署想实现负载均衡的时候，我们只需要更新注册中心的服务群即可。

2.2. gRPC与Spring Boot

这里使用SpringBoot+gRPC的形式实现RPC调用过程 项目结构分为三部分：client、grpc、server

2.2.2 grpc

pom.xml中引入依赖：

<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-all</artifactId>
<version>1.12.0</version>
</dependency>

引入bulid

<build>
<extensions>
<extension>
<groupId>kr.motd.maven</groupId>
<artifactId>os-maven-plugin</artifactId>
<version>1.4.1.Final</version>
</extension>
</extensions>
<plugins>
<plugin>
<groupId>org.xolstice.maven.plugins</groupId>
<artifactId>protobuf-maven-plugin</artifactId>
<version>0.5.0</version>
<configuration>
<pluginId>grpc-java</pluginId>
<protocArtifact>com.google.protobuf:protoc:3.0.2:exe:${os.detected.classifier}</protocArtifact>
<pluginArtifact>io.grpc:protoc-gen-grpc-java:1.2.0:exe:${os.detected.classifier}</pluginArtifact>
</configuration>
<executions>
<execution>
<goals>
<goal>compile</goal>
<goal>compile-custom</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>

创建.proto文件

syntax="proto3"; // 语法版本// stub选项optionjava_package="com.shgx.grpc.api";
optionjava_outer_classname="RPCDateServiceApi";
optionjava_multiple_files=true;
// 定义包名packagecom.shgx.grpc.api;
// 服务接口定义，服务端和客户端都要遵守该接口进行通信serviceRPCDateService {
rpcgetDate (RPCDateRequest) returns (RPCDateResponse) {}
}
// 定义消息（请求）messageRPCDateRequest {
stringuserName=1;
}
// 定义消息（响应）messageRPCDateResponse {
stringserverDate=1;
}

mvn complie

生成代码：

2.2.3 client

根据gRPC中的项目配置在client和server两个Module的pom.xml添加依赖

 <dependency>
<groupId>com.shgx</groupId>
<artifactId>grpc</artifactId>
<version>0.0.1-SNAPSHOT</version>
<scope>compile</scope>
</dependency>

编写GRPCClient

publicclassGRPCClient {
privatestaticfinalStringhost = "localhost";
privatestaticfinalintserverPort = 9999;
publicstaticvoidmain( String[] args ) throwsException {
ManagedChannelmanagedChannel = ManagedChannelBuilder.forAddress( host, serverPort ).usePlaintext().build();
try {
RPCDateServiceGrpc.RPCDateServiceBlockingStubrpcDateService = RPCDateServiceGrpc.newBlockingStub( managedChannel );
RPCDateRequestrpcDateRequest = RPCDateRequest
.newBuilder()
.setUserName("shgx")
.build();
RPCDateResponserpcDateResponse = rpcDateService.getDate( rpcDateRequest );
System.out.println( rpcDateResponse.getServerDate() );
} finally {
managedChannel.shutdown();
}
}
}

2.2.4 server

按照2.2.3 client的方式添加依赖 创建RPCDateServiceImpl

publicclassRPCDateServiceImplextendsRPCDateServiceGrpc.RPCDateServiceImplBase{
@OverridepublicvoidgetDate(RPCDateRequestrequest, StreamObserver<RPCDateResponse> responseObserver) {
RPCDateResponserpcDateResponse = null;
Datenow=newDate();
SimpleDateFormatsimpleDateFormat = newSimpleDateFormat("今天是"+"yyyy年MM月dd日 E kk点mm分");
StringnowTime = simpleDateFormat.format( now );
try {
rpcDateResponse = RPCDateResponse
.newBuilder()
.setServerDate( "Welcome " + request.getUserName() + ", " + nowTime )
.build();
} catch (Exceptione) {
responseObserver.onError(e);
} finally {
responseObserver.onNext( rpcDateResponse );
}
responseObserver.onCompleted();
}
}

创建GRPCServer

publicclassGRPCServer {
privatestaticfinalintport = 9999;
publicstaticvoidmain( String[] args ) throwsException {
Serverserver = ServerBuilder.
forPort(port)
.addService( newRPCDateServiceImpl() )
.build().start();
System.out.println( "grpc服务端启动成功, 端口=" + port );
server.awaitTermination();
}
}

2.3. 完整源码

源码参考

3. 参考文章

• gRPC - Spring Boot 示例
• RPC框架解释
• 什么是RPC