基于TCP的RPC简单实现

所谓RPC就是远程方法调用(Remote  Process Call ),简单的来说就是通过MQ,TCP,HTTP或者自己写的网络协议来传输我要调用对方的什么接口,对方处理之后再把结果返回给我.就这么简单的一个过程.在一个大型的项目之后基本上各模块都是分开的,以提供服务的方式进行相互调用.如果能够提供智能负载均衡,可选择的java对象编码解码协议,网络传输协议,服务监控,服务版本控制等很多功能的话就是一个SOA架构了.

 

前两天实现了一个基于java Socket 实现的阻塞的RPC.其原理非常简单

1. 客户端用一个TransportMessage类去包装需要调用的接口,调用的方法,调用方法的参数类型,调用方法的参数值.

2. 客户端用Socet连接服务端,序列化TransportMessage,传输给服务端.

3. 服务端循环接收请求,一旦受到请求就起一个线程扔到线程池去执行,执行的内容就是反序列化TransportMessage类,在servicePool池中获取接口实现类,通过调用方法参数类型数组获取Method对象.然后通过method.invoke去调用方法.

4. 服务器端序列化结果,然后通过socket传输给客户端.

5. 客户端收到结果,反序列化结果对象.

 

具体代码实现,(为了节省篇幅,setter,getter就不放进来了):

1.远程调用信息封装   TransportMessage.java

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31

/**  * @author Lubby  * @date 2015年4月22日 下午1:06:18  * 远程调用信息封装.  * 包括    1.调用接口名称  (包名+接口名)   2.调用方法名  3.调用参数Class类型数组  4.调用接口的参数数组  */ public class TransportMessage implements Serializable {     //包名+接口名称  如com.lubby.rpc.service.MathService.     private String interfaceName;     //调用方法名   如 getSum     private String methodName;     //参数类型 按照接口参数顺序  getSum(int a, int b, String name)方法就是int.class int.class String.class的数组     private Class[] paramsTypes;     //参数 按照接口参数顺序 getSum(int a, int b, String name)方法就是 1,3,"Tom"的数组     private Object[] parameters;       public TransportMessage() {         super();         // TODO Auto-generated constructor stub     }       public TransportMessage(String interfaceName, String methodName,             Class[] paramsTypes, Object[] parameters) {         super();         this.interfaceName = interfaceName;         this.methodName = methodName;         this.paramsTypes = paramsTypes;         this.parameters = parameters;     }   }

2.客户端调用远程方法类 RPCClient.java

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67

public class RPCClient {     // 服务端地址     private String serverAddress;     // 服务端端口     private int serverPort;     // 线程池大小     private int threadPoolSize = 10;     // 线程池     private ExecutorService executorService = null;       public RPCClient() {         super();         // TODO Auto-generated constructor stub     }       /**      * @param serverAddress      *            TPC服务地址      * @param serverPort      *            TPC服务端口      *       */     public RPCClient(String serverAddress, int serverPort) {         this.serverAddress = serverAddress;         this.serverPort = serverPort;         executorService = Executors.newFixedThreadPool(threadPoolSize);     }       /**      * 同步的请求和接收结果      *       * @param transportMessage      * @return      */     public Object sendAndReceive(TransportMessage transportMessage) {         Object result = null;         Socket socket = null;         try {              socket = new Socket(serverAddress, serverPort);                             //反序列化 TransportMessage对象             ObjectOutputStream objectOutpusStream = new ObjectOutputStream(                     socket.getOutputStream());             objectOutpusStream.writeObject(transportMessage);               ObjectInputStream objectInputStream = new ObjectInputStream(                     socket.getInputStream());             //阻塞等待读取结果并反序列化结果对象             result = objectInputStream.readObject();             socket.close();         } catch (UnknownHostException e) {             e.printStackTrace();         } catch (IOException e) {             e.printStackTrace();         } catch (ClassNotFoundException e) {             e.printStackTrace();         }finally{             try {                 //最后关闭socket                 socket.close();             } catch (IOException e) {                 e.printStackTrace();             }         }         return result;     } }

3.服务器处理类 RPCServer.java

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174

public class RPCServer {     private int threadSize = 10;     private ExecutorService threadPool;     private Map<String, Object> servicePool;     private int port = 4321;       public RPCServer() {         super();         synchronized (this) {             threadPool = Executors.newFixedThreadPool(this.threadSize);         }     }       /**      *       * @param threadSize      *            内部处理线程池大小      * @param port      *            当前TPC服务的端口号      *       */       public RPCServer(int threadSize, int port) {         this.threadSize = threadSize;         this.port = port;         synchronized (this) {             threadPool = Executors.newFixedThreadPool(this.threadSize);         }     }       /**      *       *       * @param servicePool      *            装有service对象的Map, Key为全限定接口名,Value为接口实现类对象      * @param threadSize      *            内部处理线程池大小      * @param port      *            当前TPC服务的端口号      *       */     public RPCServer(Map<String, Object> servicePool, int threadSize, int port) {         this.threadSize = threadSize;         this.servicePool = servicePool;         this.port = port;         synchronized (this) {             threadPool = Executors.newFixedThreadPool(this.threadSize);         }     }       /**      * RPC服务端处理函数 监听指定TPC端口,每次有请求过来的时候调用服务,放入线程池中处理.      *       * @throws IOException      */     public void service() throws IOException {         ServerSocket serverSocket = new ServerSocket(port);         while (true) {             Socket receiveSocket = serverSocket.accept();             final Socket socket = receiveSocket;             threadPool.execute(new Runnable() {                   public void run() {                     try {                         process(socket);                       } catch (ClassNotFoundException e) {                         e.printStackTrace();                     } catch (NoSuchMethodException e) {                         e.printStackTrace();                     } catch (SecurityException e) {                         e.printStackTrace();                     } catch (IllegalAccessException e) {                         e.printStackTrace();                     } catch (IllegalArgumentException e) {                         e.printStackTrace();                     } catch (InvocationTargetException e) {                         e.printStackTrace();                     } catch (InstantiationException e) {                         e.printStackTrace();                     } catch (IOException e) {                         e.printStackTrace();                     } finally {                         try {                             socket.close();                         } catch (IOException e) {                             e.printStackTrace();                         }                     }                   }             });         }       }       /**      * 调用服务 通过TCP Socket返回结果对象      *       * @param receiveSocket      *            请求Socket      * @throws IOException      * @throws ClassNotFoundException      * @throws NoSuchMethodException      * @throws SecurityException      * @throws IllegalAccessException      * @throws IllegalArgumentException      * @throws InvocationTargetException      * @throws InstantiationException      */     private void process(Socket receiveSocket) throws IOException,             ClassNotFoundException, NoSuchMethodException, SecurityException,             IllegalAccessException, IllegalArgumentException,             InvocationTargetException, InstantiationException {           /*          * try { Thread.sleep(10000); } catch (InterruptedException e) { // TODO          * Auto-generated catch block e.printStackTrace(); }          */         ObjectInputStream objectinputStream = new ObjectInputStream(                 receiveSocket.getInputStream());         TransportMessage message = (TransportMessage) objectinputStream                 .readObject();           // 调用服务         Object result = call(message);           ObjectOutputStream objectOutputStream = new ObjectOutputStream(                 receiveSocket.getOutputStream());         objectOutputStream.writeObject(result);         objectinputStream.close();         objectOutputStream.close();     }       /**      * 服务处理函数 通过包名+接口名在servicePool中找到对应服务 通过调用方法参数类型数组获取Method对象      * 通过Method.invoke(对象,参数)调用对应服务      *       * @return      * @throws ClassNotFoundException      * @throws SecurityException      * @throws NoSuchMethodException      * @throws InvocationTargetException      * @throws IllegalArgumentException      * @throws IllegalAccessException      * @throws InstantiationException      */     private Object call(TransportMessage message)             throws ClassNotFoundException, NoSuchMethodException,             SecurityException, IllegalAccessException,             IllegalArgumentException, InvocationTargetException,             InstantiationException {         if (servicePool == null) {             synchronized (this) {                 servicePool = new HashMap<String, Object>();             }         }         String interfaceName = message.getInterfaceName();         Object service = servicePool.get(interfaceName);         Class<?> serviceClass = Class.forName(interfaceName);         // 检查servicePool中对象,若没有着生产对象         if (service == null) {             synchronized (this) {                 service = serviceClass.newInstance();                 servicePool.put(interfaceName, service);             }         }         Method method = serviceClass.getMethod(message.getMethodName(),                 message.getParamsTypes());         Object result = method.invoke(service, message.getParameters());         return result;       } }

4.为了方便测试写了个接口和其实现类 MathService 和 MathServiceImpl

?

1 2 3

public interface MathService {     public int getSum(int a, int b, String name); }

?

1 2 3 4 5 6 7

public class MathServiceImpl implements MathService {     public int getSum(int a, int b, String name) {         System.out.println(name);         return a + b;     }   }

5.服务器端测试代码

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

public class ServerTest {           public static void main(String[] args){         Map<String,Object> servicePool = new  HashMap<String, Object>();         servicePool.put("com.lubby.rpc.service.MathService", new MathServiceImpl());         RPCServer server = new RPCServer(servicePool,4, 4321);         try {             server.service();         } catch (IOException e) {             e.printStackTrace();         }               }   }

6.客户端测试代码

?

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33

public class ClientTest {     public static void main(String[] args) {         String serverAddress = "127.0.0.1";         int serverPort = 4321;                   final RPCClient client = new RPCClient(serverAddress, serverPort);         final TransportMessage transportMessage = buildTransportMessage();                   for (int i = 0; i < 1000; i++) {             final int waitTime = i * 10;             new Thread(new Runnable() {                 public void run() {                     Object result = client.sendAndReceive(transportMessage);                     System.out.println(result);                 }             }).start();         }     }       private static TransportMessage buildTransportMessage() {           String interfaceName = "com.lubby.rpc.service.MathService";         Class[] paramsTypes = { int.class, int.class, String.class };         Object[] parameters = { 1, 3, "Lubby" };         String methodName = "getSum";           TransportMessage transportMessage = new TransportMessage(interfaceName,                 methodName, paramsTypes, parameters);           return transportMessage;     }   }

 

7.并发问题

由于ServerSocket是阻塞的,所以在ServerSocket.accept()方法同一时刻只能有一个线程进入,虽然之后的处理都另起一个线程,但是有瓶颈的,

我在用400个线程并发连接服务端的时候基本没问题,但是500个线程并发连接服务端的时候就会有部分线程连接不到服务器端.后面看下NIO回头用NIO来改一下.

 

v

http://my.oschina.net/u/2250599/blog/406474