RPC原理以及示例

  1. 云栖社区>
  2. 博客>
  3. 正文

RPC原理以及示例

小飞哥1112 2018-01-20 14:17:14 浏览1180
展开阅读全文

本博客主要讲述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的实现,在博客中已经有详细的讨论,此处就不再讨论了。

 

网友评论

登录后评论
0/500
评论
小飞哥1112
+ 关注