以ClientProtocol接口中的rename RPC調(diào)用進(jìn)行一次實(shí)例分析。

rename方法在ClientProtocol接口中定義，它的兩個(gè)參數(shù)是String類型的，不能直接通過網(wǎng)絡(luò)傳輸。

我們看誰實(shí)現(xiàn)了ClientProtocol接口并重寫rename方法。

看到是ClientNamenodeProtocolTranslatorPB這個(gè)類。看下這個(gè)類如何實(shí)現(xiàn)的rename方法，代碼如下：

??@Override
??public?boolean?rename(String?src,?String?dst)?throws?IOException?{
??????//把String類型的參數(shù)進(jìn)行包裝，變成可通過網(wǎng)絡(luò)傳輸?shù)男蛄谢愋?br />????RenameRequestProto?req?=?RenameRequestProto.newBuilder()
????????.setSrc(src)
????????.setDst(dst).build();

????try?{
????????//通過調(diào)用rpcProxy的rename方法實(shí)現(xiàn)rpc調(diào)用，接下來我們就看rpcProxy這個(gè)對(duì)象
??????return?rpcProxy.rename(null,?req).getResult();
????}?catch?(ServiceException?e)?{
??????throw?ProtobufHelper.getRemoteException(e);
????}
??}

rpcProxy是ClientNamenodeProtocolTranslatorPB這個(gè)類的成員變量，類型是ClientNamenodeProtocolPB。

那我們就繼續(xù)點(diǎn)進(jìn)去rpcProxy.rename方法。發(fā)現(xiàn)點(diǎn)不進(jìn)去了，因?yàn)閞pcProxy的類型是ClientNamenodeProtocolPB。這個(gè)類繼承了protoBuf生成的類，我們還沒有編譯所以點(diǎn)不進(jìn)去。

但是可以看看相關(guān)的.proto文件。找到ClientNamenodeProtocol.proto文件，搜索rename。能夠看到protobuf定義的參數(shù)和返回值。參數(shù)就是把原本的兩個(gè)字符串類型包裝成新的RenameRequestProto類型。

似乎我們是沒法往下看了。感覺也沒什么收獲？那你感覺錯(cuò)了。我們還可以看看ClientNamenodeProtocolTranslatorPB的構(gòu)造函數(shù)，看看是怎么初始化的rpcProxy對(duì)象。

繼續(xù)追，看看誰調(diào)用了這個(gè)構(gòu)造函數(shù)。

這三處地方調(diào)用了。我們進(jìn)入第一個(gè)方法：

我們繼續(xù)看這個(gè)proxy是怎么生成的。

這是一個(gè)鏈?zhǔn)秸{(diào)用。先調(diào)用了RPC.getProtocolProxy，然后調(diào)用了getProxy()。

調(diào)用getProtocolProxy得到一個(gè)ProtocolProxy< ClientNamenodeProtocolPB>對(duì)象，這個(gè)對(duì)象包含了用于和服務(wù)端通信的代理對(duì)象和服務(wù)端支持的方法集合。在這個(gè)對(duì)象的基礎(chǔ)上調(diào)用getProxy方法得到那個(gè)代理對(duì)象。

RPC.getProtocolProxy()的代碼如下：

可以看到首先調(diào)用了getProtocolEngine獲得通信所使用的RpcEngine，然后又調(diào)用了getProxy獲得一個(gè)客戶端-side 的代理對(duì)象。

RpcEngine.getProxy()的實(shí)現(xiàn)如下(我們進(jìn)入的是ProtobufRpcEngine的實(shí)現(xiàn))：

進(jìn)到這里一切豁然開朗，可以看到這里使用了JDK的動(dòng)態(tài)代理。我們關(guān)注這句代碼：

return?new?ProtocolProxy(protocol,?(T)?Proxy.newProxyInstance(
????????protocol.getClassLoader(),?new?Class[]{protocol},?invoker),?false);

new ProtocolProxy()的第二個(gè)參數(shù)代表這個(gè)對(duì)象里保存的代理對(duì)象。

(T)?Proxy.newProxyInstance(
????????protocol.getClassLoader(),?new?Class[]{protocol},?invoker)

生成代理對(duì)象使用JDK中Proxy類的靜態(tài)方法newProxyInstance。第三個(gè)參數(shù)是InvocationHandler的實(shí)現(xiàn)，真正的方法調(diào)用就在這個(gè)InvocationHandler里的invoke里調(diào)用。我們看到傳入的invoke是new的ProtobufRpcEngine的內(nèi)部類Invoker。這個(gè)Invoker最終會(huì)在invoke方法中去調(diào)用Client.call方法向RPC Server發(fā)送RPC請(qǐng)求，這個(gè)過程對(duì)客戶端是完全透明的。如下圖所示， 已用紅框標(biāo)出。

至此我們是明白了客戶端是如何把RPC請(qǐng)求發(fā)送到服務(wù)器了。總結(jié)一下(采用正序的方式)

因?yàn)槲覀冋{(diào)用方法一般是使用DFSClient這個(gè)類的對(duì)象去調(diào)用，所以在DFSClient里搜rename方法。可以看到這個(gè)方法內(nèi)部調(diào)用namenode.rename()方法。那么我們就需要知道那么namenode是怎么構(gòu)造出來的。

namenode的聲明如下：

final?ClientProtocol?namenode;

它是一個(gè)ClientProtocol類型的變量，因此可以調(diào)用所有客戶端對(duì)NN進(jìn)行操作的方法。接下來找構(gòu)造函數(shù)，看看如何給namenode成員變量賦值的。

DFSClient在自己的構(gòu)造函數(shù)里會(huì)調(diào)用 NameNodeProxiesClient.createProxyWithClientProtocol()來獲得一個(gè)支持ClientProtocol協(xié)議的代理對(duì)象。最后把NameNode的代理賦值給namenode成員。這樣我們就可以通過namenode對(duì)象來調(diào)用各種方法操作了。

創(chuàng)建這個(gè)代理的方法是如下代碼：

紅色箭頭所指分別代表創(chuàng)建non-HA的和HA的代理。如果配置了dfs.client.failover.proxy.provider.$nameservice屬性，就認(rèn)為是開啟HA的集群。因?yàn)楝F(xiàn)在集群幾乎都是HA的，所以我們看下面這個(gè)箭頭的創(chuàng)建HA代理對(duì)象的方法createHAProxy。記住傳入的最后兩個(gè)參數(shù)ClientProtocol.class和failoverProxyProvider。后面我們還會(huì)用到這兩個(gè)參數(shù)，到時(shí)候可以再回來看。

createHAProxy方法代碼如下：

??public?static??ProxyAndInfo?createHAProxy(
??????Configuration?conf,?URI?nameNodeUri,?Class?xface,
??????AbstractNNFailoverProxyProvider?failoverProxyProvider)?{
????Preconditions.checkNotNull(failoverProxyProvider);//?HA?case
????DfsClientConf?config?=?new?DfsClientConf(conf);//我們主要看這個(gè)，因?yàn)槟憧捶椒ǖ姆祷刂凳荘roxy和Info，我們只需要關(guān)注Proxy就好了。
????T?proxy?=?(T)?RetryProxy.create(xface,?failoverProxyProvider,
????????RetryPolicies.failoverOnNetworkException(
????????????RetryPolicies.TRY_ONCE_THEN_FAIL,?config.getMaxFailoverAttempts(),
????????????config.getMaxRetryAttempts(),?config.getFailoverSleepBaseMillis(),
????????????config.getFailoverSleepMaxMillis()));
????Text?dtService;if?(failoverProxyProvider.useLogicalURI())?{
??????dtService?=?HAUtilClient.buildTokenServiceForLogicalUri(nameNodeUri,
??????????HdfsConstants.HDFS_URI_SCHEME);
????}?else?{
??????dtService?=?SecurityUtil.buildTokenService(
??????????DFSUtilClient.getNNAddress(nameNodeUri));
????}return?new?ProxyAndInfo<>(proxy,?dtService,
????????DFSUtilClient.getNNAddressCheckLogical(conf,?nameNodeUri));
??}

因?yàn)橐话阄覀兌紩?huì)有retry機(jī)制，所以這里使用了RetryProxy去create代理。跟到RetryProxy.create中。注意create的前兩個(gè)參數(shù)就是剛剛讓大家記住的兩個(gè)。分別是：ClientProtocol.class和failoverProxyProvider。

進(jìn)到RetryProxy.create源碼：

此時(shí)這個(gè)泛型T，其實(shí)就是ClientProtocol。這個(gè)方法里使用JDK的動(dòng)態(tài)代理生成代理對(duì)象，這塊我們應(yīng)該很熟悉。newProxyInstance的三個(gè)參數(shù)，我們看第三個(gè)參數(shù)，因?yàn)樵诖韺?duì)象調(diào)用方法最終都會(huì)由這個(gè)RetryInvocationHandler的invoke方法接管。所以我們?nèi)タ纯碦etryInvocationHandler的實(shí)現(xiàn)。點(diǎn)進(jìn)去。

可以看到，RetryInvocationHandler的構(gòu)造方法中還是有我們之前讓大家記住的參數(shù)failoverProxyProvider。

因?yàn)镴DK的動(dòng)態(tài)代理最終方法會(huì)走到invoke。所以我們直接去看RetryInvocationHandler的invoke方法。

這里主要是通過循環(huán)去實(shí)現(xiàn)retry。循環(huán)中關(guān)鍵方法是invokeOnce()。點(diǎn)進(jìn)去。我們關(guān)注正常情況下的處理，先不關(guān)注異常情況。正常情況下就是再調(diào)用invoke()

再點(diǎn)進(jìn)去：

繼續(xù)點(diǎn)進(jìn)invokeMethod()方法中：

再點(diǎn)進(jìn)invokeMethod中，終于到頭了。不能再繼續(xù)往下點(diǎn)invoke了，那就是JDK的方法了。框起來的代碼表示在代理對(duì)象上調(diào)用此Method底層代表的方法(本例是rename)，并把a(bǔ)rgs作為參數(shù)傳給它。

那這個(gè)代理對(duì)象是誰呢，proxyDescriptor.getProxy()返回的是什么呢？這時(shí)候想想剛才讓大家記住的那個(gè)參數(shù)。

所以proxyDescriptor里面的proxy就是proxyProvider中保存的代理對(duì)象信息。而創(chuàng)建這個(gè)proxyProvider的時(shí)機(jī)正是在DFSClient調(diào)用的createProxyWithClientProtocol方法里。而且，從下面的的代碼可以看到，if (failoverProxyProvider == null)還會(huì)再這樣我們就可以唱：“又回到最初的起點(diǎn)，呆呆的站在代碼前”。

點(diǎn)進(jìn)去看看怎么創(chuàng)建的failoverProxyProvider。經(jīng)過一系列重載，最終走到下圖這個(gè)地方。可以看到是通過反射機(jī)制，去new的FailoverProxyProvider對(duì)象。newInstance的第三個(gè)參數(shù)xface就是我們的代理類的類型，所以最終這個(gè)failoverProxyprovider里面會(huì)有我們需要的代理對(duì)象。到此分析結(jié)束。

這里面還有一個(gè)細(xì)節(jié)。就是如何得到的FailoverProxyProvider的Class對(duì)象。通過跟蹤getFailoverProxyProviderClass方法。

上圖紅框中圈出的configKey = “dfs.client.failover.proxy.provider”+host。這個(gè)host是在core-site.xml中fs.defaultFS這個(gè)屬性里配置的，把hdfs的schema去掉就是host。這個(gè)配置正好對(duì)應(yīng)了官網(wǎng)hdfs-site.xml中的：

那這個(gè)host就是nameservice ID。為什么這么說呢，因?yàn)楣倬W(wǎng)配置解釋就這么說的：

那我們看看一般這個(gè)類是什么，可以看到，一般是ConfiguredFailoverProxyProvider這個(gè)類。我們通過這個(gè)配置項(xiàng)拿到了類的全路徑名。

最后通過Class.forName就加載了這個(gè)類。

大功告成。

創(chuàng)建完namenode這個(gè)代理對(duì)象之后，就可以通過它調(diào)用各種方法了。比如我們之前舉例的rename。我們點(diǎn)進(jìn)去，進(jìn)入了ClientProtocol接口。那最終肯定調(diào)用的是它的實(shí)現(xiàn)類。這就回到了我們文章開頭。

至此Hadoop RPC在客戶端一側(cè)的調(diào)用流程已經(jīng)全部分析完畢。

總結(jié)

以上是生活随笔為你收集整理的怎么看调用的接口_Hadoop RPC调用实例分析的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。