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