如何手搓一個自定義的RPC（遠程過程調用框架）
1.RPC（遠程過程調用概述）
遠程過程調用（RPC, Remote Procedure Call）是一種通過網絡從遠程計算機程序上請求服務，而無需了解網絡細節的通信技術。在分布式系統中，RPC是一種常用的技術，能夠簡化客戶端與服務器之間的交互。本文將介紹如何基于Netty（網絡編程框架）實現一個自定義的簡單的RPC框架。
首先簡單介紹一下RPC 主要特點：
1.1 RPC遠程過程調用的主要特點
- 透明性：調用方（客戶端）調用遠程服務就像調用本地API函數一樣，而無需關心執行過程中的底層的網絡通信細節。
- 客戶端-服務器模型：RPC通常基于客戶端-服務器模型，客戶端發送請求到服務器，服務器處理請求并返回結果。
- 序列化及反序列化：RPC需要將請求參數序列化成字節流（即數據轉換成網絡可傳輸的格式）并通過網絡傳輸到服務器端，服務器端接收到字節流后，需按照約定的協議將數據進行反序列化（即恢復成三原始格式）
- 同步及異步調用：RPC支持同步、異步調用。同步調用會阻塞直到服務器返回結果，或超時、異常等。而異步調用則可以立即返回，通過注冊一個回調函數，在有結果返回的時候再進行處理。從而讓客戶端可以繼續執行其它操作。
- 錯誤處理：PRC由于涉及網絡通信，因此需要處理各種可能的網絡異常，如網絡故障，服務宕機，請求超時，服務重啟、或上下線、擴縮容等，這些對調用方來說需要保持透明。
- 協議及傳輸居：RPC可以基于多種協議和傳輸層實現，如HTTP、TCP等，本文采用的是基于TCP的自定義協議。
1.2 RPC的應用場景
- 分布式系統：多個服務之間進行通信，如微服務框架。
- 客戶端-服務器架構：如移動應用與后臺服務器的交互。
- 跨平臺調用：不同技術棧之間的服務調用。
- API服務：通過公開API對外提供功能，使用客戶端能方便使用服務提供的功能，如支付網關，身份驗證服務等。
- 大數據處理：在大數據處理框架中，不同節點之間需要頻繁通信來協調任務和交接數據，RPC可以提供高效的節點通信機制，如Hadoop 和Spark等大數據框架中節點間的通信。
- 云計算：在云計算環境中，服務通常分布在多個虛擬機或容器中，通過RPC實現實現服務間的通信和管理。
- 跨網絡服務調用：當應用需要調用部署在不同網絡中的服務時，RPC提供了一種簡單而建議目前的調用方式，如。跨數據中心或嘴唇地域的服務調用。
1.3 常見的RPC框架
- JSF：京東開源的分布式服務框架，提供高性能、可擴展、穩定的服務治理能力，支持服務注冊及發現，負載均衡、容錯機制、服務監控、多種協議支持等。
- gRPC：基于HTTP/2和Protocol Buffers的高性能RPC框架，由Google開發。
- Dubbo：一個高性能、輕量級的Java RPC框架，用于提供基于接口的遠程服務調用，支持負載均衡、服務自動注冊及服務、容錯等。
- JSON-RPC：使用JSON格式編碼調用和結果的RPC協議。
- Apache Thrift：由Facebook開發，支持多種編程語言和協議
2.實現自定義的RPC
理解，首先 MCube 會依據模板緩存狀態判斷是否需要網絡獲取最新模板，當獲取到模板后進行模板加載，加載階段會將產物轉換為視圖樹的結構，轉換完成后將通過表達式引擎解析表達式并取得正確的值，通過事件解析引擎解析用戶自定義事件并完成事件的綁定，完成解析賦值以及事件綁定后進行視圖的渲染，最終將
要實現一個自定義的RPC框架需解決以下幾個主要問題：
1.客戶端調用：客戶端調用本地的代理函數（stub代碼，這個函數負責將調用轉換為RPC請求）。這其實就是一個接口描述文件，它可以有多種形式如JSON、XML、甚至是一份word文檔或是口頭約定均可，只要客戶端及服務端都是遵守這份接口描述文件契約即可。在我們的實際開發中一種常見的方式是服務提供者發布一個包含服務接口類的jar包到maven 中央倉庫，調用方通過pom文件將之依賴到本地。
2.參數序列化：代理函數將調用參數進行序列化，并將請求發送到服務器。
3.服務端數據接收：服務器端接收到請求，并將其反序列化，恢復成原始參數。
4.執行遠程過程：服務端調用實際的服務過程（函數）并獲取結果。
5.返回結果：服務端將調用結果進行序列化，并通過網絡傳給客戶端。
6.客戶端接收調用結果：客戶到接收到服務端傳輸的字節流，進行反序列化，轉換為實際的結果數據格式，并返回到原始調用方。
下面需我們通過代碼一一展示上述各功能是如何實現的。
2.1 自定義通信協議
本文的目的是要實現一個自定義通信協議的遠程調用框架，所以首先要定義一個通信協議數據格式。
整個自定義協議總體上分為Header 及 Body Content兩部分；Header 占16個字節，又分為4個部分。
前2位為魔法值用于Netty編解碼組件，解決網絡通信中的粘包、半包等問題，此處不展開細講。
msgtype用于表示消息的類型，如request（請求）、respone（響應）、heartbeat（心跳）等。
code 占1位，表示請求的響應狀態，成功還是失敗。
request id占8位，表示請求的序列號，用于后續調用結果的匹配，保證線程內唯一。
body size 占4位，表示實現請求內容的長度，在反序化時讀取此長度的內容字節，解析出正確的數據。
客戶端、服務端在通信過程中都要按照上述約定的通信協議進行數據的編、解碼工作。
2.2 客戶端調用
2.2.1 客戶端的使用
客戶端一般通過接口代理工廠通過動態代理技術來生成一個代理實例，所有的遠程調用中的細節，如參數序列化，網絡傳輸，異常處理等都隱藏在代理實例中實現，對調用方來說調用過程是透明的，就像調用本地方法一樣。
首先看一下客戶端的使用方式，本文假設一個IShoppingCartService （購物車）的接口類，基中有一個方法根據傳入的用戶pin，返回購物車詳情。
//接口方法
ShoppingCart shopping(String pin);//客戶端通過代理工廠實現接口的一個代理實例
IShoppingCartService serviceProxy = ProxyFactory.factory(IShoppingCartService.class)
.setSerializerType(SerializerType.JDK) //客戶端設置所使用的序列化工具，此處為JDK原生
.newProxyInstance(); //返回代理 實現//像調用本地方法一樣，調用此代理實例的shopping 方法
ShoppingCart result = serviceProxy.shopping("userPin");
log.info("result={}", JSONObject.toJSONString(result));2.2.2 客戶端代理工廠的核心功能
public class ProxyFactory<I> {
//……省略
/**
* 代理對象
*
* @return
*/
public I newProxyInstance() {
//服務的元數據信息
ServiceData serviceData = new ServiceData(
group, //分組
providerName, //服務名稱，一般為接口的class的全限定名稱
StringUtils.isNotBlank(version) ? version : "1.0.0" //版本號
);
//調用器
Calller caller = newCaller().timeoutMillis(timeoutMillis);
//集群策略，用于實現快速失敗或失敗轉等功能
Strategy strategy = StrategyConfigContext.of(strategy, retries);
Object handler = null;
switch (invokeType) {
case "syncCall":
//同步調用handler
handler = new SyncCaller(serviceData, caller);
break;
case "asyncCall":
//異步調用handler
handler = new AsyncCaller(client.appName(), serviceData, caller, strategy);
break;
default:
throw new RuntimeException("未知類型: " + invokeType);
}
//返回代理實例
return ProxyEnum.getDefault().newProxy(interfaceClass, handler);
}
//……省略
}代碼 ProxyEnum.getDefault().newProxy(interfaceClass, handler) 返回一個具體的代理實例，此方法要求傳入兩個參數，interfaceClass 被代理的接口類class，即服務方所發布的服務接口類。
handler 為動態代理所需要代碼增強邏輯，即所有的調用細節都由此增強類完成。按照動態代理的實現方式的不同，本文支持兩種動態代理方式：
1.JDK動態代碼，如采用此方式，handler 需要實現接口 InvocationHandler
2.ByteBuddy，它是一個用于在運行時生成、修改和操作Java類的庫，允許開發者通過簡單的API生成新的類或修改已有的類，而無需手動編寫字節碼，它廣泛應用于框架開發、動態代理、字節碼操作和類加載等領域。
本文默認采用第二種方式，通過代碼簡單展示一下代理實例的的生成方式。
//方法newProxy 的具體實現
public <T> T newProxy(Class<T> interfaceType, Object handler) {
Class<? extends T> cls = new ByteBuddy()
//生成接口的子類
.subclass(interfaceType)
//默認代理接口中所有聲明的方法
.method(ElementMatchers.isDeclaredBy(interfaceType))
//代碼增強，即接口中所有被代理的方法都
//委托給用戶自定義的handler處理，這也是動態代理的意義所在
.intercept(MethodDelegation.to(handler, "handlerInstance"))
.make()
//通過類加載器加載
.load(interfaceType.getClassLoader(), ClassLoadingStrategy.Default.INJECTION)
.getLoaded();
try {
//通過newInstance構建一個代理實例并返回
return cls.newInstance();
} catch (Throwable t) {
……
}
}本文以同步調用為例，現在展示一下 SyncInvoker 的具體實現邏輯。
public class SyncCaller extends AbstractCaller {
//……省略
/**
* @RuntimeType 的作用提示ByteBuddy根據被攔截方法的實際類型，對此攔截器的返回值進行類型轉換
*/
@RuntimeType
public Object syncCall(@Origin Method method, @AllArguments @RuntimeType Object[] args) throws Throwable {
//封裝請求的接口中的方法名及方法參數，組成一個request請求對象
StarGateRequest request = createRequest(methodName, args);
//集群容錯策略調度器接口
//提供快速失敗，失敗轉移等策略供調用方選擇，此處默認采用了快速失敗的策略
Invoker invoker = new FastFailInvoker();
//returnType 的類型決定了泛型方法的實際結果類型，用于后續調用結果的類型轉換
Future<?> future = invoker.invoke(request, method.getReturnType());
if (sync) {
//同步調用，線程會阻塞在get方法，直到超時或結果可用
Object result = future.getResult();
return result;
} else {
return future;
}
}
}
//同步，異步調用的關鍵點就在于InvokeFuture，它繼承了Java的CompletionStage類，用于異步編程通過以上核心代碼，客戶端就完成了服務調用環節，下一步RPC框架需要將客戶端請求的接口方法及方法參數進行序列化并通過網絡進行傳輸。下面通過代碼片段展示一下序列化的實現方式。
2.2.3 請求參數序列化
我們將請求參數序列化的目的就是將具體的請求參數轉換成字節組，填充進入上述自定義協議的 body content 部分。下面通過代碼演示一下如何進行反序列化。
本文默認采用JDK原生的對象序列化及反序列化框架，也可通過SPI技術擴展支持Protocol Buffers等。
//上述代碼行Future<?> future = invoker.invoke(request, method.getReturnType());
//具體實現
public <T> Future<T> invoke(StarGateRequest request, Class<T> returnType) throws Exception {
//對象序列化器，默認為JDK
final Serializer _serializer = serializer();
//message對象包含此次請求的接口名，方法名及實際參數列表
final Message message = request.message();
//通過軟負載均衡選擇一個 Netty channel
Channel channel = selectChannel(message.getMetadata());
byte code = _serializer.code();
//將message對象序列成字節數組
byte[] bytes = _serializer.writeObject(message);
request.bytes(code, bytes);
//數據寫入 channel 并返回 future 約定，用于同步或異步獲得調用結果
return write(channel, request, returnType);
}//對象的序列化，JDK 原生方式
public <T> byte[] writeObject(T obj) {
ByteArrayOutputStream buf = OutputStreams.getByteArrayOutputStream();
try (ObjectOutputStream output = new ObjectOutputStream(buf)) {
output.writeObject(obj);
output.flush();
return buf.toByteArray();
} catch (IOException e) {
ThrowUtil.throwException(e);
} finally {
OutputStreams.resetBuf(buf);
}
return null;
}2.2.4 請求參數通過網絡發送
//上述代碼 write(channel, request, returnType);
//具體實現
protected <T> DefaultFuture<T> write(final Channel channel,
final StarGateRequest request,
final Class<T> returnType) {
//……省略
//調用結果占位 future對象，這也是promise編程模式
final Future<T> future = DefaultFuture.newFuture(request.invokeId(), channel, timeoutMillis, returnType);
//將請求負載對象寫入Netty channel通道，并綁定監聽器處理寫入結果
channel.writeAndFlush(request).addListener((ChannelFutureListener) listener -> {
if (listener.isSuccess()) {
//網絡寫入成功
……
} else {
//異常時，構造造調用結果，供調用方進行處理
DefaultFuture.errorFuture(channel, response, dispatchType);
}
});
//因為Netty 是非阻塞的，所以寫入后可立刻返回
return future;
}2.2.4.1 Netty 消息編碼器
消息寫入Netty channel 后，會依次經過 channel pipline 上所安裝的各種handler處理，然后再通過物理網絡將數據發送出去，這里展示了客戶端及服務端所使用的自定義編、解解器。
//自定義的編碼器 繼承自Netty 的 MessageToByteEncoder
public class StarGateEncoder extends MessageToByteEncoder<Payload> {
//……省略
private void doEncodeRequest(RequestPayload request, ByteBuf out) {
byte sign = StarGateProtocolHeader.toSign(request.serializerCode(), StarGateProtocolHeader.REQUEST);
long invokeId = request.invokeId();
byte[] bytes = request.bytes();
int length = bytes.length;
out.writeShort(StarGateProtocolHeader.Head) //寫入兩個字節
.writeByte(sign) //寫入1個字節
.writeByte(0x00) //寫入1個字節
.writeLong(invokeId) //寫入8個節節
.writeInt(length) //寫入4個字節
.writeBytes(bytes);
}
}至此，通過上述核心代碼，客戶的請求已經按照自定義的協議格式進行了序列化，并把數據寫入到Netty channel中，最后通過物理網絡傳輸到服務器端。
2.3 服務端接收數據
2.3.1 消息解碼器
服務器端接收到客戶端的發送的數據后，需要進行正確的消息解碼，下面是解碼器的實現。
//消息解碼器，繼承自Netty 的ReplayingDecoder，將客戶端請求解碼為 RequestPayload 對象，供業務處理handler使用
public class StarGateDecoder extends ReplayingDecoder<StarGateDecoder.State> {
//……省略
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
switch (state()) {
case HEAD:
checkMagic(in.readShort()); // HEAD
checkpoint(State.HEAD);
case SIGN:
header.sign(in.readByte()); // 消息標志位
checkpoint(State.STATUS);
case STATUS:
header.status(in.readByte()); // 狀態位
checkpoint(State.ID);
case ID:
header.id(in.readLong()); // 消息id
checkpoint(State.BODY_SIZE);
case BODY_SIZE:
header.bodySize(in.readInt()); // 消息體長度
checkpoint(State.BODY);
case BODY:
switch (header.messageCode()) {
//……省略
case StarGateProtocolHeader.REQUEST: {
//消息體長度信息
int length = checkBodySize(header.bodySize());
byte[] bytes = new byte[length];
//讀取指定長度字節
in.readBytes(bytes);
//調用請求
RequestPayload request = new RequestPayload(header.id());
//設置序列化器編碼，有效載荷
request.bytes(header.serializerCode(), bytes);
out.add(request);
break;
}
default:
throw new Exception("錯誤標志位");
}
checkpoint(State.HEAD);
}
}
//……省略
}2.3.2 請求參數反序列化
//服務端 Netty channel pipline 上所安裝的業務處理 handler
//業務處理handler 對RequestPayload 所攜帶的字節數組進行反序列化，解析出客戶端所傳遞的實際參數
public class ServiceHandler extends ChannelInboundHandlerAdapter {
//……省略
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
Channel ch = ctx.channel();
if (msg instanceof RequestPayload) {
StarGateRequest request = new StarGateRequest((RequestPayload) msg);
//約定的反序列化器, 由客戶端設置
byte code = request.serializerCode();
Serializer serializer = SerializerFactory.getSerializer(code);
//實際請求參數字組數組
byte[] bytes = payload.bytes();
//對象反序列化
Message message = serializer.readObject(bytes, Message.class);
log.info("message={}", JSONObject.toJSONString(message));
request.message(message);
//業務處理
process(message);
} else {
//引用釋放
ReferenceCountUtil.release(msg);
}
}
//……省略
}2.3.3 處理客戶端請求
經過反序列化后，服務端可以知道用戶所請求的是哪個接口、方法、以及實際的參數值，下一步就可進行真實的方法調用。
//處理調用
public void process(Message message) {
try {
ServiceMetadata metadata = msg.getMetadata(); //客戶端請求的元數據
String providerName = metadata.getProviderName(); //服務名，即接口類名
//根據接口類名，查找服務端實現此接口的類的全限定類名
providerName = findServiceImpl(providerName);
String methodName = msg.getMethodName(); //方法名
Object[] args = msg.getArgs(); //客戶設置的實際參數
//線程上下文類加載器
ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
//加載具體實現類
Class<?> clazz = classLoader.loadClass(providerName);
//創建接口類實例
Object instance = clazz.getDeclaredConstructor().newInstance();
Method method = null;
Class<?>[] parameterTypes = new Class[args.length];
for (int i = 0; i < args.length; i++) {
parameterTypes[i] = args[i].getClass();
}
method = clazz.getMethod(methodName, parameterTypes);
//反射調用
Object invokeResult = method.invoke(instance, args);
} catch (Exception e) {
log.error("調用異常：", e);
throw new RuntimeException(e);
}
//處理同步調用結果
doProcess(invokeResult);
}2.3.4 返回調用結果
通過反射調用接口實現類，獲取調用結果，然后對結果進行序列化并包裝成response響應消息，將消息寫入到channel, 經過channel pipline 上所安裝的編碼器對消息對象進行編碼，最后發送給調用客戶端。
//處理同步調用結果,并將結果寫回到 Netty channel
private void doProcess(Object realResult) {
ResultWrapper result = new ResultWrapper();
result.setResult(realResult);
byte code = request.serializerCode();
Serializer serializer = SerializerFactory.getSerializer(code);
//new response 響應消息對象
Response response = new Response(request.invokeId());
//調用結果序列成字節數組
byte[] bytes = serializer.writeObject(result);
response.bytes(code, bytes);
response.status(Status.OK.value());
//響應消息對象 response 寫入 Netty channel
channel.writeAndFlush(response).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture channelFuture) throws Exception {
if (channelFuture.isSuccess()) {
log.info("響應成功");
} else {
//記錄調用失敗日志
log.error("響應失敗, channel: {}, cause: {}.", channel, channelFuture.cause());
}
}
});
}同樣的，消息寫入channel 后，先依次經過pipline 上所安裝的 消息編碼器，再發送給客戶端。具體編碼方式同客戶端編碼器類似，此處不再贅述。
2.4 客戶端接收調用結果
客戶端收到服務端寫入響應消息后，同樣經過Netty channel pipline 上所安裝的解碼器，進行正確的解碼。然后再對解碼后的對象進行正確的反序列化，最終獲得調用結果 。具體的解碼，反序列化過程不再贅述，流程基本同上面服務端的解碼及反序列化類似。
public class consumerHandler extends ChannelInboundHandlerAdapter {
//……省略
//客戶端處理所接收到的消息
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
Channel ch = ctx.channel();
if (msg instanceof ResponseMessage) {
try {
//類型轉換
ResponseMessage responseMessage= (ResponseMessage)msg
StarGateResponse response = new StarGateResponse(ResponseMessage.getMsg());
byte code = response.serializerCode();
Serializer serializer = SerializerFactory.getSerializer(code);
byte[] bytes = responseMessage.bytes();
//反序列化成調用結果的包裝類
Result result = serializer.readObject(bytes, Result.class);
response.result(result);
//處理調用結果
long invokeId = response.id();
//通過 rnvokeid,從地緩存中拿到客戶端調用的結果點位對象 futrue
DefaultFuture<?> future = FUTURES_MAP.remove(invokeId);
//判斷調用是否成功
byte status = response.status();
if (status == Status.OK.value()) {
//對調用結果進行強制類型轉換，并設置future結果，對阻塞在future.get()的客戶端同步調用來說，調用返回。
complete((V) response.getResult());
} else {
//todo 處理異常
}
} catch (Throwable t) {
log.error("調用記錄: {}, on {} #channelRead().", t, ch);
}
} else {
log.warn("消息類型不匹配: {}, channel: {}.", msg.getClass(), ch);
//計數器減1
ReferenceCountUtil.release(msg);
}
}
}下面再通過一個簡單的調用時序圖展示一下一次典型的Rpc調用所經歷的步驟。
3.結尾
本文首先簡單介紹了一下RPC的概念、應用場景及常用的RPC框架，然后講述了一下如何自己手動實現一個RPC框架的基本功能。目的是想讓大家對RPC框架的實現有一個大概思路，并對Netty 這一高效網絡編程框架有一個了解，通過對Netty 的編、解碼器的學習，了解如何自定義一個私有的通信協議。限于篇幅本文只簡單講解了RPC的核心的調用邏輯的實現。真正生產可用的RPC框架還需要有更多復雜的功能，如限流、負載均衡、融斷、降級、泛型調用、自動重連、自定義可擴展的攔截器等等。
另外RPC框架中一般有三種角色，服務提供者、服務消費者、注冊中心，本文并沒有介紹注冊中心如何實現。并假定服務提供者已經將服務發布到了注冊中心，服務消費者跟服務提供者之間建立起了TCP 長連接。
后續會通過其它篇章介紹注冊中心，服務自動注冊，服務發現等功能的實現原理。
