100行代碼透徹解析RPC原理

本文主要論述的是“RPC 實(shí)現(xiàn)原理”，那么首先明確一個(gè)問(wèn)題什么是 RPC 呢？RPC 是 Remote Procedure Call 的縮寫(xiě)，即，遠(yuǎn)程過(guò)程調(diào)用。RPC 是一個(gè)計(jì)算機(jī)通信協(xié)議。該協(xié)議允許運(yùn)行于一臺(tái)計(jì)算機(jī)的程序調(diào)用另一臺(tái)計(jì)算機(jī)的子程序，而開(kāi)發(fā)人員無(wú)需額外地為這個(gè)交互編程。

值得注意是，兩個(gè)或多個(gè)應(yīng)用程序都分布在不同的服務(wù)器上，它們之間的調(diào)用都像是本地方法調(diào)用一樣。接下來(lái)我們便來(lái)分析一下一次 RPC 調(diào)用發(fā)生了些什么？

—?1?—

RPC 調(diào)用的基本流程

現(xiàn)在業(yè)界內(nèi)比較流行的一些 RPC 框架，例如 Dubbo 提供的是基于接口的遠(yuǎn)程方法調(diào)用，即客戶(hù)端只需要知道接口的定義即可調(diào)用遠(yuǎn)程服務(wù)。在 Java 中接口并不能直接調(diào)用實(shí)例方法，必須通過(guò)其實(shí)現(xiàn)類(lèi)對(duì)象來(lái)完成此操作，這意味著客戶(hù)端必須為這些接口生成代理對(duì)象，對(duì)此 Java 提供了 Proxy、InvocationHandler 生成動(dòng)態(tài)代理的支持；生成了代理對(duì)象，那么每個(gè)具體的發(fā)方法是怎么調(diào)用的呢？JDK 動(dòng)態(tài)代理生成的代理對(duì)象調(diào)用指定方法時(shí)實(shí)際會(huì)執(zhí)行 InvocationHandler 中定義的?#invoke 方法，在該方法中完成遠(yuǎn)程方法調(diào)用并獲取結(jié)果。

拋開(kāi)客戶(hù)端，回過(guò)頭來(lái)看 RPC 是兩臺(tái)計(jì)算機(jī)間的調(diào)用，實(shí)質(zhì)上是兩臺(tái)主機(jī)間的網(wǎng)絡(luò)通信，涉及到網(wǎng)絡(luò)通信又必然會(huì)有序列化、反序列化，編解碼等一些必須要考慮的問(wèn)題；同時(shí)實(shí)際上現(xiàn)在大多系統(tǒng)都是集群部署的，多臺(tái)主機(jī)/容器對(duì)外提供相同的服務(wù)，如果集群的節(jié)點(diǎn)數(shù)量很大的話，那么管理服務(wù)地址也將是一件十分繁瑣的事情，常見(jiàn)的做法是各個(gè)服務(wù)節(jié)點(diǎn)將自己的地址和提供的服務(wù)列表注冊(cè)到一個(gè)注冊(cè)中心，由注冊(cè)中心來(lái)統(tǒng)一管理服務(wù)列表；這樣的做法解決了一些問(wèn)題同時(shí)為客戶(hù)端增加了一項(xiàng)新的工作——那就是服務(wù)發(fā)現(xiàn)，通俗來(lái)說(shuō)就是從注冊(cè)中心中找到遠(yuǎn)程方法對(duì)應(yīng)的服務(wù)列表并通過(guò)某種策略從中選取一個(gè)服務(wù)地址來(lái)完成網(wǎng)絡(luò)通信。

聊了客戶(hù)端和注冊(cè)中心，另外一個(gè)重要的角色自然是服務(wù)端，服務(wù)端最重要的任務(wù)便是提供服務(wù)接口的真正實(shí)現(xiàn)并在某個(gè)端口上監(jiān)聽(tīng)網(wǎng)絡(luò)請(qǐng)求，監(jiān)聽(tīng)到請(qǐng)求后從網(wǎng)絡(luò)請(qǐng)求中獲取到對(duì)應(yīng)的參數(shù)（比如服務(wù)接口、方法、請(qǐng)求參數(shù)等），再根據(jù)這些參數(shù)通過(guò)反射的方式調(diào)用接口的真正實(shí)現(xiàn)獲取結(jié)果并將其寫(xiě)入對(duì)應(yīng)的響應(yīng)流中。

綜上所述，一次基本的 RPC 調(diào)用流程大致如下：

—?2?—

基本實(shí)現(xiàn)

服務(wù)端（生產(chǎn)者）

服務(wù)接口：

在 RPC 中，生產(chǎn)者和消費(fèi)者有一個(gè)共同的服務(wù)接口 API。如下，定義一個(gè) HelloService 接口。

/**
?*?@author?孫浩
?*?@Descrption??服務(wù)接口
?***/
public?interface?HelloService?{
????String?sayHello(String?somebody);
}

服務(wù)實(shí)現(xiàn)：

生產(chǎn)者要提供服務(wù)接口的實(shí)現(xiàn)，創(chuàng)建 HelloServiceImpl 實(shí)現(xiàn)類(lèi)。

/**
?*?@author?孫浩
?*?@Descrption?服務(wù)實(shí)現(xiàn)
?***/
public?class?HelloServiceImpl?implements?HelloService?{
????@Override
????public?String?sayHello(String?somebody)?{
????????return?"hello?"?+?somebody?+?"!";
????}
}

服務(wù)注冊(cè)：

本例使用 Spring 來(lái)管理 bean，采用自定義 XML 和解析器的方式來(lái)將服務(wù)實(shí)現(xiàn)類(lèi)載入容器（當(dāng)然也可以采用自定義注解的方式，此處不過(guò)多論述）并將服務(wù)接口信息注冊(cè)到注冊(cè)中心。

首先自定義 XSD：

"service">
????
????????
????????????"beans:identifiedType">
????????????????"interface"?type="xsd:string"?use="required"/>
????????????????"timeout"?type="xsd:int"?use="required"/>
????????????????"serverPort"?type="xsd:int"?use="required"/>
????????????????"ref"?type="xsd:string"?use="required"/>
????????????????"weight"?type="xsd:int"?use="optional"/>
????????????????"workerThreads"?type="xsd:int"?use="optional"/>
????????????????"appKey"?type="xsd:string"?use="required"/>
????????????????"groupName"?type="xsd:string"?use="optional"/>
????????????
????????
????

分別指定 Schema 和 XSD，Schema 和對(duì)應(yīng) Handler 的映射。

Schema：

http\://www.storm.com/schema/storm-service.xsd=META-INF/storm-service.xsd
http\://www.storm.com/schema/storm-reference.xsd=META-INF/storm-reference.xsd

Handler：

http\://www.storm.com/schema/storm-service=com.hsunfkqm.storm.framework.spring.StormServiceNamespaceHandler
http\://www.storm.com/schema/storm-reference=com.hsunfkqm.storm.framework.spring.StormRemoteReferenceNamespaceHandler

將編寫(xiě)好的文件放入 Classpath 下的 META-INF 目錄下：

在 Spring 配置文件中配置服務(wù)類(lèi)：

?"helloService"?class="com.hsunfkqm.storm.framework.test.HelloServiceImpl"/>
?"helloServiceRegister"
?????????????????????interface="com.hsunfkqm.storm.framework.test.HelloService"
?????????????????????ref="helloService"
?????????????????????groupName="default"
?????????????????????weight="2"
?????????????????????appKey="ares"
?????????????????????workerThreads="100"
?????????????????????serverPort="8081"
?????????????????????timeout="600"/>

編寫(xiě)對(duì)應(yīng)的 Handler 和 Parser：

StormServiceNamespaceHandler：

import?org.springframework.beans.factory.xml.NamespaceHandlerSupport;

/**
?*?@author?孫浩
?*?@Descrption?服務(wù)發(fā)布自定義標(biāo)簽
?***/
public?class?StormServiceNamespaceHandler?extends?NamespaceHandlerSupport?{
????@Override
????public?void?init()?{
????????registerBeanDefinitionParser("service",?new?ProviderFactoryBeanDefinitionParser());
????}
}

ProviderFactoryBeanDefinitionParser：

protected?Class?getBeanClass(Element?element)?{
????????return?ProviderFactoryBean.class;
????}

????protected?void?doParse(Element?element,?BeanDefinitionBuilder?bean)?{

????????try?{
????????????String?serviceItf?=?element.getAttribute("interface");
????????????String?serverPort?=?element.getAttribute("serverPort");
????????????String?ref?=?element.getAttribute("ref");
????????????//?....
????????????bean.addPropertyValue("serverPort",?Integer.parseInt(serverPort));
????????????bean.addPropertyValue("serviceItf",?Class.forName(serviceItf));
????????????bean.addPropertyReference("serviceObject",?ref);
????????????//...
????????????if?(NumberUtils.isNumber(weight))?{
????????????????bean.addPropertyValue("weight",?Integer.parseInt(weight));
????????????}
????????????//...
???????}?catch?(Exception?e)?{
????????????//?...????????
??????}
????}

ProviderFactoryBean：

/**
?*?@author?孫浩
?*?@Descrption?服務(wù)發(fā)布
?***/
public?class?ProviderFactoryBean?implements?FactoryBean,?InitializingBean?{

????//服務(wù)接口
????private?Class?serviceItf;
????//服務(wù)實(shí)現(xiàn)
????private?Object?serviceObject;
????//服務(wù)端口
????private?String?serverPort;
????//服務(wù)超時(shí)時(shí)間
????private?long?timeout;
????//服務(wù)代理對(duì)象，暫時(shí)沒(méi)有用到
????private?Object?serviceProxyObject;
????//服務(wù)提供者唯一標(biāo)識(shí)
????private?String?appKey;
????//服務(wù)分組組名
????private?String?groupName?=?"default";
????//服務(wù)提供者權(quán)重，默認(rèn)為?1?,?范圍為?[1-100]
????private?int?weight?=?1;
????//服務(wù)端線程數(shù)，默認(rèn)?10?個(gè)線程
????private?int?workerThreads?=?10;

????@Override
????public?Object?getObject()?throws?Exception?{
????????return?serviceProxyObject;
????}

????@Override
????public?Class?getObjectType()?{
????????return?serviceItf;
????}

????@Override
????public?void?afterPropertiesSet()?throws?Exception?{
????????//啟動(dòng)?Netty?服務(wù)端
????????NettyServer.singleton().start(Integer.parseInt(serverPort));
????????//注冊(cè)到?zk,?元數(shù)據(jù)注冊(cè)中心
????????List?providerServiceList?=?buildProviderServiceInfos();
????????IRegisterCenter4Provider?registerCenter4Provider?=?RegisterCenter.singleton();
????????registerCenter4Provider.registerProvider(providerServiceList);
????}
}

//================RegisterCenter#registerProvider======================
@Override
public?void?registerProvider(final?List?serviceMetaData)?{
????if?(CollectionUtils.isEmpty(serviceMetaData))?{
????????return;
????}

????//連接?zk,?注冊(cè)服務(wù)
????synchronized?(RegisterCenter.class)?{
????????for?(ProviderService?provider?:?serviceMetaData)?{
????????????String?serviceItfKey?=?provider.getServiceItf().getName();

????????????List?providers?=?providerServiceMap.get(serviceItfKey);
????????????if?(providers?==?null)?{
????????????????providers?=?Lists.newArrayList();
????????????}
????????????providers.add(provider);
????????????providerServiceMap.put(serviceItfKey,?providers);
????????}

????????if?(zkClient?==?null)?{
????????????zkClient?=?new?ZkClient(ZK_SERVICE,?ZK_SESSION_TIME_OUT,?ZK_CONNECTION_TIME_OUT,?new?SerializableSerializer());
????????}

????????//創(chuàng)建?ZK?命名空間/當(dāng)前部署應(yīng)用?APP?命名空間/
????????String?APP_KEY?=?serviceMetaData.get(0).getAppKey();
????????String?ZK_PATH?=?ROOT_PATH?+?"/"?+?APP_KEY;
????????boolean?exist?=?zkClient.exists(ZK_PATH);
????????if?(!exist)?{
????????????zkClient.createPersistent(ZK_PATH,?true);
????????}

????????for?(Map.Entry>?entry?:?providerServiceMap.entrySet())?{
????????????//服務(wù)分組
????????????String?groupName?=?entry.getValue().get(0).getGroupName();
????????????//創(chuàng)建服務(wù)提供者
????????????String?serviceNode?=?entry.getKey();
????????????String?servicePath?=?ZK_PATH?+?"/"?+?groupName?+?"/"?+?serviceNode?+?"/"?+?PROVIDER_TYPE;
????????????exist?=?zkClient.exists(servicePath);
????????????if?(!exist)?{
????????????????zkClient.createPersistent(servicePath,?true);
????????????}

????????????//創(chuàng)建當(dāng)前服務(wù)器節(jié)點(diǎn)
????????????int?serverPort?=?entry.getValue().get(0).getServerPort();//服務(wù)端口
????????????int?weight?=?entry.getValue().get(0).getWeight();//服務(wù)權(quán)重
????????????int?workerThreads?=?entry.getValue().get(0).getWorkerThreads();//服務(wù)工作線程
????????????String?localIp?=?IPHelper.localIp();
????????????String?currentServiceIpNode?=?servicePath?+?"/"?+?localIp?+?"|"?+?serverPort?+?"|"?+?weight?+?"|"?+?workerThreads?+?"|"?+?groupName;
????????????exist?=?zkClient.exists(currentServiceIpNode);
????????????if?(!exist)?{
????????????????//注意，這里創(chuàng)建的是臨時(shí)節(jié)點(diǎn)
????????????????zkClient.createEphemeral(currentServiceIpNode);
????????????}
????????????//監(jiān)聽(tīng)注冊(cè)服務(wù)的變化，同時(shí)更新數(shù)據(jù)到本地緩存
????????????zkClient.subscribeChildChanges(servicePath,?new?IZkChildListener()?{
????????????????@Override
????????????????public?void?handleChildChange(String?parentPath,?List?currentChilds)?throws?Exception?{
????????????????????if?(currentChilds?==?null)?{
????????????????????????currentChilds?=?Lists.newArrayList();
????????????????????}
????????????????????//存活的服務(wù)?IP?列表
????????????????????List?activityServiceIpList?=?Lists.newArrayList(Lists.transform(currentChilds,?new?Function()?{
????????????????????????@Override
????????????????????????public?String?apply(String?input)?{
????????????????????????????return?StringUtils.split(input,?"|")[0];
????????????????????????}
????????????????????}));
????????????????????refreshActivityService(activityServiceIpList);
????????????????}
????????????});

????????}
????}
}

至此服務(wù)實(shí)現(xiàn)類(lèi)已被載入 Spring 容器中，且服務(wù)接口信息也注冊(cè)到了注冊(cè)中心。

網(wǎng)絡(luò)通信：

作為生產(chǎn)者對(duì)外提供 RPC 服務(wù)，必須有一個(gè)網(wǎng)絡(luò)程序來(lái)來(lái)監(jiān)聽(tīng)請(qǐng)求和做出響應(yīng)。在 Java 領(lǐng)域 Netty 是一款高性能的 NIO 通信框架，很多的框架的通信都是采用 Netty 來(lái)實(shí)現(xiàn)的，本例中也采用它當(dāng)做通信服務(wù)器。

構(gòu)建并啟動(dòng) Netty 服務(wù)監(jiān)聽(tīng)指定端口：

public?void?start(final?int?port)?{
????????synchronized?(NettyServer.class)?{
????????????if?(bossGroup?!=?null?||?workerGroup?!=?null)?{
????????????????return;
????????????}

????????????bossGroup?=?new?NioEventLoopGroup();
????????????workerGroup?=?new?NioEventLoopGroup();
????????????ServerBootstrap?serverBootstrap?=?new?ServerBootstrap();
????????????serverBootstrap
????????????????????.group(bossGroup,?workerGroup)
????????????????????.channel(NioServerSocketChannel.class)
????????????????????.option(ChannelOption.SO_BACKLOG,?1024)
????????????????????.childOption(ChannelOption.SO_KEEPALIVE,?true)
????????????????????.childOption(ChannelOption.TCP_NODELAY,?true)
????????????????????.handler(new?LoggingHandler(LogLevel.INFO))
????????????????????.childHandler(new?ChannelInitializer()?{
????????????????????????@Override
????????????????????????protected?void?initChannel(SocketChannel?ch)?throws?Exception?{
????????????????????????????//注冊(cè)解碼器?NettyDecoderHandler
????????????????????????????ch.pipeline().addLast(new?NettyDecoderHandler(StormRequest.class,?serializeType));
????????????????????????????//注冊(cè)編碼器?NettyEncoderHandler
????????????????????????????ch.pipeline().addLast(new?NettyEncoderHandler(serializeType));
????????????????????????????//注冊(cè)服務(wù)端業(yè)務(wù)邏輯處理器?NettyServerInvokeHandler
????????????????????????????ch.pipeline().addLast(new?NettyServerInvokeHandler());
????????????????????????}
????????????????????});
????????????try?{
????????????????channel?=?serverBootstrap.bind(port).sync().channel();
????????????}?catch?(InterruptedException?e)?{
????????????????throw?new?RuntimeException(e);
????????????}
????????}
????}

上面的代碼中向 Netty 服務(wù)的 Pipeline 中添加了編解碼和業(yè)務(wù)處理器，當(dāng)接收到請(qǐng)求時(shí)，經(jīng)過(guò)編解碼后，真正處理業(yè)務(wù)的是業(yè)務(wù)處理器，即 NettyServerInvokeHandler，該處理器繼承自 SimpleChannelInboundHandler，當(dāng)數(shù)據(jù)讀取完成將觸發(fā)一個(gè)事件，并調(diào)用 NettyServerInvokeHandler#channelRead0 方法來(lái)處理請(qǐng)求。

@Override
protected?void?channelRead0(ChannelHandlerContext?ctx,?StormRequest?request)?throws?Exception?{
????if?(ctx.channel().isWritable())?{
????????//從服務(wù)調(diào)用對(duì)象里獲取服務(wù)提供者信息
????????ProviderService?metaDataModel?=?request.getProviderService();
????????long?consumeTimeOut?=?request.getInvokeTimeout();
????????final?String?methodName?=?request.getInvokedMethodName();

????????//根據(jù)方法名稱(chēng)定位到具體某一個(gè)服務(wù)提供者
????????String?serviceKey?=?metaDataModel.getServiceItf().getName();
????????//獲取限流工具類(lèi)
????????int?workerThread?=?metaDataModel.getWorkerThreads();
????????Semaphore?semaphore?=?serviceKeySemaphoreMap.get(serviceKey);
????????if?(semaphore?==?null)?{
????????????synchronized?(serviceKeySemaphoreMap)?{
????????????????semaphore?=?serviceKeySemaphoreMap.get(serviceKey);
????????????????if?(semaphore?==?null)?{
????????????????????semaphore?=?new?Semaphore(workerThread);
????????????????????serviceKeySemaphoreMap.put(serviceKey,?semaphore);
????????????????}
????????????}
????????}

????????//獲取注冊(cè)中心服務(wù)
????????IRegisterCenter4Provider?registerCenter4Provider?=?RegisterCenter.singleton();
????????List?localProviderCaches?=?registerCenter4Provider.getProviderServiceMap().get(serviceKey);

????????Object?result?=?null;
????????boolean?acquire?=?false;

????????try?{
????????????ProviderService?localProviderCache?=?Collections2.filter(localProviderCaches,?new?Predicate()?{
????????????????@Override
????????????????public?boolean?apply(ProviderService?input)?{
????????????????????return?StringUtils.equals(input.getServiceMethod().getName(),?methodName);
????????????????}
????????????}).iterator().next();
????????????Object?serviceObject?=?localProviderCache.getServiceObject();

????????????//利用反射發(fā)起服務(wù)調(diào)用
????????????Method?method?=?localProviderCache.getServiceMethod();
????????????//利用?semaphore?實(shí)現(xiàn)限流
????????????acquire?=?semaphore.tryAcquire(consumeTimeOut,?TimeUnit.MILLISECONDS);
????????????if?(acquire)?{
????????????????result?=?method.invoke(serviceObject,?request.getArgs());
????????????????//System.out.println("---------------"+result);
????????????}
????????}?catch?(Exception?e)?{
????????????System.out.println(JSON.toJSONString(localProviderCaches)?+?"??"?+?methodName+"?"+e.getMessage());
????????????result?=?e;
????????}?finally?{
????????????if?(acquire)?{
????????????????semaphore.release();
????????????}
????????}
????????//根據(jù)服務(wù)調(diào)用結(jié)果組裝調(diào)用返回對(duì)象
????????StormResponse?response?=?new?StormResponse();
????????response.setInvokeTimeout(consumeTimeOut);
????????response.setUniqueKey(request.getUniqueKey());
????????response.setResult(result);
????????//將服務(wù)調(diào)用返回對(duì)象回寫(xiě)到消費(fèi)端
????????ctx.writeAndFlush(response);
????}?else?{
????????logger.error("------------channel?closed!---------------");
????}
}

此處還有部分細(xì)節(jié)如自定義的編解碼器等，篇幅所限不在此詳述，繼承 MessageToByteEncoder 和 ByteToMessageDecoder 覆寫(xiě)對(duì)應(yīng)的 encode 和 decode 方法即可自定義編解碼器，使用到的序列化工具如 Hessian/Proto 等可參考對(duì)應(yīng)的官方文檔。

請(qǐng)求和響應(yīng)包裝：

為便于封裝請(qǐng)求和響應(yīng)，定義兩個(gè) bean 來(lái)表示請(qǐng)求和響應(yīng)。

請(qǐng)求：

/**
?*?@author?孫浩
?*?@Descrption
?***/
public?class?StormRequest?implements?Serializable?{

????private?static?final?long?serialVersionUID?=?-5196465012408804755L;
????//UUID，唯一標(biāo)識(shí)一次返回值
????private?String?uniqueKey;
????//服務(wù)提供者信息
????private?ProviderService?providerService;
????//調(diào)用的方法名稱(chēng)
????private?String?invokedMethodName;
????//傳遞參數(shù)
????private?Object[]?args;
????//消費(fèi)端應(yīng)用名
????private?String?appName;
????//消費(fèi)請(qǐng)求超時(shí)時(shí)長(zhǎng)
????private?long?invokeTimeout;
????//?getter/setter
}

響應(yīng)：

/**
?*?@author?孫浩
?*?@Descrption
?***/
public?class?StormResponse?implements?Serializable?{
????private?static?final?long?serialVersionUID?=?5785265307118147202L;
????//UUID,?唯一標(biāo)識(shí)一次返回值
????private?String?uniqueKey;
????//客戶(hù)端指定的服務(wù)超時(shí)時(shí)間
????private?long?invokeTimeout;
????//接口調(diào)用返回的結(jié)果對(duì)象
????private?Object?result;
????//getter/setter
}

客戶(hù)端（消費(fèi)者）

客戶(hù)端（消費(fèi)者）在 RPC 調(diào)用中主要是生成服務(wù)接口的代理對(duì)象，并從注冊(cè)中心獲取對(duì)應(yīng)的服務(wù)列表發(fā)起網(wǎng)絡(luò)請(qǐng)求。

客戶(hù)端和服務(wù)端一樣采用 Spring 來(lái)管理 bean 解析 XML 配置等不再贅述，重點(diǎn)看下以下幾點(diǎn)：

1、通過(guò) JDK 動(dòng)態(tài)代理來(lái)生成引入服務(wù)接口的代理對(duì)象

public?Object?getProxy()?{
????return?Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),?new?Class[]{targetInterface},?this);
}

2、從注冊(cè)中心獲取服務(wù)列表并依據(jù)某種策略選取其中一個(gè)服務(wù)節(jié)點(diǎn)

//服務(wù)接口名稱(chēng)
String?serviceKey?=?targetInterface.getName();
//獲取某個(gè)接口的服務(wù)提供者列表
IRegisterCenter4Invoker?registerCenter4Consumer?=?RegisterCenter.singleton();
List?providerServices?=?registerCenter4Consumer.getServiceMetaDataMap4Consume().get(serviceKey);
//根據(jù)軟負(fù)載策略，從服務(wù)提供者列表選取本次調(diào)用的服務(wù)提供者
ClusterStrategy?clusterStrategyService?=?ClusterEngine.queryClusterStrategy(clusterStrategy);
ProviderService?providerService?=?clusterStrategyService.select(providerServices);

3、通過(guò) Netty 建立連接，發(fā)起網(wǎng)絡(luò)請(qǐng)求

/**
?*?@author?孫浩
?*?@Descrption?Netty?消費(fèi)端?bean?代理工廠
?***/
public?class?RevokerProxyBeanFactory?implements?InvocationHandler?{
????private?ExecutorService?fixedThreadPool?=?null;
????//服務(wù)接口
????private?Class?targetInterface;
????//超時(shí)時(shí)間
????private?int?consumeTimeout;
????//調(diào)用者線程數(shù)
????private?static?int?threadWorkerNumber?=?10;
????//負(fù)載均衡策略
????private?String?clusterStrategy;

????@Override
????public?Object?invoke(Object?proxy,?Method?method,?Object[]?args)?throws?Throwable?{

????????...

????????//復(fù)制一份服務(wù)提供者信息
????????ProviderService?newProvider?=?providerService.copy();
????????//設(shè)置本次調(diào)用服務(wù)的方法以及接口
????????newProvider.setServiceMethod(method);
????????newProvider.setServiceItf(targetInterface);

????????//聲明調(diào)用?AresRequest?對(duì)象，AresRequest?表示發(fā)起一次調(diào)用所包含的信息
????????final?StormRequest?request?=?new?StormRequest();
????????//設(shè)置本次調(diào)用的唯一標(biāo)識(shí)
????????request.setUniqueKey(UUID.randomUUID().toString()?+?"-"?+?Thread.currentThread().getId());
????????//設(shè)置本次調(diào)用的服務(wù)提供者信息
????????request.setProviderService(newProvider);
????????//設(shè)置本次調(diào)用的方法名稱(chēng)
????????request.setInvokedMethodName(method.getName());
????????//設(shè)置本次調(diào)用的方法參數(shù)信息
????????request.setArgs(args);

????????try?{
????????????//構(gòu)建用來(lái)發(fā)起調(diào)用的線程池
????????????if?(fixedThreadPool?==?null)?{
????????????????synchronized?(RevokerProxyBeanFactory.class)?{
????????????????????if?(null?==?fixedThreadPool)?{
????????????????????????fixedThreadPool?=?Executors.newFixedThreadPool(threadWorkerNumber);
????????????????????}
????????????????}
????????????}
????????????//根據(jù)服務(wù)提供者的?ip,port,?構(gòu)建?InetSocketAddress?對(duì)象，標(biāo)識(shí)服務(wù)提供者地址
????????????String?serverIp?=?request.getProviderService().getServerIp();
????????????int?serverPort?=?request.getProviderService().getServerPort();
????????????InetSocketAddress?inetSocketAddress?=?new?InetSocketAddress(serverIp,?serverPort);
????????????//提交本次調(diào)用信息到線程池?fixedThreadPool,?發(fā)起調(diào)用
????????????Future?responseFuture?=?fixedThreadPool.submit(RevokerServiceCallable.of(inetSocketAddress,?request));
????????????//獲取調(diào)用的返回結(jié)果
????????????StormResponse?response?=?responseFuture.get(request.getInvokeTimeout(),?TimeUnit.MILLISECONDS);
????????????if?(response?!=?null)?{
????????????????return?response.getResult();
????????????}
????????}?catch?(Exception?e)?{
????????????throw?new?RuntimeException(e);
????????}
????????return?null;
????}
????//??...
}

Netty 的響應(yīng)是異步的，為了在方法調(diào)用返回前獲取到響應(yīng)結(jié)果，需要將異步的結(jié)果同步化。

4、Netty 異步返回的結(jié)果存入阻塞隊(duì)列

@Override
protected?void?channelRead0(ChannelHandlerContext?channelHandlerContext,?StormResponse?response)?throws?Exception?{
????//將?Netty?異步返回的結(jié)果存入阻塞隊(duì)列，以便調(diào)用端同步獲取
????RevokerResponseHolder.putResultValue(response);
}

5、請(qǐng)求發(fā)出后同步獲取結(jié)果

//提交本次調(diào)用信息到線程池?fixedThreadPool,?發(fā)起調(diào)用
Future?responseFuture?=?fixedThreadPool.submit(RevokerServiceCallable.of(inetSocketAddress,?request));
//獲取調(diào)用的返回結(jié)果
StormResponse?response?=?responseFuture.get(request.getInvokeTimeout(),?TimeUnit.MILLISECONDS);
if?(response?!=?null)?{
????return?response.getResult();
}

//===================================================
//從返回結(jié)果容器中獲取返回結(jié)果，同時(shí)設(shè)置等待超時(shí)時(shí)間為?invokeTimeout
long?invokeTimeout?=?request.getInvokeTimeout();
StormResponse?response?=?RevokerResponseHolder.getValue(request.getUniqueKey(),?invokeTimeout);

—?3?—

測(cè)試

Server：

/**
?*?@author?孫浩
?*?@Descrption
?***/
public?class?MainServer?{
????public?static?void?main(String[]?args)?throws?Exception?{
????????//發(fā)布服務(wù)
????????final?ClassPathXmlApplicationContext?context?=?new?ClassPathXmlApplicationContext("storm-server.xml");
????????System.out.println("?服務(wù)發(fā)布完成");
????}
}

Client：

public?class?Client?{

????private?static?final?Logger?logger?=?LoggerFactory.getLogger(Client.class);

????public?static?void?main(String[]?args)?throws?Exception?{

????????final?ClassPathXmlApplicationContext?context?=?new?ClassPathXmlApplicationContext("storm-client.xml");
????????final?HelloService?helloService?=?(HelloService)?context.getBean("helloService");
????????String?result?=?helloService.sayHello("World");
????????System.out.println(result);
????????for?(;;)?{

????????}
????}
}

結(jié)果

生產(chǎn)者：

消費(fèi)者：

注冊(cè)中心：

—?4?—

總結(jié)

本文簡(jiǎn)單介紹了 RPC 的整個(gè)流程，并實(shí)現(xiàn)了一個(gè)簡(jiǎn)單的 RPC 調(diào)用。希望閱讀完本文之后，能加深你對(duì) RPC 的一些認(rèn)識(shí)。

生產(chǎn)者端流程：

• 加載服務(wù)接口，并緩存

• 服務(wù)注冊(cè)，將服務(wù)接口以及服務(wù)主機(jī)信息寫(xiě)入注冊(cè)中心（本例使用的是 ZooKeeper)

• 啟動(dòng)網(wǎng)絡(luò)服務(wù)器并監(jiān)聽(tīng)

• 反射，本地調(diào)用

消費(fèi)者端流程：

• 代理服務(wù)接口生成代理對(duì)象

• 服務(wù)發(fā)現(xiàn)（連接 ZooKeeper，拿到服務(wù)地址列表，通過(guò)客戶(hù)端負(fù)載策略獲取合適的服務(wù)地址）

• 遠(yuǎn)程方法調(diào)用（本例通過(guò) Netty，發(fā)送消息，并獲取響應(yīng)結(jié)果）

限于篇幅，本文代碼并不完整，如有需要，可以參考以下鏈接，我貼出來(lái)了完整代碼給你。

https://github.com/fankongqiumu/storm.git

推薦閱讀：

世界的真實(shí)格局分析，地球人類(lèi)社會(huì)底層運(yùn)行原理

不是你需要中臺(tái)，而是一名合格的架構(gòu)師（附各大廠中臺(tái)建設(shè)PPT）

企業(yè)IT技術(shù)架構(gòu)規(guī)劃方案

論數(shù)字化轉(zhuǎn)型——轉(zhuǎn)什么，如何轉(zhuǎn)？

華為干部與人才發(fā)展手冊(cè)（附PPT）

企業(yè)10大管理流程圖，數(shù)字化轉(zhuǎn)型從業(yè)者必備！

【中臺(tái)實(shí)踐】華為大數(shù)據(jù)中臺(tái)架構(gòu)分享.pdf

華為的數(shù)字化轉(zhuǎn)型方法論

華為如何實(shí)施數(shù)字化轉(zhuǎn)型（附PPT）

超詳細(xì)280頁(yè)Docker實(shí)戰(zhàn)文檔！開(kāi)放下載

華為大數(shù)據(jù)解決方案（PPT）