首页 > 代码库 > java 远程调用 RPC

java 远程调用 RPC

2024-11-25 05:37:02   203人阅读

1. 概念

　　RPC，全称为Remote Procedure Call，即远程过程调用，它是一个计算机通信协议。它允许像调用本地服务一样调用远程服务。它可以有不同的实现方式。如RMI(远程方法调用)、Hessian、Http invoker等。RPC是与语言无关的。直观说法就是A通过网络调用B的过程方法。也就是说两台服务器A，B，一个应用部署在A服务器上，想要调用B服务器上应用提供的函数/方法，由于不在一个内存空间，不能直接调用，需要通过网络来表达调用的语义和传达调用的数据。

 

1、首先要解决寻址的问题，也就是说，A服务器上的应用怎么告诉底层的RPC框架，B服务器的IP，以及应用绑定的端口，还有方法的名称，这样才能完成调用

2、方法的参数需要通过底层的网络协议如TCP传递到B服务器，由于网络协议是基于二进制的，内存中的参数的值要序列化成二进制的形式

3、在B服务器上完成寻址后，需要对参数进行反序列化，恢复为内存中的表达方式，然后找到对应的方法进行本地调用，然后得到返回值，

4、返回值还要发送回服务器A上的应用，也要经过序列化的方式发送，服务器A接到后，再反序列化，恢复为内存中的表达方式，交给应用

2. RPC的实现

　　RPC能够让本地应用简单、高效地调用服务器中的过程（服务）。它主要应用在分布式系统。如Hadoop中的IPC组件。但怎样实现一个RPC框架呢？

　　从下面几个方面思考，仅供参考：

　　1.通信模型：假设通信的为A机器与B机器，A与B之间有通信模型，在Java中一般基于BIO或NIO；。

　　2.过程（服务）定位：使用给定的通信方式，与确定IP与端口及方法名称确定具体的过程或方法；

　　3.远程代理对象：本地调用的方法(服务)其实是远程方法的本地代理，因此可能需要一个远程代理对象，对于Java而言，远程代理对象可以使用Java的动态对象实现，封装了调用远程方法调用；

　　4.序列化，将对象名称、方法名称、参数等对象信息进行网络传输需要转换成二进制传输，这里可能需要不同的序列化技术方案。如:protobuf，Arvo等。

2.1 实现技术方案

下面使用比较原始的方案实现RPC框架，采用Socket通信、动态代理与反射与Java原生的序列化。

2.2 RPC框架架构

RPC架构分为三部分：

1）服务提供者，运行在服务器端，提供服务接口定义与服务实现类。

2）服务中心，运行在服务器端，负责将本地服务发布成远程服务，管理远程服务，提供给服务消费者使用。

3）服务消费者，运行在客户端，通过远程代理对象调用远程服务。

2.3 Demo

2.3.1 服务提供者接口定义

<style></style>

HelloService.java

1 package com.loveincode.rpc;2 3 public interface HelloService {4 5     String hello(String name);6 7 }

2.3.2 服务提供者接口实现

HelloServiceImpl.java

 1 package com.loveincode.rpc; 2  3 //HelloServices接口实现类： 4 public class HelloServiceImpl implements HelloService { 5       6     public String hello(String name) { 7         return "hello, " + name; 8     } 9  10 }

 

2.3.3 RPC框架服务中心

RpcFramework.java

  1 package com.loveincode.rpc;  2   3 import java.io.ObjectInputStream;  4 import java.io.ObjectOutputStream;  5 import java.lang.reflect.InvocationHandler;  6 import java.lang.reflect.Method;  7 import java.lang.reflect.Proxy;  8 import java.net.ServerSocket;  9 import java.net.Socket; 10  11 public class RpcFramework { 12     /** 13      * 暴露服务 14      *  15      * @param service 16      *            服务实现 17      * @param port 18      *            服务端口 19      * @throws Exception 20      */ 21     public static void export(final Object service, int port) throws Exception { 22         if (service == null) 23             throw new IllegalArgumentException("service instance == null"); 24         if (port <= 0 || port > 65535) 25             throw new IllegalArgumentException("Invalid port " + port); 26         System.out.println("Export service " + service.getClass().getName() + " on port " + port); 27         ServerSocket server = new ServerSocket(port);// 前面都是验证调用是否符合规则，这里在被调用端开一个服务。下面就是死循环运行。 28         for (;;) { 29             try { 30                 final Socket socket = server.accept(); 31                 new Thread(new Runnable() {// 对每一个请求new一个线程，匿名类 32                     @Override 33                     public void run() { 34                         try { 35                             try { 36                                 ObjectInputStream input = new ObjectInputStream(socket.getInputStream()); 37                                 try { 38                                     String methodName = input.readUTF(); 39                                     Class<?>[] parameterTypes = (Class<?>[]) input.readObject(); 40                                     Object[] arguments = (Object[]) input.readObject(); 41                                     ObjectOutputStream output = new ObjectOutputStream(socket.getOutputStream());// 接收客户端传来的方法名、参数类型、参数 42                                     try { 43                                         Method method = service.getClass().getMethod(methodName, parameterTypes);// 在本地生成对应的方法， 44                                         Object result = method.invoke(service, arguments);// 调用 45                                         output.writeObject(result);// 返回结果 46                                     } catch (Throwable t) { 47                                         output.writeObject(t); 48                                     } finally { 49                                         output.close(); 50                                     } 51                                 } finally { 52                                     input.close(); 53                                 } 54                             } finally { 55                                 socket.close(); 56                             } 57                         } catch (Exception e) { 58                             e.printStackTrace(); 59                         } 60                     } 61                 }).start(); 62             } catch (Exception e) { 63                 e.printStackTrace(); 64             } 65         } 66     } 67  68     /** 69      * 引用服务 70      *  71      * @param <T> 72      *            接口泛型 73      * @param interfaceClass 74      *            接口类型 75      * @param host 76      *            服务器主机名 77      * @param port 78      *            服务器端口 79      * @return 远程服务 80      * @throws Exception 81      */ 82     @SuppressWarnings("unchecked") 83     public static <T> T refer(final Class<T> interfaceClass, final String host, final int port) throws Exception { 84         if (interfaceClass == null) 85             throw new IllegalArgumentException("Interface class == null"); 86         if (!interfaceClass.isInterface()) 87             throw new IllegalArgumentException("The " + interfaceClass.getName() + " must be interface class!"); 88         if (host == null || host.length() == 0) 89             throw new IllegalArgumentException("Host == null!"); 90         if (port <= 0 || port > 65535) 91             throw new IllegalArgumentException("Invalid port " + port); 92         System.out.println("Get remote service " + interfaceClass.getName() + " from server " + host + ":" + port); 93         return (T) Proxy.newProxyInstance(interfaceClass.getClassLoader(), new Class<?>[] { interfaceClass }, 94                 new InvocationHandler() {// 用动态代理的方法进行包装，看起来是在调用一个方法，其实在内部通过socket通信传到服务器，并接收运行结果 95                     public Object invoke(Object proxy, Method method, Object[] arguments) throws Throwable { 96                         Socket socket = new Socket(host, port); 97                         try { 98                             ObjectOutputStream output = new ObjectOutputStream(socket.getOutputStream()); 99                             try {100                                 output.writeUTF(method.getName());101                                 output.writeObject(method.getParameterTypes());102                                 output.writeObject(arguments);103                                 ObjectInputStream input = new ObjectInputStream(socket.getInputStream());104                                 try {105                                     Object result = input.readObject();106                                     if (result instanceof Throwable) {107                                         throw (Throwable) result;108                                     }109                                     return result;// 返回结果110                                 } finally {111                                     input.close();112                                 }113                             } finally {114                                 output.close();115                             }116                         } finally {117                             socket.close();118                         }119                     }120                 });121     }122 }

 

2.3.4 服务提供者

RpcProvider.java

1 package com.loveincode.rpc;2 3 public class RpcProvider {4     public static void main(String[] args) throws Exception {5         HelloService service = new HelloServiceImpl();6         RpcFramework.export(service, 1234);7     }8 }

 

2.3.5 服务消费者

RpcConsumer.java

package com.loveincode.rpc;public class RpcConsumer {    public static void main(String[] args) throws Exception {        HelloService service = RpcFramework.refer(HelloService.class, "127.0.0.1", 1234);        for (int i = 0; i < Integer.MAX_VALUE; i++) {            String hello = service.hello("World " + i);            System.out.println(hello);            Thread.sleep(1000);        }    }}

 

运行结果

<style></style>

执行RpcProvider 结果：

Export service com.loveincode.rpc.HelloServiceImpl on port 1234

 

执行RpcConsumer 结果：

Get remote service com.loveincode.rpc.HelloService from server 127.0.0.1:1234hello, World 0hello, World 1hello, World 2hello, World 3hello, World 4hello, World 5hello, World 6hello, World 7hello, World 8hello, World 9
...

 

java 远程调用 RPC


联系
我们