RPC

基本概念

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 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更复杂。

创建三个maven项目

服务者

消费者

API

让服务者和消费者都依赖API

在消费者创建ConsumerApp类

使用代理对象

具体代码在ProxyUtils中

public class ConsumerApp {

public static void main(String[] args) {

//while死循环是为了测试调用提供者是否为随机

while (true) {

try {

Thread.sleep(2000);

// 获得代理对象

AddService addService = ProxyUtils.getProxy(AddService.class);

// 只要调用方法就会进入代理对象invoke方法

int result = addService.add(15, 684);

System.out.println(result);

} catch (InterruptedException e) {

e.printStackTrace();

}

}

}

}

API中创建Request，AddService，ProxyUtils，ZkUtils

创建Request(该类为传输对象，必须实现序列化)

public class Request implements Serializable{

/**

*

*/

private static final long serialVersionUID = 1L;

private String interfaceName;

private String methodName;

private Object[] args;

public String getInterfaceName() {

return interfaceName;

}

public void setInterfaceName(String interfaceName) {

this.interfaceName = interfaceName;

}

public String getMethodName() {

return methodName;

}

public void setMethodName(String methodName) {

this.methodName = methodName;

}

public Object[] getArgs() {

return args;

}

public void setArgs(Object[] args) {

this.args = args;

}

@Override

public String toString() {

return "Request [interfaceName=" + interfaceName + ", methodName=" + methodName + ", args="

+ Arrays.toString(args) + "]";

}

}

创建AddService

package com.chenlei.service;

public interface AddService {

public int add(Integer a, Integer b);

}

创建ProxyUtils(重点)

public class ProxyUtils {

private static Random RDM = new Random();

@SuppressWarnings("unchecked")

public static T getProxy(Class interfaces) {

T proxy = (T) Proxy.newProxyInstance(ProxyUtils.class.getClassLoader(), new Class>[] { interfaces },

new InvocationHandler() {

@Override

public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {

String methodName = method.getName();

if ("toString".equals(methodName)) {

return interfaces.getClass().getName() + "$Proxy";

}

if ("hashCode".equals(methodName)) {

return Object.class.hashCode();

}

if ("equals".equals(methodName)) {

return Object.class.equals(this);

}

// 消费者发送过去

Request request = new Request();

request.setInterfaceName(interfaces.getName());

request.setMethodName(methodName);

request.setArgs(args);

// 找到interfaces下的所有节点

List serverList = ZkUtils.discover(interfaces.getName());

String one = randomOne(serverList);// 拿到的结果为ip:port 如127.0.0.1:8888

String[] split = one.split(":");

String address = split[0];

Integer port = Integer.valueOf(split[1]);

Socket socket = null;

// 打开书出管道，发送请求

Object result = null;

OutputStream outputStream = null;

ObjectOutputStream objectOutputStream = null;

InputStream inputStream = null;

ObjectInputStream objectInputStream = null;

try {

socket = new Socket(address, port);

outputStream = socket.getOutputStream();

objectOutputStream = new ObjectOutputStream(outputStream);

objectOutputStream.writeObject(request);

inputStream = socket.getInputStream();

objectInputStream = new ObjectInputStream(inputStream);

result = objectInputStream.readObject();

System.out.println("本次调用的是======" + port);

} catch (Exception e) {

e.printStackTrace();

} finally {

closeResources(objectInputStream, inputStream, objectOutputStream, outputStream, socket);

}

return result;

}

});

return proxy;

}

/**

* 从节点中随机找出一个

*

* @param serverList

* @return

*/

private static String randomOne(List serverList) {

if (null == serverList || 0 == serverList.size()) {

return null;

}

int index = RDM.nextInt(serverList.size());

return serverList.get(index);

}

/**

* 关闭资源的方法

*/

public static void closeResources(Closeable... resources) {

for (Closeable resource : resources) {

if (null != resource) {

try {

resource.close();

} catch (IOException e) {

e.printStackTrace();

} finally {

resource = null;

}

}

}

}

}

创建ZkUtils(zookeeper注册和发现，另加缓存解决脏读)

在API项目中导入zkclient的依赖

public class ZkUtils {

private static final String ZK_URL = "自己的域名:2181";

private static ZkClient zkClient = null;

//创建zookeeper缓存

private static Map> cache = new HashMap>();

static {

zkClient = new ZkClient(ZK_URL, 10000, 10000);

}

/**

* 服务节点向zookeeper的注册

*

* @param serverName

* @param serverPort

*/

public static void register(String serverName, String serverPort) {

if (null == serverName || "".equals(serverName)) {

throw new RuntimeException("服务名不能为空");

}

if (null == serverPort || "".equals(serverPort)) {

throw new RuntimeException("服务ip和端口不能为空");

}

if (!zkClient.exists("/" + serverName)) {

zkClient.createPersistent("/" + serverName);

}

if (!zkClient.exists("/" + serverName + "/" + serverPort)) {

zkClient.createEphemeral("/" + serverName + "/" + serverPort);

}

System.out.println("注册一个服务节点为" + "/" + serverName + "/" + serverPort);

}

/**

* 向zookeeper发现服务节点

*

* @param serverName

* @return

*/

public static List discover(String serverName) {

if (null == serverName || "".equals(serverName)) {

throw new RuntimeException("服务名不能为空");

}

// 先从缓存里找

if (cache.containsKey(serverName)) {

System.out.println("在缓存中找到" + serverName + "节点");

}

// 如果该节点在zookeeper中不存在，直接返回空

if (!zkClient.exists("/" + serverName)) {

return null;

}

zkClient.subscribeChildChanges("/" + serverName, new IZkChildListener() {

@Override

public void handleChildChange(String parentPath, List currentChilds) throws Exception {

// 一旦进入此方法，证明有节点改变

cache.put(serverName, currentChilds);

System.out.println(serverName + "节点有变化-----" + "缓存完成更新");

}

});

return zkClient.getChildren("/" + serverName);

}

}

写提供者代码

创建AddServiceImpl

注意类名最好是AddService+Impl，并且类全路径也要对应com.chenlei.service.impl.AddServiceImpl,否则代码需要调整

package com.chenlei.service.impl;

import com.chenlei.service.AddService;

public class AddServiceImpl implements AddService {

@Override

public int add(Integer a, Integer b) {

return a + b;

}

}

创建ProviderApp(重点)

public class ProviderApp {

public static void main(String[] args) {

Integer port = 7777;

ServerSocket serverSocket = bind(port);

// 向zookeeper注册

ZkUtils.register(AddService.class.getName(), "127.0.0.1" + ":" + port);

// 监听+处理请求

listener(serverSocket);

}

/**

* 监听和处理请求

*

* @param serverSocket

*/

private static void listener(ServerSocket serverSocket) {

//此处死循环是为了让次提供者一直处于工作状态

while (true) {

Socket socket = null;

InputStream inputStream = null;

ObjectInputStream objectInputStream = null;

OutputStream outputStream = null;

ObjectOutputStream objectOutputStream = null;

try {

socket = serverSocket.accept();

inputStream = socket.getInputStream();

objectInputStream = new ObjectInputStream(inputStream);

Request request = (Request) objectInputStream.readObject();

Object answer = invoker(request);

outputStream = socket.getOutputStream();

objectOutputStream = new ObjectOutputStream(outputStream);

objectOutputStream.writeObject(answer);

} catch (Exception e) {

e.printStackTrace();

} finally {

ProxyUtils.closeResources(objectOutputStream, outputStream, objectInputStream, inputStream, socket);

}

}

}

/**

* 处理请求返回结果

*

* @param request

* @return

*/

private static Object invoker(Request request) {

// 获得从消费者传过来的信息

String interfaceName = request.getInterfaceName();

String methodName = request.getMethodName();

Object[] args = request.getArgs();

// 获得对应实现类全名

String className = getClassNameByInterfaceName(interfaceName);

Object answer = null;

try {

// 找到该类

Class> clazz = Class.forName(className);

// 创建一个对象

Object object = clazz.newInstance();

Class>[] argsType = new Class>[args.length];

if (null != args || 0 != args.length) {

for (int i = 0; i < args.length; i++) {

argsType[i] = args[i].getClass();

}

}

Method method = clazz.getMethod(methodName, argsType);

answer = method.invoke(object, args);

} catch (Exception e) {

e.printStackTrace();

}

return answer;

}

/**

* 通过请求者传来的类信息，获得对应实现类的所有信息,并返回实现类的全名

*

* @param interfaceName

* @return

*/

private static String getClassNameByInterfaceName(String interfaceName) {

// 传过来的接口名为com.chenlei.service.AddService

int index = interfaceName.lastIndexOf(".");

StringBuilder sb = new StringBuilder();

// com.chenlei.service

sb.append(interfaceName.subSequence(0, index));

// com.chenlei.service.impl.

sb.append(".impl.");

// com.chenlei.service.impl.AddService

sb.append(interfaceName.substring(index + 1)).append("Impl");

return sb.toString();

}

/**

* 绑定一个端口

*

* @param port

* @return

*/

private static ServerSocket bind(Integer port) {

ServerSocket serverSocket = null;

try {

serverSocket = new ServerSocket(port);

} catch (IOException e) {

e.printStackTrace();

}

return serverSocket;

}

/**

* 测试代码

*/

//public static void main(String[] args) {

// String interfaceName = "com.chenlei.service.AddService";

// int index = interfaceName.lastIndexOf(".");

// StringBuilder sb = new StringBuilder();

// // com.chenlei.service

// sb.append(interfaceName.subSequence(0, index));

// // com.chenlei.service.impl.

// sb.append(".impl.");

// // com.chenlei.service.impl.AddService

// sb.append(interfaceName.substring(index + 1)).append("Impl");

// System.out.println(sb.toString());

//}

}

更改提供者端口，分别启动三个提供者

再启动消费者，查看结果

其他错误和注意事项

写代码思路