开发者社区> 小飞哥1112> 正文
阿里云
为了无法计算的价值
打开APP
阿里云APP内打开

RPC原理以及示例

简介: 讲述RPC原理,以及一个简单的示例
+关注继续查看

本博客主要讲述RPC的原理以及通过一个简单的示例来讲述RPC的实现过程。之前写过关于dubbo rpc原理实现的博客,而且是偏细节的,有兴趣的话可以参考一下:

Dubbo RPC源码解读:

https://yq.aliyun.com/articles/272405

Dubbo-多线程通信原理:

https://yq.aliyun.com/articles/272406

本博客示例代码见:

git@gitee.com:wuzhengfei/great-truth.git中ConsumerTest、ProviderTest类。

一、   分析

1.  Provider分析

在java中唯一确定一个类需要知道类的全限定名(当然还有ClassLoader,不过Classloader与这里要讨论的内容关系不大,所以暂时忽略);唯一确定一个方法需要知道方法的签名,即方法名、方法参数类型。对于同一个方法,不同的参数将产生不同的结果,基于此两点,RPC的Provider收到调用请求时,至少需要从请求信息中解析出类名、方法签名、方法参数。

取得类名、方法签名、方法参数信息以后,可以通过反射为此类创建一个Proxy,通过动态代理的方式调用此方法,并返回结果。

2.  Consumer方分析

因为是远程调用,所以Consumer需要能够和远程Provider建立连接,并将请求发送的类名、方法名、方法参数类型、方法参数等信息发送过去;然后等待调用返回结果;最后将返回结果解析为需要的对象。

3.  其他分析

最简单的RPC只需要Consumer、Provider,但从维护性上、负载均衡、failover、监控等等层面看,这显然只能作为一个小孩子的玩具,而上不了台面。以下模型为dubbo的rpc模型,网上有大把的资料讲述这个模型,所以我就不再这里赘言了。

 

 

二、   demo

接下来我们使用一个简单的demo来实现一次rpc调用。

HelloService、HelloServiceImpl是真正的服务。

Provider是服务提供者。

Consumer是服务消费者。

RpcContainer中管理者Provider信息。

RpcFramework管理RPC服务的生命周期。

1.  service

1)       HelloService


public interface HelloService {
    String sayHello(String name);
}



2)       HelloServiceImpl

public classHelloServiceImpl implements HelloService {
    @Override
    public String sayHello(String name) {
        return "Hello RPC!, I'm " + name;
    }
}



 

2.  RpcContainer


public class RpcContainer {
     private static final HashMap<String, Class<?>> EXPORTED_SERVICES = newHashMap<String, Class<?>>();
     public void register(Class interfaceClass, Class interfaceImplClass) {
           EXPORTED_SERVICES.put(interfaceClass.getName(), interfaceImplClass);
     }
     public Class<?> getTargetClass(String interfaceName){
           return EXPORTED_SERVICES.get(interfaceName);
     }
}



 

3.  Provider

public class Provider implements Runnable {
     Socket socket = null;
     RpcContainer rpcContainer = null;
    
 
     public Provider(Socket socket,RpcContainer rpcContainer) {
           this.socket = socket;
           this.rpcContainer= rpcContainer ;
     }
 
     public voidrun() {
           ObjectInputStreaminput = null;
           ObjectOutputStreamoutput = null;
           try {
                // 2.将客户端发送的码流反序列化成对象,反射调用服务实现者,获取执行结果
                input = newObjectInputStream(socket.getInputStream());
                StringserviceName = input.readUTF();
                StringmethodName = input.readUTF();
                Class<?>[]parameterTypes = (Class<?>[]) input.readObject();
                Object[]arguments = (Object[]) input.readObject();
                Class<?>serviceClass = rpcContainer.getTargetClass(serviceName);
                if (serviceClass== null) {
                     throw newClassNotFoundException(serviceName + " not found");
                }
                Methodmethod = serviceClass.getMethod(methodName, parameterTypes);
                Objectresult = method.invoke(serviceClass.newInstance(), arguments);
 
                // 3.将执行结果反序列化,通过socket发送给客户端
                output = newObjectOutputStream(socket.getOutputStream());
                output.writeObject(result);
           } catch (Exception e){
                e.printStackTrace();
           }finally {
                if (output != null) {
                     try {
                          output.close();
                     }catch (IOException e){
                          e.printStackTrace();
                     }
                }
                if (input != null) {
                     try {
                          input.close();
                     }catch (IOException e){
                          e.printStackTrace();
                     }
                }
                if (socket != null) {
                     try {
                          socket.close();
                     }catch (IOException e){
                          e.printStackTrace();
                     }
                }
           }
 
     }
}




 

4.  RpcFramework


public class RpcFramework {
     private staticExecutorService executor =Executors.newFixedThreadPool(10);
 
     private RpcContainer rpcContainer;
 
     private int port;
 
     public RpcFramework(intport) {
           this.port = port;
     }
 
     public voidstart() throws IOException {
           ServerSocketserver = newServerSocket();
           server.bind(newInetSocketAddress(port));
           System.out.println("****** rpc framework started!******");
 
           try {
                while (true) {
                     // 1.监听客户端的TCP连接,接到TCP连接后将其封装成task,由线程池执行
                     Providerprovider = newProvider(server.accept(), rpcContainer);
                     executor.execute(provider);
                }
           }finally {
                server.close();
                executor.shutdown();
           }
     }
 
     public intgetPort() {
           return port;
     }
 
     public RpcContainer getRpcContainer() {
           return rpcContainer;
     }
 
     public void setRpcContainer(RpcContainerrpcContainer) {
           this.rpcContainer = rpcContainer;
     }
 
}



 

5.  Consumer

public classConsumer<T> {
     public static<T> T getRpcProxy(final Class<T>serviceInterface, finalInetSocketAddress addr) {
           // 1.将本地的接口调用转换成JDK的动态代理,在动态代理中实现接口的远程调用
           ConsumerInvocationHandlerhandler = newConsumerInvocationHandler(addr, serviceInterface);
           T proxy  = (T)Proxy.newProxyInstance(serviceInterface.getClassLoader(), new Class<?>[] { serviceInterface}, handler);
           return proxy ;
     }
 
     public static class ConsumerInvocationHandler implements InvocationHandler {
           private InetSocketAddress addr;
           private Class<?> serviceInterface;
 
           public ConsumerInvocationHandler(InetSocketAddressaddr, Class<?> serviceInterface) {
                super();
                this.addr = addr;
                this.serviceInterface = serviceInterface;
           }
 
           public Object invoke(Object proxy, Method method,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());
                     return input.readObject();
                } finally {
                     if (socket != null)
                          socket.close();
                     if (output != null)
                          output.close();
                     if (input != null)
                          input.close();
                }
           }
 
     }
}




 

6.  Test

1)       ProviderTest


public class ProviderTest {
     public static void main(String[] args)throws IOException {
           int port = 8888;
           RpcContainerrpcContainer = newRpcContainer();
           rpcContainer.register(HelloService.class, HelloServiceImpl.class);
           RpcFrameworkserviceServer = newRpcFramework(port);
           serviceServer.setRpcContainer (rpcContainer);
           serviceServer.start();
     }
}


 

2)       ConsumerTest


public class ConsumerTest {
     public static void main(String[] args)throws IOException {
           int port =8888;
           HelloServiceservice = Consumer.getRpcProxy(HelloService.class, newInetSocketAddress("localhost", port));
           String name = "wzf";
           String result = service.sayHello(name);
           System.err.println("name="+ name + " result=" + result);
     }
}



 

三、   Demo的问题

demo中的示例太过于简单,缺少一下几部分内容:

²  consumer无法发现provider的服务。

²  Consumer无法及时感知provider是否可用。

²  没有负载均衡功能。

²  Consumer、provider均不支持多线程,多线程情况下数据会错乱。

²  不支持集群。

²  其他性能问题:demo使用的是阻塞时io,高并发情况下不可取。

²  等等。。。

这些问题dubbo都已经非常好的解决了,关于dubbo的实现,在博客中已经有详细的讨论,此处就不再讨论了。

 

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
SpringMVC札集(03)——基于注解的SpringMVC入门完整详细示例
自定义View系列教程00–推翻自己和过往,重学自定义View 自定义View系列教程01–常用工具介绍 自定义View系列教程02–onMeasure源码详尽分析 自定义View系列教程0...
1149 0
SpringMVC札集(02)——SpringMVC入门完整详细示例(下)
自定义View系列教程00–推翻自己和过往,重学自定义View 自定义View系列教程01–常用工具介绍 自定义View系列教程02–onMeasure源码详尽分析 自定义View系列教程0...
1104 0
SpringMVC札集(01)——SpringMVC入门完整详细示例(上)
自定义View系列教程00–推翻自己和过往,重学自定义View 自定义View系列教程01–常用工具介绍 自定义View系列教程02–onMeasure源码详尽分析 自定义View系列教程0...
1446 0
SpringMVC解析1-使用示例
Spring MVC分离了控制器、模型对象、分派器以及处理程序对象的角色,这种分离让它们更容易进行定制。Spring的MVC是基于servlet功能实现的,通过实现Servlet接口的DispatchSerlver来封装其核心功能实现,通过将请求分派给处理程序,同时带有可配置的处理程序映射,视图解析,本地语言,主题解析以及上载文件支持。
652 0
BFC原理剖析
本文讲了BFC的概念是什么; BFC的约束规则;咋样才能触发生成新的BFC;BFC在布局中的应用:防止margin重叠(塌陷,以最大的为准); 清除内部浮动;自适应两(多)栏布局。 1. BFC是什么? Block fomatting context = block-level box + Form...
1279 0
BFC 原理
首先我们了解下何为** BFC** bfc:Formatting context(格式化上下文) 是 W3C CSS2.1 规范中的一个概念。它是页面中的一块渲染区域,并且有一套渲染规则,它决定了其子元素将如何定位,以及和其他元素的关系和相互作用。
0 0
+关注
小飞哥1112
java,架构相关技术专家
文章
问答
文章排行榜
最热
最新
相关电子书
更多
探究 Node.js 的服务端之路
立即下载
低代码开发师(初级)实战教程
立即下载
阿里巴巴DevOps 最佳实践手册
立即下载