RPC框架的简单实现
Posted IT哈哈
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了RPC框架的简单实现相关的知识，希望对你有一定的参考价值。
1.什么是RPC
① RPC简介
RPC(Remote Procedure Call Protocol)-远程过程调用协议。通过网络从远程计算机程序上请求服务，而不需要了解底层网络技术的协议。它假定某种传输协议的存在，如TCP，UDP，为通信程序之间携带信息数据。在OSI网络通信模型中，RPC跨越了传输层和应用层，因分布式，微服务等而兴起
其实简单点来理解，就是比如有一个应用1，通过网络调用了应用2的服务，而不用关心应用2的协议。这就是最简单的示例。
② 为什么是远程过程调用协议？
这里的过程指的是业务处理，计算任务，更直白一点，就是程序，就像是在调用本地方法一样
与本地调用的区别可以用异地恋这个场景来理解，本地调用即是，大家同居了，都住在一起，而远程调用就像异地恋，想要见面就要不男坐车过来要不女坐车过来，而且也说异地恋一般没有好结果。远程调用中间需要网络，所以我们就可以明白
因为是通过网络通信的，所以
响应会慢上几个数量级
不像本地如此可靠
③ RPC的C/S模式
RPC采用client-server结构，通过request-response消息模式实现，如果服务器的程序进行了更新，那客户端方面也需要同步更新才能正常使用
④ RPC的三个过程
可我们说的是中文，他说英语，大家互相都听不懂大家说话，这就没法沟通了，所以序列化其实就是一个翻译作用
把这三个过程都做好之后RPC才能正常工作
⑤ 我们为什么要使用RPC
微服务，分布式系统架构的兴起，为了实现服务可重用或者说系统间的交互调用
⑥ RPC与其他协议的区别
RPC和RMI的区别，RMI远程方法调用是OOP领域中RPC的一种具体实现。也就是RPC是父，RMI是子，且RMI仅能存在Java中调用
WebService，restful接口调用其实都是RPC，只是它们的消息组织方式和消息协议不同而已
⑦ 与MQ的使用场景对比：
架构上它们的差异是：MQ有一个中间节点Queue作为消息存储点
RPC是同步调用，对于要等待返回结果的实例，RPC处理的比较自然，由于等待结果时Consumer会有线程消耗，
如果以异步的方式使用，可以避免Consumer的线程消耗，
但它不能做到像MQ一样暂存请求，压力会堆积在服务Provider那
而MQ会把请求的压力暂缓，让处理者根据自己的节奏处理，但是由于消息是暂存在了它的队列中，
所以这个系统的可靠性会受到这个队列的影响，它是异步单向，发送消息设计为不需要等待消息处理的完成，
所以如果有需要设计为同步返回的功能的话，MQ就会变得不好使用
所以如果是非常着急需要等待返回结果的，又或者说是希望使用方便简单，RPC就比较好用，它的使用基于接口，类似于本地调用，异步编程会相对复杂，比如gRPC。而不希望发送端受限于处理端的速度时，就使用MQ。随着业务增长，也会出现处理端的处理速度不够，从而进行同步调用到异步消息的改造
2.RPC的流程及其协议
① RPC的流程
1.客户端处理过程中调用client stub（就像调用本地方法一样），传入参数
2.Client stub将参数编组为消息，然后通过系统调用向服务端发送消息
3.客户端本地操作系统将消息从客户端机器发送到服务端机器
4.服务端操作系统将接收到的数据包传递给client stub
5.server stub解组消息为参数
6.server stub再调用服务端的过程，过程执行结果以反方向的相同步骤响应给客户端
stub：分布式计算中的存根是一段代码，它转换在远程过程调用期间Client和server之间传递的参数
② 整个流程中需要注意处理的问题
1.client stub和server stub的开发问题
2.参数如何编组为消息，以及如何解组
3.消息如何发送
4.过程结果如何表示，异常情况如何处理
5.如何实现安全的访问控制
③ 核心概念术语
client和server，calls调用，replies响应，services，programs，procedures，version，Marshalling和unmarshalling编组和解组
关于services，programs，procedures，version
一个网络服务由一个或者多个远程程序集构成，而一个远程程序实现一个或多个远程过程。过程与过程参数，结果在程序协议说明书中定义说明，而为兼容程序协议变更，一个服务端可能支持多个版本的远程程序
④ RPC协议
RPC调用过程中需要将参数编组为消息进行发送，接收方需要解组消息为参数，过程处理结果同样需要经过编组解组。消息由哪些部分构成及消息的表示形式就构成了消息协议。RPC调用过程中采用的消息协议成为RPC协议
RPC是规定要做的事，RPC协议规定请求响应消息的格式，在TCP之上我们可以选用或自定义消息协议来完成我们的RPC交互，此时我们可以选用http或者https这种通用的标准协议，也可以根据自身的需要定义自己的消息协议（较多）
RPC框架
RPC框架封装好了参数编组解组，底层网络通信的RPC程序开发框架，让我们可以直接在其基础之上只需要专注于过程代码编写
Java领域的比较常见的RPC框架：webService，Apache的CXF,Axis2，Java自带的jax-ws，微服务常见的dubbo，spring cloud，Apache Thrift，ICE,google的GRPC等
① RPC框架的服务暴露
远程提供者需要以某种形式提供服务调用相关的信息，包括但不限于服务接口定义，数据结构，或者中间态的服务定义文件，例如Thrift的IDL文件，webService的WSDL文件，服务的调用者需要通过一定的途径或者远程服务调用的相关信息，其实简单点说，就是需要告诉别人怎么调用服务
② RPC框架的远程代理对象
代理处理技术：服务调用者用的服务实际是远程服务的本地代理，其实就是通过动态代理来实现
Java里至少提供了两种方式来提供动态代码生成，一种是jdk动态代理，另一种是字节码生成，动态代理相比字节码生成使用起来更方便，但动态代理方式在性能上比字节码要差，而字节码生成在代码可读性上要差很多，所以我们一般都是牺牲一些性能来获得代码可读性和可维护性的提高
③ RPC框架的通信
RPC框架通信和具体的协议是无关的，它可基于HTTP或TCP协议，webService就是基于http协议的RPC，具有更好的跨平台性，但性能却不如基于TCP协议的RPC
NIO其实不一定会比BIO性能要高，NIO只是为了支持高并发而已，特点就是非阻塞的，适合小数据量高并发的场景，大数据包情况下，BIO会更为合适
④ RPC框架的序列化
两个方面会直接影响RPC的性能，一是传输方式，二是序列化
二：RPC的流程和任务
1. RPC的流程
其实这个在上一篇的2 - ① 也已经提到过了，如果忘了，没关系，我再复制过来
stub：分布式计算中的存根是一段代码，它转换在远程过程调用期间Client和server之间传递的参数
1.客户端处理过程中调用client stub（就像调用本地方法一样），传入参数
2.Client stub将参数编组为消息，然后通过系统调用向服务端发送消息
3.客户端本地操作系统将消息从客户端机器发送到服务端机器
4.服务端操作系统将接收到的数据包传递给client stub
5.server stub解组消息为参数
6.server stub再调用服务端的过程，过程执行结果以反方向的相同步骤响应给客户端
2. 从使用者的角度开始分析
1.定义过程接口
2.服务端实现接口的整个过程
3.客户端使用生成的stub代理对象
三：RPC框架的设计及实现
1. 准备一个Student的实体类及基础接口
客户端生成过程接口的代理对象，通过设计一个客户端代理工厂，使用JDK动态代理即可生成接口的代理对象
① 定义一个StudentService接口
Student类有三个属性name（String），age（int），sex（String），节省篇幅就不贴代码了，提供getter，setter和toString方法即可
并且提供一个简单的实现，其实就是打印一个Student的信息出来而已
2.客户端的搭建
① 从测试类去了解所需
首先，客户端通过我们的本地代理，获得我们的StudentService的代理类，此时我们客户端本地是肯定不存在StudentService的实现的，此时寻址我们是直接给出来了
此时我们的关注点转到客户端是如何帮我们进行代理的
② 实现了InvocationHandler接口的RpcClientProxy
JDK提供了Proxy类来实现我们的动态代理，可以通过newProxyInstance(ClassLoader var0, Class<?>[] var1, InvocationHandler var2)方法来实例化一个代理对象，此时我们传入的参数clazz是规定必须为一个接口的，如果不是接口就不能使用JDK动态代理
而第三个参数RpcClientProxy.this则是newProxyInstance()方法虽然帮我们创建好了实例，但是创建实例完成后的具体动作必须由这个InvocationHandler来提供
InvocationHandler这个接口里面仅仅只有一个 Object invoke(Object var1, Method var2, Object[] var3) throws Throwable，这个方法的参数相信不难理解，第一个是代理对象，第二个是执行的方法，第三个是所需的参数集
回到我们刚刚的代码，在我执行System.out.println(service.getInfo())这条语句的时候，我们的逻辑就会跳到invoke()的实现中来，在invoke()方法的注释中也把过程很详细的说明了，首先我们需要调用远程服务了，进行一个参数的封装，之后就进行一个网络连接把这些参数发送给我们的服务端，此时我们需要用到RpcClient了
③ RpcClient
在start()方法中，我们的RpcRequest request是实现了Serializable接口的，所以此时封装好的数据会转换成一个二进制然后被flush()过去，此时我们消息已经发送了，需要等待服务端的响应，响应我们就需要通过我们的服务端ObjectOutputStream来接收一个输入流
④ 进行参数封装的RpcRequest
⑤ Rpc服务端响应结果包装RpcResponse
同时也是实现了JDK默认的序列化Serializable
3.服务端的搭建
① 服务端的模拟ServerTest
给到一个端口号，参数中带有一个包，功能是扫描某个包下的服务
② start()方法的实现
创建一个Map类型的集合services存放扫描到提供rpc服务的类，此时因为没有放在注册中心上所以就不存在寻址了。后面将会把它放入zookeeper的注册中心
getService()下，我们在ServerTest不是提供了一个包名吗，此时我们先去找到了它们所有的classes（请参考getClasses()方法），getClasses()中我们其实主要是先根据提供的包名往下找，要是目录都有问题的话就抛出异常，如果没问题，就开始遍历此目录下的所有文件，遍历出来的结果如果发现这个文件是class文件，就把其实例化，并且进行判断是否存在一个自定义注解@service，标注了这个注解的类就是RPC服务的实现类。如果存在这个注解，那就是我们需要找的rpc服务，就把它装到一个结果集classes中，如果目录下面仍然是目录，那就自己调用自己，直到看到class文件为止
当我们把所有的class都找到了，回到getService()方法下，就都集中放于一个classList中，然后把它们Map化，就是把接口的名称作为key，把实例作为value（services.put(cla.getAnnotation(Service.class).value().getName(), obj)）。
最后再回到start()，进行完服务扫描之后还会有一个RpcServerHandler来进行处理
/**
* RpcServer
* Rpc服务提供者
*/
public class RpcServer {
/**
* 启动指定的网络端口号服务，并监听端口上的请求数据。获得请求数据以后将请求信息委派给服务处理器，放入线程池中执行。
* @param port 监听端口
* @param clazz 服务类所在包名，多个用英文逗号隔开
*/
public void start(int port, String clazz) {
ServerSocket server = null;
try {
// 1. 创建服务端指定端口的socket连接
server = new ServerSocket(port);
// 2. 获取所有rpc服务类
Map<String, Object> services = getService(clazz);
// 3. 创建线程池
Executor executor = new ThreadPoolExecutor(5, 10, 10, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
while(true){
// 4. 获取客户端连接
Socket client = server.accept();
// 5. 放入线程池中执行
RpcServerHandler service = new RpcServerHandler(client, services);
executor.execute(service);
}
} catch (IOException e) {
e.printStackTrace();
}finally{
//关闭监听
if(server != null)
try {
server.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
/**
* 实例化所有rpc服务类，也可用于暴露服务信息到注册中心。
* @param clazz 服务类所在包名，多个用英文逗号隔开
* @return
*/
public Map<String,Object> getService(String clazz){
try {
Map<String, Object> services = new HashMap<String, Object>();
// 获取所有服务类
String[] clazzes = clazz.split(",");
List<Class<?>> classes = new ArrayList<Class<?>>();
for(String cl : clazzes){
List<Class<?>> classList = getClasses(cl);
classes.addAll(classList);
}
// 循环实例化
for(Class<?> cla:classes){
Object obj = cla.newInstance();
services.put(cla.getAnnotation(Service.class).value().getName(), obj);
}
return services;
} catch (Exception e) {
throw new RuntimeException(e);
}
}
/**
* 获取包下所有有@Sercive注解的类
* @param pckgname
* @return
* @throws ClassNotFoundException
*/
public static List<Class<?>> getClasses(String pckgname) throws ClassNotFoundException {
// 需要查找的结果
List<Class<?>> classes = new ArrayList<Class<?>>();
// 找到指定的包目录
File directory = null;
try {
ClassLoader cld = Thread.currentThread().getContextClassLoader();
if (cld == null)
throw new ClassNotFoundException("无法获取到ClassLoader");
String path = pckgname.replace('.', '/');
URL resource = cld.getResource(path);
if (resource == null)
throw new ClassNotFoundException("没有这样的资源：" + path);
directory = new File(resource.getFile());
} catch (NullPointerException x) {
throw new ClassNotFoundException(pckgname + " (" + directory + ") 不是一个有效的资源");
}
if (directory.exists()) {
// 获取包目录下的所有文件
String[] files = directory.list();
File[] fileList = directory.listFiles();
// 获取包目录下的所有文件
for (int i = 0; fileList != null && i < fileList.length; i++) {
File file = fileList[i];
//判断是否是Class文件
if (file.isFile() && file.getName().endsWith(".class")) {
Class<?> clazz = Class.forName(pckgname + '.' + files[i].substring(0, files[i].length() - 6));
if(clazz.getAnnotation(Service.class) != null){
classes.add(clazz);
}
}else if(file.isDirectory()){ //如果是目录，递归查找
List<Class<?>> result = getClasses(pckgname+"."+file.getName());
if(result != null && result.size() != 0){
classes.addAll(result);
}
}
}
} else{
throw new ClassNotFoundException(pckgname + "不是一个有效的包名");
}
return classes;
}
}
③ 进行处理的RpcServerHandler
和刚刚的RpcClient非常类似，都是序列化和反序列化的过程，主要是第三步中获得了实例和方法及其参数后，再调用invoke()方法然后把结果放入response的过程
/**
* RpcServerHandler
* 服务端请求处理，处理来自网络IO的服务请求，并响应结果给网络IO。
*/
public class RpcServerHandler implements Runnable {
// 客户端网络请求socket，可以从中获得网络请求信息
private Socket clientSocket;
// 服务端提供处理请求的类集合
private Map<String, Object> serviceMap;
/**
* @param client 客户端socket
* @param services 所有服务
*/
public RpcServerHandler(Socket client, Map<String, Object> services) {
this.clientSocket = client;
this.serviceMap = services;
}
/**
* 读取网络中客户端请求的信息，找到请求的方法，执行本地方法获得结果，写入网络IO输出中。
*
*/
public void run() {
ObjectInputStream oin = null;
ObjectOutputStream oout = null;
RpcResponse response = new RpcResponse();
try {
// 1. 获取流以待操作
oin = new ObjectInputStream(clientSocket.getInputStream());
oout = new ObjectOutputStream(clientSocket.getOutputStream());
// 2. 从网络IO输入流中请求数据，强转参数类型
Object param = oin.readObject();
RpcRequest request = null;
if(!(param instanceof RpcRequest)){
response.setError(new Exception("参数错误"));
oout.writeObject(response);
oout.flush();
return;
}else{
// 反序列化RpcRequest
request = (RpcRequest) param;
}
// 3. 查找并执行服务方法
Object service = serviceMap.get(request.getClassName());
Class<?> clazz= service.getClass();
Method method = clazz.getMethod(request.getMethodName(), request.getParamTypes());
Object result = method.invoke(service, request.getParams());
// 4. 返回RPC响应，序列化RpcResponse
response.setResult(result);
// 序列化结果
oout.writeObject(response);
oout.flush();
return;
} catch (Exception e) {
try { //异常处理
if(oout != null){
response.setError(e);
oout.writeObject(response);
oout.flush();
}
} catch (Exception e1) {
e1.printStackTrace();
}
return;
}finally{
try { // 回收资源，关闭流
if(oin != null) oin.close();
if(oout != null) oout.close();
if(clientSocket != null) clientSocket.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
4.运行结果
先开启ServerTest再开启ClientTest，简单快捷，注意别去右键跑main方法即可
四：优化客户端的举措
1.发现者的引入
设计客户端的时候，在ClientStubInvocationHandler中需要完成的两件事为编组消息和发送网络请求，而将请求的内容编组为消息这件事就交由客户端的stub代理，它除了消息协议和网络层的事务以外，可能还存在一个服务信息发现，此外消息协议可能也是会存在变化的，我们也需要去支持多种协议，这个其实是和框架对协议的支持广度有关的。比如dubbo相对于spring cloud而言对协议的支持就相对灵活一些
此时我们需要得知某服务用的是什么协议，所以我们需要引入一个服务发现者
2.协议层
我们想要做到支持多种协议，类该如何设计（面向接口，策略模式，组合）
此时我们的协议需要抽象出来，对于协议的内容需要进行编组和解组，比如我们上面提供的JSON和HTTP两种不同的实现，而此时客户端的存根里面就不仅仅只是需要服务发现者，还需要我们对于这个协议的支持
① 补充：如何从zookeeper中获取注册信息
public class RegistCenter {
ZkClient client = new ZkClient("localhost:2181");
private String centerRootPath = "/rpc";
public RegistCenter() {
client.setZkSerializer(new MyZkSerializer());
}
public void regist(ServiceResource serviceResource) {
String serviceName = serviceResource.getServiceName();
String uri = JsonMapper.toJsonString(serviceResource);
try {
uri = URLEncoder.encode(uri, "UTF-8");
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
String servicePath = centerRootPath + "/"+serviceName+"/service";
if(! client.exists(servicePath)) {
client.createPersistent(servicePath, true);
}
String uriPath = servicePath+"/"+uri;
client.createEphemeral(uriPath);
}
/**
* 加载配置中心中服务资源信息
* @param serviceName
* @return
*/
public List<ServiceResource> loadServiceResouces(String serviceName) {
String servicePath = centerRootPath + "/"+serviceName+"/service";
List<String> children = client.getChildren(servicePath);
List<ServiceResource> resources = new ArrayList<ServiceResource>();
for(String ch : children) {
try {
String deCh = URLDecoder.decode(ch, "UTF-8");
ServiceResource r = JsonMapper.fromJsonString(deCh, ServiceResource.class);
resources.add(r);
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
return resources;
}
private void sub(String serviceName, ChangeHandler handler) {
/*
String path = centerRootPath + "/"+serviceName+"/service";
client.subscribeChildChanges(path, new IZkChildListener() {
@Override
public void handleChildChange(String parentPath, List<String> currentChilds) throws Exception {
handler();
}
});
client.subscribeDataChanges(path, new IZkDataListener() {
@Override
public void handleDataDeleted(String dataPath) throws Exception {
handler();
}
@Override
public void handleDataChange(String dataPath, Object data) throws Exception {
handler();
}
});
*/
}
interface ChangeHandler {
/**
* 发生变化后给一个完整的属性对象
* @param resource
*/
void itemChange(ServiceResource resource);
}
}
② ClientStubProxyFactory
/**
* ClientStubProxyFactory
* 客户端存根代理工厂
*/
public class ClientStubProxyFactory {
private ServiceInfoDiscoverer sid;
private Map<String, MessageProtocol> supportMessageProtocols;
private NetClient netClient;
private Map<Class<?>, Object> objectCache = new HashMap<>();
/**
*
*
* @param <T>
* @param interf
* @return
*/
@SuppressWarnings("unchecked")
public <T> T getProxy(Class<T> interf) {
T obj = (T) this.objectCache.get(interf);
if (obj == null) {
obj = (T) Proxy.newProxyInstance(interf.getClassLoader(), new Class<?>[] { interf },
new ClientStubInvocationHandler(interf));
this.objectCache.put(interf, obj);
}
return obj;
}
public ServiceInfoDiscoverer getSid() {
return sid;
}
public void setSid(ServiceInfoDiscoverer sid) {
this.sid = sid;
}
public Map<String, MessageProtocol> getSupportMessageProtocols() {
return supportMessageProtocols;
}
public void setSupportMessageProtocols(Map<String, MessageProtocol> supportMessageProtocols) {
this.supportMessageProtocols = supportMessageProtocols;
}
public NetClient getNetClient() {
return netClient;
}
public void setNetClient(NetClient netClient) {
this.netClient = netClient;
}
/**
* ClientStubInvocationHandler
* 客户端存根代理调用实现
* @date 2019年4月12日 下午2:38:30
*/
private class ClientStubInvocationHandler implements InvocationHandler {
private Class<?> interf;
public ClientStubInvocationHandler(Class<?> interf) {
super();
this.interf = interf;
}
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
// 1、获得服务信息
String serviceName = this.interf.getName();
ServiceInfo sinfo = sid.getServiceInfo(serviceName);
if (sinfo == null) {
throw new Exception("远程服务不存在！");
}
// 2、构造request对象
Request req = new Request();
req.setServiceName(sinfo.getName());
req.setMethod(method.getName());
req.setPrameterTypes(method.getParameterTypes());
req.setParameters(args);
// 3、协议层编组
// 获得该方法对应的协议
MessageProtocol protocol = supportMessageProtocols.get(sinfo.getProtocol());
// 编组请求
byte[] data = protocol.marshallingRequest(req);
// 4、调用网络层发送请求
byte[] repData = netClient.sendRequest(data, sinfo);
// 5解组响应消息
Response rsp = protocol.unmarshallingResponse(repData);
// 6、结果处理
if (rsp.getException() != null) {
throw rsp.getException();
}
return rsp.getReturnValue();
}
}
}
ClientStub中有两个引用，一个是服务发现接口ServiceInfoDiscoverer，作用为根据服务名获得远程服务信息，提供一个ServiceInfo getServiceInfo(String name)方法，还有就是对于不同协议的支持supportMessageProtocols，MessageProtocol我们也是定义了一个接口，这个接口就需要比较详细了，编码成二级制，和解码成Request等，对于response也是同样这么个过程
此时又存在一些问题，单纯依靠编组和解组的方法是不够的，编组和解组的操作对象是请求，响应，但是它们的内容是不同的，此时我们又需要定义框架标准的请求响应类
request有具体的服务名，服务方法，消息头，参数类型和参数，同样的response也有状态（通过枚举），消息头，返回值及类型以及是否存在异常。
此时协议层扩展为4个方法
将消息协议独立为一层，客户端和服务端都需要使用
3. 网络层
所以我们需要的三个依赖就都出来了，一个是服务发现者，一个是协议支持，再然后就是我们网络层的NetClient
4. 总图
紫色代表客户端代理部分，浅绿色属于服务发现，浅蓝色属于协议部分
5.代码部分（可直接无视）
因为这些代码和主要的思路已经没有瓜葛了，只是一些功能代码，所以可以直接忽略了。如果实在是想自己跑一下，也可以问我要一个小样。
① 依旧是回到我们的ClientStubProxyFactory
可以和内容二的RpcClientProxy做一个对比，在原有的基础上加上了三个依赖ServiceInfoDiscoverer，supportMessageProtocols，netClient
在ClientStubProxyFactory中对Object做了一个缓存，如果已经存在这个缓存就直接返回，没有的话加入到缓存中然后new出来，只是一个小小的不同。
② invoke()方法的改变
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
// 1、获得服务信息
String serviceName = this.interf.getName();
ServiceInfo sinfo = sid.getServiceInfo(serviceName);
if (sinfo == null) {
throw new Exception("远程服务不存在！");
}
// 2、构造request对象
Request req = new Request();
req.setServiceName(sinfo.getName());
req.setMethod(method.getName());
req.setPrameterTypes(method.getParameterTypes());
req.setParameters(args);
// 3、协议层编组
// 获得该方法对应的协议
MessageProtocol protocol = supportMessageProtocols.get(sinfo.getProtocol());
// 编组请求
byte[] data = protocol.marshallingRequest(req);
// 4、调用网络层发送请求
byte[] repData = netClient.sendRequest(data, sinfo);
// 5、解组响应消息
Response rsp = protocol.unmarshallingResponse(repData);
// 6、结果处理
if (rsp.getException() != null) {
throw rsp.getException();
}
return rsp.getReturnValue();
}
首先是服务发现，在我们执行 ① 中提到的getProxy()方法时，此时代理的接口已经直接告诉我们了，所以我们就直接获得了接口信息interf，然后调用getName()方法获取接口的名称，通过接口名，调用服务发现者ServiceInfo提供的getServiceInfo()方法就能获取服务的具体信息，然后放入请求参数request里面，接下来给request的各个属性赋值
之后我们就开始寻找这个服务所对应的协议，获得协议之后可以获取协议支持对象，之后进行编组请求，转换成二进制，通过netClient发送过去，顺带连同服务端信息给出去。获取结果repData进行解组（二进制回到response），之后进行结果处理。
③ 服务发现者的实现
之前也提到了，服务发现者ServiceInfoDiscoverer是作为一个接口提供了getServiceInfo()方法的
有两种不同的实现，本地实现我们可以自己搞一个配置文件加载进来，把相关的服务信息弄进去得了
zookeeper的服务发现实现如下，类似于我们一开始在2 - ① 中补充的zookeeper的内容
public class ZookeeperServiceInfoDiscoverer implements ServiceInfoDiscoverer {
ZkClient client = new ZkClient("localhost:2181");
private String centerRootPath = "/rpc";
public ZookeeperServiceInfoDiscoverer() {
client.setZkSerializer(new MyZkSerializer());
}
public void regist(ServiceInfo serviceResource) {
String serviceName = serviceResource.getName();
String uri = JSON.toJSONString(serviceResource);
try {
uri = URLEncoder.encode(uri, "UTF-8");
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
String servicePath = centerRootPath + "/"+serviceName+"/service";
if(! client.exists(servicePath)) {
client.createPersistent(servicePath, true);
}
String uriPath = servicePath+"/"+uri;
client.createEphemeral(uriPath);
}
/**
* 加载配置中心中服务资源信息
* @param serviceName
* @return
*/
public List<ServiceInfo> loadServiceResouces(String serviceName) {
String servicePath = centerRootPath + "/"+serviceName+"/service";
List<String> children = client.getChildren(servicePath);
List<ServiceInfo> resources = new ArrayList<ServiceInfo>();
for(String ch : children) {
try {
String deCh = URLDecoder.decode(ch, "UTF-8");
ServiceInfo r = JSON.parseObject(deCh, ServiceInfo.class);
resources.add(r);
} catch (UnsupportedEncodingException e) {
e.printStackTrace();
}
}
return resources;
}
@Override
public ServiceInfo getServiceInfo(String name) {
List<ServiceInfo> list = loadServiceResouces(name);
ServiceInfo info = list.get(0);
list.forEach((e)->{
if(e != info) {
info.addAddress(e.getAddress().get(0));
}
});
return info;
}
}
④ 协议支持相关
这里只实现了JSON的，通过fastJSON来实现
public class JSONMessageProtocol implements MessageProtocol {
@Override
public byte[] marshallingRequest(Request req) {
Request temp = new Request();
temp.setServiceName(req.getServiceName());
temp.setMethod(req.getMethod());
temp.setHeaders(req.getHeaders());
temp.setPrameterTypes(req.getPrameterTypes());
if (req.getParameters() != null) {
Object[] params = req.getParameters();
Object[] serizeParmas = new Object[params.length];
for (int i = 0; i < params.length; i++) {
serizeParmas[i] = JSON.toJSONString(params[i]);
}
temp.setParameters(serizeParmas);
}
return JSON.toJSONBytes(temp);
}
@Override
public Request unmarshallingRequest(byte[] data) {
Request req = JSON.parseObject(data, Request.class);
if(req.getParameters() != null) {
Object[] serizeParmas = req.getParameters();
Object[] params = new Object[serizeParmas.length];
for(int i = 0; i < serizeParmas.length; i++) {
Object param = JSON.parseObject(serizeParmas[i].toString(), Object.class);
params[i] = param;
}
req.setParameters(params);
}
return req;
}
@Override
public byte[] marshallingResponse(Response rsp) {
Response resp = new Response();
resp.setHeaders(rsp.getHeaders());
resp.setException(rsp.getException());
resp.setReturnValue(rsp.getReturnValue());
resp.setStatus(rsp.getStatus());
return JSON.toJSONBytes(resp);
}
@Override
public Response unmarshallingResponse(byte[] data) {
return JSON.parseObject(data, Response.class);
}
}
⑤ NetClient相关
分为BIO和Netty两种模式，netty中使用了EventLoopGroup
BIO:
netty模式：
⑥ 运行结果
可以自行模拟一个消费者和一个生产者进行测试，这里就不贴出来了
说出你的愿望吧～
https://juejin.im/post/5d6809b6f265da03e276008a
以上是关于RPC框架的简单实现的主要内容，如果未能解决你的问题，请参考以下文章