1.1 RPC 是什么
定義：RPC（Remote Procedure Call Protocol）——遠程過程調用協議 ，RPC協議假定某些傳輸協議的存在，如TCP或UDP，為通信程序之間攜帶信息數據。在OSI網絡通信模型中，RPC跨越了傳輸層和應用層 ，RPC使得開發包括網絡分布式多程序在內的應用程序更加容易。
我的理解：與其說把RPC 看作是一種協議，倒不如把 它看作是一種 客戶機/服務器交互的模式，但是 RPC一定是基於 TCP 或者 其他 通信協議的
下面我們來看一下一個RPC調用的流程涉及哪些通信細節：
- 服務消費方（client）調用以本地調用方式調用服務；(1)
- client stub接收到調用后負責將方法、參數等組裝成能夠進行網絡傳輸的消息體；(2)
- client stub找到服務地址，並將消息發送到服務端；(3)
- server stub收到消息后進行解碼；(4)
- server stub根據解碼結果調用本地的服務；(5)
- 本地服務執行並將結果返回給server stub；(6)
- server stub將返回結果打包成消息並發送至消費方；(7)
- client stub接收到消息，並進行解碼；(8)
- 服務消費方得到最終結果。(9)
RPC的目標就是要2~8這些步驟都封裝起來，讓用戶對這些細節透明。
1.2 手動實現
1.2.1 先做一個空接口實現序列化接口
public interface IRpcService extends Serializable{
}
1.2.2 做一個需要被遠程調用的接口 以及對應的接口實現類
public interface IHelloService extends IRpcService{
String sayHi(String name,String message);
}
public class HelloServiceImpl implements IHelloService{
private static final long serialVersionUID = 146468468464364698L;
@Override
public String sayHi(String name, String message) {
return new StringBuilder().append("hi~!").append(",").append(message).toString();
}
}
1.2.3 需要寫一個服務端，主要的作用 是進行服務注冊(接口注冊) 以及 接收客戶端的調用參數 執行調用請求 返回結果
注：這個地方 我沒有采用dom4j 解析配置文件的形式 進行接口注冊 有時間的朋友可以多加一層
public interface Server {
//Socket端口
int PORT = 8080;
//啟動服務端
void start() throws IOException;
//停止服務端
void stop();
/**
* 服務注冊
* -- serviceInterface 對外暴露接口
* -- 內部實現類
*/
void regist(Class<? extends IRpcService> serviceInterface,Class<? extends IRpcService> impl);
}
public class ServerCenter implements Server{
/**線程池 接收客戶端調用**/
private static ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 20, 200, TimeUnit.MILLISECONDS,new ArrayBlockingQueue<Runnable>(10));
/**服務注冊緩存**/
public static final Map<String,Class<?>> serviceRegistry = new HashMap<>();
/**
* 啟動服務
*/
@Override
public void start() throws IOException {
ServerSocket server = new ServerSocket();
server.bind(new InetSocketAddress(PORT));
try {
while(true){
executor.execute(new ServiceTask(server.accept()));
}
} finally {
server.close();
}
}
/**
* 停止服務
*/
@Override
public void stop() {
executor.shutdown();
}
/**
* 注冊服務
*/
@Override
public void regist(Class<? extends IRpcService> serviceInterface, Class<? extends IRpcService> impl) {
serviceRegistry.put(serviceInterface.getName(), impl);
}
private static class ServiceTask implements Runnable{
Socket client = null;
public ServiceTask(Socket client) {
this.client = client;
}
@Override
public void run() {
ObjectInputStream input = null;
ObjectOutputStream output = null;
try {
input = new ObjectInputStream(client.getInputStream());
String serviceName = input.readUTF();
String methodName = input.readUTF();
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);
//將執行結果反序列化 通過socket返給客戶端
output = new ObjectOutputStream(client.getOutputStream());
output.writeObject(result);
} catch (Exception e) {
e.printStackTrace();
} finally {
if(input != null){
try {
input.close();
} catch (IOException e) {
e.printStackTrace();
}
}
if(output != null){
try {
output.close();
} catch (IOException e) {
e.printStackTrace();
}
}
if(client != null){
try {
client.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
}
public static void main(String[] args) throws Exception {
ServerCenter center = new ServerCenter();
center.regist(IHelloService.class,new HelloServiceImpl().getClass());
center.start();
}
}
1.2.4 寫一個客戶端，用動態代理 獲取被代理接口的 各種參數 傳輸給 服務端，接收返回結果，打印到控制台
public class Client {
@SuppressWarnings("unchecked")
public static <T extends IRpcService>T getRemoteProxyObj(final Class<? extends IRpcService> serviceInterface,final InetSocketAddress addr){
return (T) Proxy.newProxyInstance(serviceInterface.getClassLoader(), new Class<?>[]{serviceInterface}, new InvocationHandler() {
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
Socket socket = null;
ObjectOutputStream output = null;
ObjectInputStream input = null;
try {
//1.創建Socket客戶端，根據指定地址連接遠程服務提供者
socket = new Socket();
socket.connect(addr);
//2.將遠程服務調用所需的接口類、方法名、參數列表等編碼后發送給服務提供者
output = new ObjectOutputStream(socket.getOutputStream());
output.writeUTF(serviceInterface.getName());
output.writeUTF(method.getName());
output.writeObject(method.getParameterTypes());
output.writeObject(args);
//3.同步阻塞等待服務器返回應答 獲取應答后返回
input = new ObjectInputStream(socket.getInputStream());
return input.readObject();
} finally{
if(socket != null){
socket.close();
}
if(output != null){
output.close();
}
if(input != null){
input.close();
}
}
}
});
}
}
1.2.5 測試
注：測試之前 需要開啟服務端
public class RpcTest {
public static void main(String[] args) throws IOException {
IHelloService service = Client.getRemoteProxyObj(IHelloService.class, new InetSocketAddress(8080));
System.out.println(service.sayHi("張三", "新年快樂!"));
}
}
就這樣我們實現了一個簡陋的RPC
本文意在通過實現簡單的RPC，去真正意義上對RPC框架的實現原理有初步的了解，而不是人雲亦雲。
此RPC實現有諸多缺點，但是 我們只要明白RPC的基座 其他的RPC框架只是完善基座以及擴展而已 。