从头创建一个简单的 RPC 服务框架

2,460 阅读5分钟
原文链接: blog.csdn.net
  • 概念解释

    RPC(Remote Procedure Call Protocol)——远程过程调用协议,它是一种通过网络从远程计算机程序上请求服务,而不需要了解底层网络技术的协议。RPC采用客户机/服务器模式。请求程序就是一个客户机,而服务提供程序就是一个服务器。首先,客户机调用进程发送一个有进程参数的调用信息到服务进程,然后等待应答信息。在服务器端,进程保持睡眠状态直到调用信息到达为止。当一个调用信息到达,服务器获得进程参数,计算结果,发送答复信息,然后等待下一个调用信息,最后,客户端调用进程接收答复信息,获得进程结果,然后调用执行继续进行。

  • ServerSocket 与 Socket

    • 客户端与服务端建立连接基本原理
      http://hi.csdn.net/attachment/201009/12/0_1284280120c10H.gif
    • 客户端与服务端建立连接流程
      http://hi.csdn.net/attachment/201009/12/0_1284280127Kfu0.gif
    • ServerSocket

      ServerSocket server = new ServerSocket(1234);  
      Socket socket = server.accept(); 

      创建并监听一个端口为1234的ServerSocket对象,然后调用了ServerSocket对象的accept()方法,这个方法的执行将使Server端的程序处于阻塞状态,程序将一直阻塞直到捕捉到一个来自Client端的请求,并返回一个用于与该Client通信的Socket对象。此后Server程序只要向这个Socket对象读写数据,就可以实现向远端的Client读写数据。结束监听时,关闭ServerSocket对象

    • Socket

      Socket socket = new Socket("127.0.0.1", 1234);

      客户端创建一个socket对象,与服务端建立连接,服务端管理客户连接请求的任务由操作系统完成,操作系统把连接请求存储在一个先进先出的队列中,队列长度一般为50。当服务端连接队列长度大于50时,连接被拒绝。

  • 动态代理机制
    Java的动态代理机制中,有两个重要的类和接口,Proxy、InvocationHandler,他们都在java.lang.reflect包下,这个类和接口是在实现动态代理过程中必须用到的

    • Proxy
      Proxy用于创建一个代理类对象,它提供了许多方法,其中应用最多的是 newProxyInstance这个方法。

      public static Object newProxyInstance(ClassLoader loader, Class<?>[] interfaces, InvocationHandler h)

      参数解释:
      loader: 一个ClassLoader对象,定义了代理对象是由哪个Classloader对象来加载
      interfaces: 一个interface数组,表示代理我们将为代理对象提供一组什么接口,如果我提供了一组接口,则代理对象就宣称实现了该组接口,这样我们就能通过代理对象调用相应接口了
      h: 一个InvocationHandler对象,表示动态代理对象在调用方法时具体执行的对象

    • InvocationHandler
      每个动态代理类都必须实现InvocationHandler接口,且每个代理类的实例都关联到一个InvocationHandler对象,当代理对象调用一个方法时,这个方法的调用会映射到InvocationHandler的invoke方法执行

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

      参数解释:
      proxy: 指代代理的真实对象
      method: 指代所要调用的真实对象的某个方法
      args: 指代调用真实对象方法时接受的参数

    • 动态代理示例

      • 定义测试接口

        public interface TestHello {
        
            void hello(String name);
        
            void sayGood();
        }
      • 定义接口实现

        public class TestHelloImpl implements TestHello{
        
            @Override
            public void hello(String name) {
                System.out.println("hello " + name);
            }
        
            @Override
            public void sayGood() {
                System.out.println("you are the best!");
            }
        }
      • 定义动态代理类

        public class MyProxy implements InvocationHandler {
        
            private Object proxy;
        
            public MyProxy(Object proxy) {
                this.proxy = proxy;
            }
        
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                method.invoke(this.proxy, args);
                return null;
            }
        
        }
      • 模拟代理调用

        public class Client {
        
            public static void main(String[] args) {
        
                TestHello th = new TestHelloImpl();
        
                MyProxy mp = new MyProxy(th);
        
                TestHello thProxy =
                    (TestHello) Proxy.newProxyInstance(mp.getClass().getClassLoader(), th.getClass().getInterfaces(), mp);
        
                thProxy.hello("Jerry");
        
                thProxy.sayGood();
            }
        }
      • 测试输出

        hello Jerry
        you are the best!
  • 一个简单的RPC实现

    • 简单RPC框架类,定义服务注册方法 export, 服务消费方法 refer

      public class RpcFramework {
      
          /** 
           * 暴露服务 
           *  
           * @param service 服务实现 
           * @param port 服务端口 
           * @throws Exception 
           */  
          public static void export(final Object service, int port) throws Exception {  
              if (service == null)  
                  throw new IllegalArgumentException("service instance == null");  
              if (port <= 0 || port > 65535)  
                  throw new IllegalArgumentException("Invalid port " + port);  
              System.out.println("Export service " + service.getClass().getName() + " on port " + port);  
              ServerSocket server = new ServerSocket(port);  
              for(;;) {  
                  try {  
                      final Socket socket = server.accept();  
                      new Thread(new Runnable() {  
                          @Override  
                          public void run() { 
                              try {  
                                  try {  
                                      ObjectInputStream input = new ObjectInputStream(socket.getInputStream());  
                                      try {  
                                          String methodName = input.readUTF(); 
                                          Class<?>[] parameterTypes = (Class<?>[])input.readObject();  
                                          Object[] arguments = (Object[])input.readObject();  
                                          ObjectOutputStream output = new ObjectOutputStream(socket.getOutputStream());  
                                          try {  
                                              Method method = service.getClass().getMethod(methodName, parameterTypes);  
                                              Object result = method.invoke(service, arguments);  
                                              output.writeObject(result);
                                          } catch (Throwable t) {  
                                              output.writeObject(t);  
                                          } finally {  
                                              output.close();  
                                          }  
                                      } finally {  
                                          input.close();  
                                      }  
                                  } finally {  
                                      socket.close();  
                                  }
                              } catch (Exception e) {  
                                  e.printStackTrace();  
                              }
                          }  
                      }).start();  
                  } catch (Exception e) {  
                      e.printStackTrace();  
                  }
              }
          }  
      
          /** 
           * 引用服务 
           *  
           * @param <T> 接口泛型 
           * @param interfaceClass 接口类型 
           * @param host 服务器主机名 
           * @param port 服务器端口 
           * @return 远程服务 
           * @throws Exception 
           */  
          @SuppressWarnings("unchecked")  
          public static <T> T refer(final Class<T> interfaceClass, final String host, final int port) throws Exception {  
              if (interfaceClass == null)  
                  throw new IllegalArgumentException("Interface class == null");  
              if (! interfaceClass.isInterface())  
                  throw new IllegalArgumentException("The " + interfaceClass.getName() + " must be interface class!");  
              if (host == null || host.length() == 0)  
                  throw new IllegalArgumentException("Host == null!");  
              if (port <= 0 || port > 65535)  
                  throw new IllegalArgumentException("Invalid port " + port);  
              System.out.println("Get remote service " + interfaceClass.getName() + " from server " + host + ":" + port);  
              return (T) Proxy.newProxyInstance(interfaceClass.getClassLoader(), new Class<?>[] {interfaceClass}, new InvocationHandler() {  
                  public Object invoke(Object proxy, Method method, Object[] arguments) throws Throwable {  
                      Socket socket = new Socket(host, port);  
                      try {  
                          ObjectOutputStream output = new ObjectOutputStream(socket.getOutputStream());  
                          try {  
                              output.writeUTF(method.getName());  
                              output.writeObject(method.getParameterTypes());  
                              output.writeObject(arguments);  
                              InputStream is = socket.getInputStream();
                              ObjectInputStream input = new ObjectInputStream(is);  
                              try {  
                                  Object result = input.readObject();  
                                  if (result instanceof Throwable) {  
                                      throw (Throwable) result;  
                                  }  
                                  return result;  
                              } finally {  
                                  input.close();  
                              }  
                          } finally {  
                              output.close();  
                          }  
                      } finally {  
                          socket.close();  
                      }  
                  }  
              });  
      
      }  
      
      }
    • 定义具体接口和实现

      public interface HelloService {
      
          String hello(String name);  
      
      }
      public class HelloServiceImpl implements HelloService{
      
          @Override
          public String hello(String name) {
              return "hello " + name;
          }
      
      }
    • 生成并注册服务对象到服务端

      public class RpcProvider {
      
          public static void main(String[] args) {
              try {
                  HelloService service = new HelloServiceImpl();  
                  RpcFramework.export(service, 1235);            
              } catch (Exception e) {
                  e.printStackTrace();
              }
          }
      }
    • 客户端声明服务端定义的服务对象,使用服务

      public class RpcConsumer {
      
          public static void main(String[] args) {
              try {
                  HelloService service = RpcFramework.refer(HelloService.class, "127.0.0.1", 1235);  
                  for (int i = 0; i < Integer.MAX_VALUE; i ++) {  
                      String hello = service.hello("World" + i);  
                      System.out.println(hello);  
                      Thread.sleep(1000);  
                  }              
              } catch (Exception e) {
                  e.printStackTrace();
              }
          }
      }
    • 本文多数内容非原创,仅供学习使用,欢迎指正!
  • 参考