100 行代碼透徹解析 RPC 原理
文章來源：https://sourl.cn/HpZHvy
引 言
一次基本的 RPC 調(diào)用會涉及到什么？
現(xiàn)在業(yè)界內(nèi)比較流行的一些 RPC 框架，例如 Dubbo 提供的是基于接口的遠(yuǎn)程方法調(diào)用，即客戶端只需要知道接口的定義即可調(diào)用遠(yuǎn)程服務(wù)。在 Java 中接口并不能直接調(diào)用實(shí)例方法，必須通過其實(shí)現(xiàn)類對象來完成此操作，這意味著客戶端必須為這些接口生成代理對象，對此 Java 提供了?Proxy、InvocationHandler?生成動(dòng)態(tài)代理的支持；生成了代理對象，那么每個(gè)具體的發(fā)方法是怎么調(diào)用的呢？jdk 動(dòng)態(tài)代理生成的代理對象調(diào)用指定方法時(shí)實(shí)際會執(zhí)行?InvocationHandler?中定義的?#invoke?方法，在該方法中完成遠(yuǎn)程方法調(diào)用并獲取結(jié)果。
拋開客戶端，回過頭來看 RPC 是兩臺計(jì)算機(jī)間的調(diào)用，實(shí)質(zhì)上是兩臺主機(jī)間的網(wǎng)絡(luò)通信，涉及到網(wǎng)絡(luò)通信又必然會有序列化、反序列化，編解碼等一些必須要考慮的問題；同時(shí)實(shí)際上現(xiàn)在大多系統(tǒng)都是集群部署的，多臺主機(jī)/容器對外提供相同的服務(wù)，如果集群的節(jié)點(diǎn)數(shù)量很大的話，那么管理服務(wù)地址也將是一件十分繁瑣的事情，常見的做法是各個(gè)服務(wù)節(jié)點(diǎn)將自己的地址和提供的服務(wù)列表注冊到一個(gè)?注冊中心，由?注冊中心?來統(tǒng)一管理服務(wù)列表；這樣的做法解決了一些問題同時(shí)為客戶端增加了一項(xiàng)新的工作——那就是服務(wù)發(fā)現(xiàn)，通俗來說就是從注冊中心中找到遠(yuǎn)程方法對應(yīng)的服務(wù)列表并通過某種策略從中選取一個(gè)服務(wù)地址來完成網(wǎng)絡(luò)通信。
聊了客戶端和?注冊中心，另外一個(gè)重要的角色自然是服務(wù)端，服務(wù)端最重要的任務(wù)便是提供服務(wù)接口的真正實(shí)現(xiàn)并在某個(gè)端口上監(jiān)聽網(wǎng)絡(luò)請求，監(jiān)聽到請求后從網(wǎng)絡(luò)請求中獲取到對應(yīng)的參數(shù)（比如服務(wù)接口、方法、請求參數(shù)等），再根據(jù)這些參數(shù)通過反射的方式調(diào)用接口的真正實(shí)現(xiàn)獲取結(jié)果并將其寫入對應(yīng)的響應(yīng)流中。
綜上所述，一次基本的 RPC 調(diào)用流程大致如下：
基本實(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);
}
生產(chǎn)者要提供服務(wù)接口的實(shí)現(xiàn)，創(chuàng)建 HelloServiceImpl 實(shí)現(xiàn)類。
/**
?*?@author?孫浩
?*?@Descrption?服務(wù)實(shí)現(xiàn)
?***/
public?class?HelloServiceImpl?implements?HelloService?{
????@Override
????public?String?sayHello(String?somebody)?{
????????return?"hello?"?+?somebody?+?"!";
????}
}
服務(wù)注冊
本例使用 Spring 來管理 bean，采用自定義 xml 和解析器的方式來將服務(wù)實(shí)現(xiàn)類載入容器（當(dāng)然也可以采用自定義注解的方式，此處不過多論述）并將服務(wù)接口信息注冊到注冊中心。
首先自定義xsd,
<xsd:element?name="service">
????<xsd:complexType>
????????<xsd:complexContent>
????????????<xsd:extension?base="beans:identifiedType">
????????????????<xsd:attribute?name="interface"?type="xsd:string"?use="required"/>
????????????????<xsd:attribute?name="timeout"?type="xsd:int"?use="required"/>
????????????????<xsd:attribute?name="serverPort"?type="xsd:int"?use="required"/>
????????????????<xsd:attribute?name="ref"?type="xsd:string"?use="required"/>
????????????????<xsd:attribute?name="weight"?type="xsd:int"?use="optional"/>
????????????????<xsd:attribute?name="workerThreads"?type="xsd:int"?use="optional"/>
????????????????<xsd:attribute?name="appKey"?type="xsd:string"?use="required"/>
????????????????<xsd:attribute?name="groupName"?type="xsd:string"?use="optional"/>
????????????xsd:extension>
????????xsd:complexContent>
????xsd:complexType>
xsd:element>
分別指定 schema 和 xmd，schema 和對應(yīng) handler 的映射：
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
將編寫好的文件放入?classpath?下的?META-INF?目錄下：
在 Spring 配置文件中配置服務(wù)類：
?<bean?id="helloService"?class="com.hsunfkqm.storm.framework.test.HelloServiceImpl"/>
?<storm:service?id="helloServiceRegister"
?????????????????????interface="com.hsunfkqm.storm.framework.test.HelloService"
?????????????????????ref="helloService"
?????????????????????groupName="default"
?????????????????????weight="2"
?????????????????????appKey="ares"
?????????????????????workerThreads="100"
?????????????????????serverPort="8081"
?????????????????????timeout="600"/>
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());
????}
}
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)?{
????????????//?...????????
??????}
????}/**
?*?@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ù)代理對象，暫時(shí)沒有用到
????private?Object?serviceProxyObject;
????//服務(wù)提供者唯一標(biāo)識
????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));
????????//注冊到?zk,?元數(shù)據(jù)注冊中心
????????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,?注冊服務(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)聽注冊服務(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)類已被載入 Spring 容器中，且服務(wù)接口信息也注冊到了注冊中心。
作為生產(chǎn)者對外提供 RPC 服務(wù)，必須有一個(gè)網(wǎng)絡(luò)程序來來監(jiān)聽請求和做出響應(yīng)。在 Java 領(lǐng)域 Netty 是一款高性能的 NIO 通信框架，很多的框架的通信都是采用 Netty 來實(shí)現(xiàn)的，本例中也采用它當(dāng)做通信服務(wù)器。
構(gòu)建并啟動(dòng) Netty 服務(wù)監(jiān)聽指定端口：
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?{
????????????????????????????//注冊解碼器?NettyDecoderHandler
????????????????????????????ch.pipeline().addLast(new?NettyDecoderHandler(StormRequest.class,?serializeType));
????????????????????????????//注冊編碼器?NettyEncoderHandler
????????????????????????????ch.pipeline().addLast(new?NettyEncoderHandler(serializeType));
????????????????????????????//注冊服務(wù)端業(yè)務(wù)邏輯處理器?NettyServerInvokeHandler
????????????????????????????ch.pipeline().addLast(new?NettyServerInvokeHandler());
????????????????????????}
????????????????????});
????????????try?{
????????????????channel?=?serverBootstrap.bind(port).sync().channel();
????????????}?catch?(InterruptedException?e)?{
????????????????throw?new?RuntimeException(e);
????????????}
????????}
????}
@Override
protected?void?channelRead0(ChannelHandlerContext?ctx,?StormRequest?request)?throws?Exception?{
????if?(ctx.channel().isWritable())?{
????????//從服務(wù)調(diào)用對象里獲取服務(wù)提供者信息
????????ProviderService?metaDataModel?=?request.getProviderService();
????????long?consumeTimeOut?=?request.getInvokeTimeout();
????????final?String?methodName?=?request.getInvokedMethodName();
????????//根據(jù)方法名稱定位到具體某一個(gè)服務(wù)提供者
????????String?serviceKey?=?metaDataModel.getServiceItf().getName();
????????//獲取限流工具類
????????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);
????????????????}
????????????}
????????}
????????//獲取注冊中心服務(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)用返回對象
????????StormResponse?response?=?new?StormResponse();
????????response.setInvokeTimeout(consumeTimeOut);
????????response.setUniqueKey(request.getUniqueKey());
????????response.setResult(result);
????????//將服務(wù)調(diào)用返回對象回寫到消費(fèi)端
????????ctx.writeAndFlush(response);
????}?else?{
????????logger.error("------------channel?closed!---------------");
????}
}
此處還有部分細(xì)節(jié)如自定義的編解碼器等，篇幅所限不在此詳述，繼承?MessageToByteEncoder?和?ByteToMessageDecoder?覆寫對應(yīng)的?encode?和?decode?方法即可自定義編解碼器，使用到的序列化工具如 Hessian/Proto 等可參考對應(yīng)的官方文檔。
請求和響應(yīng)包裝
為便于封裝請求和響應(yīng)，定義兩個(gè) bean 來表示請求和響應(yīng)。
請求：
/**
?*?@author?孫浩
?*?@Descrption
?***/
public?class?StormRequest?implements?Serializable?{
????private?static?final?long?serialVersionUID?=?-5196465012408804755L;
????//UUID,?唯一標(biāo)識一次返回值
????private?String?uniqueKey;
????//服務(wù)提供者信息
????private?ProviderService?providerService;
????//調(diào)用的方法名稱
????private?String?invokedMethodName;
????//傳遞參數(shù)
????private?Object[]?args;
????//消費(fèi)端應(yīng)用名
????private?String?appName;
????//消費(fèi)請求超時(shí)時(shí)長
????private?long?invokeTimeout;
????//?getter/setter
}
響應(yīng)：
/**
?*?@author?孫浩
?*?@Descrption
?***/
public?class?StormResponse?implements?Serializable?{
????private?static?final?long?serialVersionUID?=?5785265307118147202L;
????//UUID,?唯一標(biāo)識一次返回值
????private?String?uniqueKey;
????//客戶端指定的服務(wù)超時(shí)時(shí)間
????private?long?invokeTimeout;
????//接口調(diào)用返回的結(jié)果對象
????private?Object?result;
????//getter/setter
}
客戶端（消費(fèi)者）在 RPC 調(diào)用中主要是生成服務(wù)接口的代理對象，并從注冊中心獲取對應(yīng)的服務(wù)列表發(fā)起網(wǎng)絡(luò)請求。
客戶端和服務(wù)端一樣采用 Spring 來管理 bean 解析 xml 配置等不再贅述，重點(diǎn)看下以下幾點(diǎn)：
通過 jdk 動(dòng)態(tài)代理來生成引入服務(wù)接口的代理對象
public?Object?getProxy()?{
????return?Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),?new?Class[]{targetInterface},?this);
}
從注冊中心獲取服務(wù)列表并依據(jù)某種策略選取其中一個(gè)服務(wù)節(jié)點(diǎn)
//服務(wù)接口名稱
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);
通過 Netty 建立連接，發(fā)起網(wǎng)絡(luò)請求
/**
?*?@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?對象，AresRequest?表示發(fā)起一次調(diào)用所包含的信息
????????final?StormRequest?request?=?new?StormRequest();
????????//設(shè)置本次調(diào)用的唯一標(biāo)識
????????request.setUniqueKey(UUID.randomUUID().toString()?+?"-"?+?Thread.currentThread().getId());
????????//設(shè)置本次調(diào)用的服務(wù)提供者信息
????????request.setProviderService(newProvider);
????????//設(shè)置本次調(diào)用的方法名稱
????????request.setInvokedMethodName(method.getName());
????????//設(shè)置本次調(diào)用的方法參數(shù)信息
????????request.setArgs(args);
????????try?{
????????????//構(gòu)建用來發(fā)起調(diào)用的線程池
????????????if?(fixedThreadPool?==?null)?{
????????????????synchronized?(RevokerProxyBeanFactory.class)?{
????????????????????if?(null?==?fixedThreadPool)?{
????????????????????????fixedThreadPool?=?Executors.newFixedThreadPool(threadWorkerNumber);
????????????????????}
????????????????}
????????????}
????????????//根據(jù)服務(wù)提供者的?ip,port,?構(gòu)建?InetSocketAddress?對象，標(biāo)識服務(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é)果同步化。
Netty 異步返回的結(jié)果存入阻塞隊(duì)列
@Override
protected?void?channelRead0(ChannelHandlerContext?channelHandlerContext,?StormResponse?response)?throws?Exception?{
????//將?Netty?異步返回的結(jié)果存入阻塞隊(duì)列，以便調(diào)用端同步獲取
????RevokerResponseHolder.putResultValue(response);
}
請求發(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); 測 試
/**
?*?@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ā)布完成");
????}
}
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é) 果
總 結(jié)
本文簡單介紹了 RPC 的整個(gè)流程，并實(shí)現(xiàn)了一個(gè)簡單的 RPC 調(diào)用。希望閱讀完本文之后，能加深你對 RPC 的一些認(rèn)識。
- 生產(chǎn)者端流程：
加載服務(wù)接口，并緩存
服務(wù)注冊，將服務(wù)接口以及服務(wù)主機(jī)信息寫入注冊中心（本例使用的是 zookeeper)
啟動(dòng)網(wǎng)絡(luò)服務(wù)器并監(jiān)聽
反射，本地調(diào)用
- 消費(fèi)者端流程：
代理服務(wù)接口生成代理對象
服務(wù)發(fā)現(xiàn)（連接 zookeeper，拿到服務(wù)地址列表，通過客戶端負(fù)載策略獲取合適的服務(wù)地址）
遠(yuǎn)程方法調(diào)用（本例通過 Netty，發(fā)送消息，并獲取響應(yīng)結(jié)果）
如有錯(cuò)誤之處，還望大家指正。
完
往期推薦
前OPPO 員工控訴：離職時(shí)績效為B，離職后被悄悄改成D，就為了少發(fā)年終獎(jiǎng)！
同事多線程使用不當(dāng)導(dǎo)致OOM，被我懟了一頓
IntelliJ IDEA 卡成球了？
有道無術(shù)，術(shù)可成；有術(shù)無道，止于術(shù)
歡迎大家關(guān)注Java之道公眾號
好文章，我在看??
評論
圖片
表情