五層協定中，RPC在第幾層?
| 五層協定 |
|---|
| 應用層 |
| 傳輸層 |
| 網路層 |
| 鏈路層 |
| 物理層 |
遠端過程呼叫（RPC），比較樸素的說法就是，從某臺機器呼叫另一臺機器的一段程式碼，並獲取返回結果。
這之前的一個基層問題就是程序間通訊方式（IPC），從是否設計網路通訊分為：
和共用記憶體不同，Socket實現不併不是隻依靠記憶體屏障，它還額外需要物理/虛擬網路卡裝置。
關於網路卡，只需要知道網路卡可以幫助我們從網路中讀寫資訊，這也是RPC的基礎。
遠端過程呼叫，不如先來研究呼叫。
先來一段普通的程式碼。
public class EchoService {
public static EchoResponse echo(EchoRequest req) throws Exception {
return new EchoResponse("echo:" + req.content);
}
public static void main(String[] args) throws Exception {
System.out.println(EchoService.echo(new EchoRequest("ping")).content); // echo:ping
}
}
class EchoRequest {
String content;
public EchoRequest(String content) {
this.content = content;
}
}
class EchoResponse {
String content;
public EchoResponse() {
}
public EchoResponse(String content) {
this.content = content;
}
}
回聲服務對傳入引數直接返回，就像你在山谷中的回聲一樣。
現在如果使用遠端傳輸，我們需要給網路卡註冊自己的IP和埠，以便和伺服器端建立連線。連線建立後，我們還需要確定資料如何傳輸。
為了樸素性，我們假設只有10臺機器和我們進行連線。
public Runnable apply(Integer port) {
return () -> {
try {
try (ServerSocket serverSocket = new ServerSocket(port)) {
for (;;) {
Socket clientSocket = serverSocket.accept();
new Thread(() -> {
// 資料如何傳輸
}).start();
}
}
} catch (Exception e) {
e.printStackTrace();
}
};
}
根據Socket的檔案，我們可以很快迭代出一臺伺服器應該如何與他的使用者端連線。對於每個使用者端，我們提供了獨立的執行緒支援兩臺機器間的長連線。
試想一下，此時的長連線如果是百萬甚至千萬，為每個連線分配一個執行緒不可取，有什麼好辦法可以支援到呢？這個問題這裡不解了，有興趣自行研究下。
一說起序列化，最怕異口同聲json。
使用json就難免會使用到 第三方庫，如果沒有必要，並不希望引入。除了json外，java其實本身就有Serializable實現，他和synchronized一樣，java官方提供並維護。
public class EchoService {
public static EchoResponse echo(EchoRequest req) throws Exception {
throw new UnsupportedOperationException();
}
}
class EchoRequest implements Serializable {
String content;
public EchoRequest(String content) {
this.content = content;
}
}
class EchoResponse implements Serializable {
String content;
public EchoResponse() {
}
public EchoResponse(String content) {
this.content = content;
}
}
除了引數外，一個rpc需要知道，ip、埠、服務名、方法名。
ip和埠在呼叫時應該已經知道，為此還需要支援一個header來完成服務名和方法名的指定。
class Header implements Serializable {
String stub;
String method;
public Header(String stub, String method) {
this.stub = stub;
this.method = method;
}
}
通過編碼解碼器對Serializable的資料編碼和解碼。
public class Codec {
Socket clientSocket;
ObjectInputStream objectInputStream;
ObjectOutputStream objectOutputStream;
public Codec(Socket clientSocket)
throws Exception {
this.clientSocket = clientSocket;
this.objectOutputStream = new ObjectOutputStream(clientSocket.getOutputStream());
this.objectInputStream = new ObjectInputStream(clientSocket.getInputStream());
}
public Header header() throws Exception {
return (Header) this.objectInputStream.readObject();
}
public Object read() throws Exception {
return this.objectInputStream.readObject();
}
public void write(Header header, Object obj) throws Exception {
this.objectOutputStream.writeObject(header);
this.objectOutputStream.writeObject(obj);
}
}
回到伺服器端，將空缺的地方通過反射補全。
Codec codec = new Codec(clientSocket);
for (;;) {
Header header = codec.header();
Class<?> stub = Class.forName(header.stub);
Map<String, Method> methods = Arrays.asList(stub.getDeclaredMethods()).stream()
.collect(Collectors.toMap(t -> t.getName(), t -> t));
Method method = methods.get(header.method);
codec.write(header, method.invoke(null, header, codec.read()));
}
通過codec解碼stub和method來找到對應的方法，呼叫對應方法，獲取結果後再通過編碼返回使用者端。
想一下，如果一個使用者端傳送了10個請求，其中第2個由於種種原因被阻塞掉，後面的請求會被卡在阻塞的請求之後而無法獲得響應。
簡單的處理方法，就是抽象掉呼叫過程，並給其唯一標識。需要一個map來存全部的呼叫請求。
class Call {
Long seq;
Object req;
Object rsp;
Thread thread;
public Call(Long seq, Object req) {
this.seq = seq;
this.req = req;
}
}
對call抽象後，對client也就迎刃而解了。
我知道了，map，用map解。
Long seq;
Codec codec;
ReentrantLock clock;
Map<Long, Call> calls;
ReentrantLock metux;
在map之上提供對seq的操作。
Call register(Call call) {
try {
clock.lock();
call.seq = seq;
calls.put(seq, call);
seq++;
return call;
} finally {
clock.unlock();
}
}
Call remove(Call call) {
try {
clock.lock();
call.seq = seq;
calls.remove(seq);
return call;
} finally {
clock.unlock();
}
}
對伺服器端的響應監聽，喚醒阻塞的執行緒。
void receive() throws Exception {
for (;;) {
Header header = codec.header();
Call call = calls.remove(header.seq);
Object rsp = codec.read();
call.rsp = rsp;
LockSupport.unpark(call.thread);
}
}
最後就是發起使用者端呼叫的程式碼。
FutureTask<Object> start(Header header, Object req) throws Exception {
Call call = new Call(seq, req);
try {
metux.lock();
final Call fcall = register(call);
header.seq = call.seq;
codec.write(header, req);
FutureTask<Object> task = new FutureTask<>(() -> {
fcall.thread = Thread.currentThread();
LockSupport.park();
return fcall.rsp;
});
task.run();
return task;
} finally {
metux.unlock();
}
}
public static void main(String[] args) throws UnknownHostException, IOException, Exception {
new Thread(new Server().apply(8080)).start(); // 伺服器端啟動
// 模擬呼叫
ExecutorService newFixedThreadPool = Executors.newFixedThreadPool(10);
Client client = new Client(new Codec(new Socket("127.0.0.1", 8080)));
for (int i = 0; i < 100; i++) {
newFixedThreadPool.submit(() -> {
try {
FutureTask<Object> call = client.start(
new Header("EchoService", "echo"),
new EchoRequest("~hello"));
EchoResponse rsp = (EchoResponse) call.get();
System.out.println(rsp.content);
} catch (Exception e) {
e.printStackTrace();
}
});
}
}
Output
RPC echo~hello 0
RPC echo~hello 1
RPC echo~hello 2
RPC echo~hello 3
RPC echo~hello 4
RPC echo~hello 6
RPC echo~hello 5
RPC echo~hello 7
RPC echo~hello 9
RPC echo~hello 8
至此，只是實現了rpc的通訊過程，完成度比較高。