爲何要寫這個RPC
市面上常見的RPC框架不少，grpc，motan，dubbo等，可是隨着愈來愈多的元素加入，複雜的架構設計等因素似使得這些框架就想spring同樣，雖然號稱是輕量級，可是用起來倒是讓咱們很蹩腳，大量的配置，繁雜的API設計，其實，咱們根本用不上這些東西！！！ 我也算得上是在不少個互聯網企業廝殺過，見過不少不少的內部RPC框架，有些優秀的設計讓我很是讚揚，有一天我忽然想着，爲何不對這些設計原型進行聚合歸類，本身搞一套【輕量級】RPC框架呢，礙於工做緣由，一直沒有時間倒騰出空，十一期間工做閒暇，說搞就搞吧，但願源碼對你們對認識RPC框架起到推動的做用。東西越寫越多，有各類問題歡迎隨時拍磚java
RPC基礎概念spring
RPC就是遠程過程調用協議，其做用就是客戶端與服務端之間的遠程調用，就像本地本身調用同樣，讓服務端進行服務化，功能惟一性，負載熱點流量。apache
RPC框架的實現
RPC可以讓本地應用簡單、高效地調用服務器中的過程（服務）。它主要應用在分佈式系統。如Hadoop中的IPC組件。但怎樣實現一個RPC框架呢？服務器
從下面幾個方面思考，僅供參考：網絡
1.通訊模型：假設通訊的爲A機器與B機器，A與B之間有通訊模型，在Java中通常基於BIO或NIO；。架構
2.過程（服務）定位：使用給定的通訊方式，與肯定IP與端口及方法名稱肯定具體的過程或方法；框架
3.遠程代理對象：本地調用的方法(服務)實際上是遠程方法的本地代理，所以可能須要一個遠程代理對象，對於Java而言，遠程代理對象可使用Java的動態對象實現，封裝了調用遠程方法調用；異步
4.序列化，將對象名稱、方法名稱、參數等對象信息進行網絡傳輸須要轉換成二進制傳輸，這裏可能須要不一樣的序列化技術方案。如:thrift,protobuf,Arvo等。socket
下面用簡單的原生java socket來實現rpc調用，方便你們更深層瞭解rpc原理tcp
服務類接口
package socketrpc;
2
3 public interface IHello {
4 String sayHello(String string);
5 }
服務實現類
1 package socketrpc.server;
2
3 import socketrpc.IHello;
4
5 public class HelloServiceImpl implements IHello {
6
7 @Override
8 public String sayHello(String string) {
9 return "你好:".concat ( string );
10 }
11 }
客戶端實現
package socketrpc.client;
import socketrpc.IHello;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;
import java.net.Socket;
public class RpcClientProxy<T> implements InvocationHandler {
private Class<T> serviceInterface;
private InetSocketAddress addr;
public RpcClientProxy(Class<T> serviceInterface, String ip,String port) {
this.serviceInterface = serviceInterface;
this.addr = new InetSocketAddress(ip, Integer.parseInt ( port ));
}
public T getClientIntance(){
return (T) Proxy.newProxyInstance (serviceInterface.getClassLoader(),new Class<?>[]{serviceInterface},this);
}
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
Socket socket = null;
ObjectOutputStream output = null;
ObjectInputStream input = null;
try {
// 2.建立Socket客戶端，根據指定地址鏈接遠程服務提供者
socket = new Socket();
socket.connect(addr);
// 3.將遠程服務調用所需的接口類、方法名、參數列表等編碼後發送給服務提供者
output = new ObjectOutputStream(socket.getOutputStream());
output.writeUTF(serviceInterface.getName());
output.writeUTF(method.getName());
output.writeObject(method.getParameterTypes());
output.writeObject(args);
// 4.同步阻塞等待服務器返回應答，獲取應答後返回
input = new ObjectInputStream(socket.getInputStream());
return input.readObject();
} finally {
if (socket != null) socket.close();
if (output != null) output.close();
if (input != null) input.close();
}
}
public static void main(String[] args) {
RpcClientProxy client = new RpcClientProxy<>(IHello.class,"localhost","6666");
IHello hello = (IHello) client.getClientIntance ();
System.out.println (hello.sayHello ( "socket rpc" ));
}
}
服務端實現
package socketrpc.server;
import socketrpc.IHello;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.HashMap;
public class RpcServer {
private static final HashMap<String, Class<?>> serviceRegistry = new HashMap<>();
private int port;
public RpcServer(int port) {
this.port =port;
}
public RpcServer register(Class serviceInterface, Class impl) {
serviceRegistry.put(serviceInterface.getName(), impl);
return this;
}
public void run() throws IOException {
ServerSocket server = new ServerSocket();
server.bind(new InetSocketAddress (port));
System.out.println("start server");
ObjectInputStream input =null;
ObjectOutputStream output =null;
Socket socket=null;
try {
while(true){
socket = server.accept ();
input =new ObjectInputStream(socket.getInputStream());
String serviceName = input.readUTF();
String methodName = input.readUTF();
System.out.println (methodName);
Class<?>[] parameterTypes = (Class<?>[]) input.readObject();
Object[] arguments = (Object[]) input.readObject();
Class serviceClass = serviceRegistry.get(serviceName);
if (serviceClass == null) {
throw new ClassNotFoundException(serviceName + " not found");
}
Method method = serviceClass.getMethod(methodName, parameterTypes);
Object result = method.invoke(serviceClass.newInstance(), arguments);
output = new ObjectOutputStream (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();
}
}
}
}
public static void main(String[] args) throws IOException {
new RpcServer ( 6666 ).register ( IHello.class,HelloServiceImpl.class).run ();
}
}
啓動服務端和客戶端後運行結果以下
客戶端代碼簡單講解
private Class<T> serviceInterface;
2 private InetSocketAddress addr;
3
4 public RpcClientProxy(Class<T> serviceInterface, String ip,String port) {
5 this.serviceInterface = serviceInterface;
6 this.addr = new InetSocketAddress(ip, Integer.parseInt ( port ));
7 }
8
9 public T getClientIntance(){
10 return (T) Proxy.newProxyInstance (serviceInterface.getClassLoader(),new Class<?>[]{serviceInterface},this);
11 }
傳入接口類和ip端口,調用getClientIntance方法時,對當前接口進行代理,實際調用方法爲
1 @Override
2 public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
3
4 Socket socket = null;
5 ObjectOutputStream output = null;
6 ObjectInputStream input = null;
7
8 try {
9 // 2.建立Socket客戶端，根據指定地址鏈接遠程服務提供者
10 socket = new Socket();
11 socket.connect(addr);
12
13 // 3.將遠程服務調用所需的接口類、方法名、參數列表等編碼後發送給服務提供者
14 output = new ObjectOutputStream(socket.getOutputStream());
15 output.writeUTF(serviceInterface.getName());
16 output.writeUTF(method.getName());
17 output.writeObject(method.getParameterTypes());
18 output.writeObject(args);
19
20 // 4.同步阻塞等待服務器返回應答，獲取應答後返回
21 input = new ObjectInputStream(socket.getInputStream());
22 return input.readObject();
23 } finally {
24 if (socket != null) socket.close();
25 if (output != null) output.close();
26 if (input != null) input.close();
27 }
28 }
遵循下面步驟
1：建立socket，並和遠程進行三次鏈接握手【socket.connect(addr)】。
2：封裝socket輸出流【ObjectOutputStream】。
3：輸出類名稱，方法名稱，參數類型和參數值給server。
4：獲取socket輸入流，等待server返回結果。
服務端代碼簡單講解
1 public RpcServer(int port) {
2 this.port =port;
3 }
4
5 public RpcServer register(Class serviceInterface, Class impl) {
6 serviceRegistry.put(serviceInterface.getName(), impl);
7 return this;
8 }
初始化服務端，將服務類註冊到hashMap【模擬spring上下文】
1 public void run() throws IOException {
2
3 ServerSocket server = new ServerSocket();
4 server.bind(new InetSocketAddress (port));
5 System.out.println("start server");
6 ObjectInputStream input =null;
7 ObjectOutputStream output =null;
8 Socket socket=null;
9 try {
10 while(true){
11 socket = server.accept ();
12 input =new ObjectInputStream(socket.getInputStream());
13
14 String serviceName = input.readUTF();
15 String methodName = input.readUTF();
16 System.out.println (methodName);
17 Class<?>[] parameterTypes = (Class<?>[]) input.readObject();
18 Object[] arguments = (Object[]) input.readObject();
19 Class serviceClass = serviceRegistry.get(serviceName);
20 if (serviceClass == null) {
21 throw new ClassNotFoundException(serviceName + " not found");
22 }
23 Method method = serviceClass.getMethod(methodName, parameterTypes);
24 Object result = method.invoke(serviceClass.newInstance(), arguments);
25 output = new ObjectOutputStream (socket.getOutputStream());
26 output.writeObject(result);
27 }
28 } catch (Exception e){
29 e.printStackTrace();
30 }finally {
31 if (output != null) {
32 try {
33 output.close();
34 } catch (IOException e) {
35 e.printStackTrace();
36 }
37 }
38 if (input != null) {
39 try {
40 input.close();
41 } catch (IOException e) {
42 e.printStackTrace();
43 }
44 }
45 if (socket != null) {
46 try {
47 socket.close();
48 } catch (IOException e) {
49 e.printStackTrace();
50 }
51 }
52 }
53
54 }
服務端執行作了如下幾件事：
1：綁定端口，阻塞等待客戶端調用【socket = server.accept ()】。
2：封裝輸入流【socket.getInputStream()】。
3：從輸入流中獲取到接口名，方法名，參數類型，參數值。
4：找到初始化時hashmap中的服務類。
5：反射獲取服務實現類方法並根據請求參數進行服務調用。
6：封裝輸出流【ObjectOutputStream】，而且返回結果。
到目前爲止，整個簡單的socket實現的RPC服務就已經所有完成了，能夠優化的部分。
1：序列化侷限，原生序列化只能序列化實現了【Serializable】接口的服務類，而且序列化複雜對象時，內容龐大效率極低，須要高效的序列化協議進行序列化參數方法等必要請求入參
2：BIO性能侷限，socket服務端採用默認的BIO來阻塞獲取輸入流，效率低下，需採用NIO等異步非阻塞服務端方案，例如netty,mina和java nio等。
3：在大型企業級RPC解決方案中，客戶端和服務端的長鏈接須要一直保持，不然每次調用時都要從新進行三次握手和四次揮手，這樣頻繁的建立tcp鏈接對機器性能是極大的損耗，對socket的鏈接能夠採用apache pool2鏈接池等方案
4：服務端負載，須要考慮服務自動發現，讓客戶端在不須要重啓的狀況下能動態感知服務端的變化，從而實現熱部署等。能夠採用辦法定時自動輪詢，zookeeper等。
5：服務端服務類執行異常，客戶端感知等。
好了，相信看完本章內容對於rpc框架來講，你們已經將基礎瞭解的差很少了，下面我將會給你們全面講解基於zk，thrift，netty的企業級RPC解決方案