java開發(fā)Dubbo負(fù)載均衡與集群容錯示例詳解

第java開發(fā)Dubbo負(fù)載均衡與集群容錯示例詳解目錄負(fù)載均衡與集群容錯Invoker服務(wù)目錄RegistryDirectory獲取Invoker列表監(jiān)聽注冊中心刷新Invoker列表StaticDirectory服務(wù)路由ClusterFailoverClusterInvokerFailfastClusterInvokerFailsafeClusterInvokerFailbackClusterInvokerForkingClusterInvokerBroadcastClusterInvokerAbstractClusterInvoker小結(jié)負(fù)載均衡AbstractLoadBalanceRandomLoadBalanceLeastActiveLoadBalanceConsistentHashLoadBalance

RoundRobinLoadBalance總結(jié)

負(fù)載均衡與集群容錯

Invoker

在Dubbo中Invoker就是一個具有調(diào)用功能的對象，在服務(wù)提供端就是實際的服務(wù)實現(xiàn)，只是將服務(wù)實現(xiàn)封裝起來變成一個Invoker。

在服務(wù)消費端，從注冊中心得到服務(wù)提供者的信息之后，將一條條信息封裝為Invoker，這個Invoker就具備了遠(yuǎn)程調(diào)用的能力。

綜上，Dubbo就是創(chuàng)建了一個統(tǒng)一的模型，將可調(diào)用（可執(zhí)行體）的服務(wù)對象都統(tǒng)一封裝為Invoker。

而ClusterInvoker就是將多個服務(wù)引入的Invoker封裝起來，對外統(tǒng)一暴露一個Invoker，并且賦予這些Invoker集群容錯的功能。

服務(wù)目錄

服務(wù)目錄，即Directory，實際上它就是多個Invoker的集合，服務(wù)提供端一般都會集群分布，同樣的服務(wù)會有多個提供者，因此需要一個服務(wù)目錄來統(tǒng)一存放它們，需要調(diào)用服務(wù)的時候便從這個服務(wù)目錄中進(jìn)行挑選。

同時服務(wù)目錄還是實現(xiàn)了NotifyListener接口，當(dāng)集群中新增了一臺服務(wù)提供者或者下線了一臺服務(wù)提供者，目錄都會對服務(wù)提供者進(jìn)行更新，新增或者刪除對應(yīng)的Invoker。

從上圖中，可以看到用了一個抽象類AbstractDirectory來實現(xiàn)Directory接口，抽象類中運用到了模板方法模式，將一些公共方法和邏輯寫好，作為一個骨架，然后具體實現(xiàn)由了兩個子類來完成，兩個子類分別為StaticDirectory和RegistryDirectory。

RegistryDirectory

RegistryDirectory實現(xiàn)了NotifyListener接口，可以監(jiān)聽注冊中心的變化，當(dāng)注冊中心配置發(fā)生變化時，服務(wù)目錄也可以收到變更通知，然后根據(jù)更新之后的配置刷新Invoker列表。

由此可知RegistryDirectory共有三個作用：

獲取Invoker列表監(jiān)聽注冊中心刷新Invoker列表

獲取Invoker列表

RegistryDirectory實現(xiàn)了父類AbstractDirectory的抽象方法doList()，該方法可以得到Invoker列表

publicListInvokerTdoList(Invocationinvocation){

if(this.forbidden){

thrownewRpcException(....);

}else{

ListInvokerTinvokers=null;

MapString,ListInvokerTlocalMethodInvokerMap=this.methodInvokerMap;//獲取方法調(diào)用名和Invoker的映射表

if(localMethodInvokerMap!=nulllocalMethodInvokerMap.size()0){

StringmethodName=RpcUtils.getMethodName(invocation);

Object[]args=RpcUtils.getArguments(invocation);

//以下就是根據(jù)方法名和方法參數(shù)獲取可調(diào)用的Invoker

if(args!=nullargs.length0args[0]!=null(args[0]instanceofString||args[0].getClass().isEnum())){

invokers=(List)localMethodInvokerMap.get(methodName+"."+args[0]);

if(invokers==null){

invokers=(List)localMethodInvokerMap.get(methodName);

if(invokers==null){

invokers=(List)localMethodInvokerMap.get("*");

if(invokers==null){

IteratorListInvokerTiterator=localMethodInvokerMap.values().iterator();

if(iterator.hasNext()){

invokers=(List)iterator.next();

return(List)(invokers==nullnewArrayList(0):invokers);

監(jiān)聽注冊中心

通過實現(xiàn)NotifyListener接口可以感知注冊中心的數(shù)據(jù)變更。

RegistryDirectory定義了三個集合invokerUrlsrouterUrlsconfiguratorUrls分別處理對應(yīng)的配置然后轉(zhuǎn)化成對象。

publicsynchronizedvoidnotify(ListURLurls){

ListURLinvokerUrls=newArrayList();

ListURLrouterUrls=newArrayList();

ListURLconfiguratorUrls=newArrayList();

Iteratori$=urls.iterator();

while(true){

while(true){

while(i$.hasNext()){

//....根據(jù)urls填充上述三個列表

if(configuratorUrls!=null!configuratorUrls.isEmpty()){

this.configurators=toConfigurators(configuratorUrls);//根據(jù)urls轉(zhuǎn)化為configurators配置

ListlocalConfigurators;

if(routerUrls!=null!routerUrls.isEmpty()){

localConfigurators=this.toRouters(routerUrls);

if(localConfigurators!=null){

this.setRouters(localConfigurators);//根據(jù)urls轉(zhuǎn)化為routers配置

localConfigurators=this.configurators;

this.overrideDirectoryUrl=this.directoryUrl;

Configuratorconfigurator;

if(localConfigurators!=null!localConfigurators.isEmpty()){

for(Iteratori$=localConfigurators.iterator();i$.hasNext();this.overrideDirectoryUrl=configurator.configure(this.overrideDirectoryUrl)){

configurator=(Configurator)i$.next();

this.refreshInvoker(invokerUrls);//根據(jù)invokerUrls刷新invoker列表

return;

刷新Invoker列表

privatevoidrefreshInvoker(ListURLinvokerUrls){

//如果invokerUrls只有一個URL并且協(xié)議是empty，那么清除所有invoker

if(invokerUrls!=nullinvokerUrls.size()==1invokerUrls.get(0)!=null"empty".equals(((URL)invokerUrls.get(0)).getProtocol())){

this.forbidden=true;

this.methodInvokerMap=null;

this.destroyAllInvokers();

}else{

this.forbidden=false;

MapString,InvokerToldUrlInvokerMap=this.urlInvokerMap;//獲取舊的Invoker列表

if(invokerUrls.isEmpty()this.cachedInvokerUrls!=null){

invokerUrls.addAll(this.cachedInvokerUrls);

}else{

this.cachedInvokerUrls=newHashSet();

this.cachedInvokerUrls.addAll(invokerUrls);

if(invokerUrls.isEmpty()){

return;

//根據(jù)URL生成InvokerMap

MapString,InvokerTnewUrlInvokerMap=this.toInvokers(invokerUrls);

//根據(jù)新的InvokerMap生成方法名和Invoker列表對應(yīng)的Map

MapString,ListInvokerTnewMethodInvokerMap=this.toMethodInvokers(newUrlInvokerMap);

if(newUrlInvokerMap==null||newUrlInvokerMap.size()==0){

logger.error(newIllegalStateException("urlstoinvokerserror.invokerUrls.size:"+invokerUrls.size()+",invoker.size:0.urls:"+invokerUrls.toString()));

return;

this.methodInvokerMap=this.multiGroupthis.toMergeMethodInvokerMap(newMethodInvokerMap):newMethodInvokerMap;

this.urlInvokerMap=newUrlInvokerMap;

try{

this.destroyUnusedInvokers(oldUrlInvokerMap,newUrlInvokerMap);//銷毀無效的Invoker

}catch(Exceptionvar6){

logger.warn("destroyUnusedInvokerserror.",var6);

上述操作就是根據(jù)invokerUrls數(shù)量以及協(xié)議頭是否為empty來判斷是否禁用所有invokers，如果不禁用的話將invokerUrls轉(zhuǎn)化為Invoker，并且得到url,Invoker的映射關(guān)系。

再進(jìn)一步進(jìn)行轉(zhuǎn)化，得到methodName,List的映射關(guān)系，再將同一組的Invoker進(jìn)行合并，將合并結(jié)果賦值給methodInvokerMap，這個methodInvokerMap就是在doList中使用到的Map。

最后刷新InvokerMap，銷毀無效的Invoker。

StaticDirectory

StaticDirectory是靜態(tài)目錄，所有Invoker是固定的不會刪減的，并且所有Invoker由構(gòu)造器來傳入。

內(nèi)部邏輯也相當(dāng)簡單，只定義了一個列表用于存儲Invokers。實現(xiàn)父類的方法也只是將這些Invokers原封不動地返回。

privatefinalListInvokerTinvokers;

protectedListInvokerTdoList(Invocationinvocation)throwsRpcException{

returnthis.invokers;

服務(wù)路由

服務(wù)路由規(guī)定了服務(wù)消費者可以調(diào)用哪些服務(wù)提供者，Dubbo常用的是條件路由ConditionRouter。

條件路由由兩個條件組成，格式為[服務(wù)消費者匹配條件]=[服務(wù)提供者匹配條件]，例如5=9規(guī)定了只有IP為5的服務(wù)消費者才可以訪問IP為9的服務(wù)提供者，不可以調(diào)用其他的服務(wù)。

路由一樣是通過RegistryDirectory中的notify()更新的，在調(diào)用toMethodInvokers()的時候會進(jìn)行服務(wù)器級別的路由和方法級別的路由。

Cluster

在前面的流程中我們已經(jīng)通過Directory獲取了服務(wù)目錄，并且通過路由獲取了一個或多個Invoker，但是對于服務(wù)消費者還是需要進(jìn)行選擇，篩選出一個Invoker進(jìn)行調(diào)用。

Dubbo默認(rèn)的Cluster實現(xiàn)有多種，如下：

FailoverClusterFailfastClusterFailsafeClusterFailbackClusterBroadcastClusterAvailableCluster

每個Cluster內(nèi)部返回的都是xxxClusterInvoker，例如FailoverCluster：

publicclassFailoverClusterimplementsCluster{

publicstaticfinalStringNAME="failover";

publicFailoverCluster(){

publicTInvokerTjoin(DirectoryTdirectory)throwsRpcException{

returnnewFailoverClusterInvoker(directory);

FailoverClusterInvoker

FailoverClusterInvoker實現(xiàn)的功能是失敗調(diào)用(有重試次數(shù))自動切換。

publicResultdoInvoke(Invocationinvocation,ListInvokerTinvokers,LoadBalanceloadbalance)throwsRpcException{

ListInvokerTcopyinvokers=invokers;

this.checkInvokers(invokers,invocation);

//重試次數(shù)

intlen=this.getUrl().getMethodParameter(invocation.getMethodName(),"retries",2)+1;

if(len=0){

len=1;

RpcExceptionle=null;

ListInvokerTinvoked=newArrayList(invokers.size());

SetStringproviders=newHashSet(len);

//根據(jù)重試次數(shù)循環(huán)調(diào)用

for(inti=0;ilen;++i){

if(i0){

this.checkWhetherDestroyed();

copyinvokers=this.list(invocation);

this.checkInvokers(copyinvokers,invocation);

//負(fù)載均衡篩選出一個Invoker作本次調(diào)用

InvokerTinvoker=this.select(loadbalance,invocation,copyinvokers,invoked);

//將使用過的Invoker保存起來，下次重試時做過濾用

invoked.add(invoker);

//記錄到上下文中

RpcContext.getContext().setInvokers(invoked);

try{

//發(fā)起調(diào)用

Resultresult=invoker.invoke(invocation);

if(le!=nulllogger.isWarnEnabled()){

logger.warn("....");

Resultvar12=result;

returnvar12;

}catch(RpcExceptionvar17){//catch異常繼續(xù)下次循環(huán)重試

if(var17.isBiz()){

throwvar17;

le=var17;

}catch(Throwablevar18){

le=newRpcException(var18.getMessage(),var18);

}finally{

providers.add(invoker.getUrl().getAddress());

thrownewRpcException(....);

上述方法中，首先獲取重試次數(shù)len，根據(jù)重試次數(shù)進(jìn)行循環(huán)調(diào)用，調(diào)用發(fā)生異常會被catch住，然后重新調(diào)用。

每次循環(huán)會通過負(fù)載均衡選出一個Invoker，然后利用這個Invoker進(jìn)行遠(yuǎn)程調(diào)用，每次選出的Invoker會記錄下來，在下次調(diào)用的select()中會將使用上次調(diào)用的Invoker進(jìn)行重試，如果上一次沒有調(diào)用或者上次調(diào)用的Invoker下線了，那么會重新進(jìn)行負(fù)載均衡進(jìn)行選擇。

FailfastClusterInvoker

FailfastClusterInvoker只會進(jìn)行一次遠(yuǎn)程調(diào)用，如果失敗后立馬拋出異常。

publicResultdoInvoke(Invocationinvocation,ListInvokerTinvokers,LoadBalanceloadbalance)throwsRpcException{

this.checkInvokers(invokers,invocation);

Invokerinvoker=this.select(loadbalance,invocation,invokers,(List)null);//負(fù)載均衡選擇Invoker

try{

returninvoker.invoke(invocation);//發(fā)起遠(yuǎn)程調(diào)用

}catch(Throwablevar6){//失敗調(diào)用直接將錯誤拋出

if(var6instanceofRpcException((RpcException)var6).isBiz()){

throw(RpcException)var6;

}else{

thrownewRpcException(....);

FailsafeClusterInvoker

FailsafeClusterInvoker是一種安全失敗的cluster，調(diào)用發(fā)生錯誤僅僅是記錄一下日志，然后就返回了空結(jié)果。

publicResultdoInvoke(Invocationinvocation,ListInvokerTinvokers,LoadBalanceloadbalance)throwsRpcException{

try{

this.checkInvokers(invokers,invocation);

//負(fù)載均衡選出Invoker后直接進(jìn)行調(diào)用

InvokerTinvoker=this.select(loadbalance,invocation,invokers,(List)null);

returninvoker.invoke(invocation);

}catch(Throwablevar5){//調(diào)用錯誤只是打印日志

logger.error("Failsafeignoreexception:"+var5.getMessage(),var5);

returnnewRpcResult();

FailbackClusterInvoker

FailbackClusterInvoker調(diào)用失敗后，會記錄下本次調(diào)用，然后返回一個空結(jié)果給服務(wù)消費者，并且會通過一個定時任務(wù)對失敗的調(diào)用進(jìn)行重試。適用于執(zhí)行消息通知等最大努力場景。

protectedResultdoInvoke(Invocationinvocation,ListInvokerTinvokers,LoadBalanceloadbalance)throwsRpcException{

try{

this.checkInvokers(invokers,invocation);

//負(fù)載均衡選出Invoker

InvokerTinvoker=this.select(loadbalance,invocation,invokers,(List)null);

//執(zhí)行調(diào)用，執(zhí)行成功返回調(diào)用結(jié)果

returninvoker.invoke(invocation);

}catch(Throwablevar5){

//調(diào)用失敗

logger.error("....");

//記錄下本次失敗調(diào)用

this.addFailed(invocation,this);

//返回空結(jié)果

returnnewRpcResult();

privatevoidaddFailed(Invocationinvocation,AbstractClusterInvokerrouter){

if(this.retryFuture==null){

synchronized(this){

//如果未創(chuàng)建重試本次調(diào)用的定時任務(wù)

if(this.retryFuture==null){

//創(chuàng)建定時任務(wù)

this.retryFuture=this.scheduledExecutorService.scheduleWithFixedDelay(newRunnable(){

publicvoidrun(){

try{

//定時進(jìn)行重試

FailbackClusterInvoker.this.retryFailed();

}catch(Throwablevar2){

FailbackClusterInvoker.logger.error("....",var2);

},5000L,5000L,TimeUnit.MILLISECONDS);

//將invocation和router存入map

this.failed.put(invocation,router);

voidretryFailed(){

if(this.failed.size()!=0){

Iteratori$=(newHashMap(this.failed)).entrySet().iterator();

while(i$.hasNext()){

EntryInvocation,AbstractClusterInvokerentry=(Entry)i$.next();

Invocationinvocation=(Invocation)entry.getKey();

Invokerinvoker=(Invoker)entry.getValue();

try{

//進(jìn)行重試調(diào)用

invoker.invoke(invocation);

//調(diào)用成功未產(chǎn)生異常則移除本次失敗調(diào)用的記錄，銷毀定時任務(wù)

this.failed.remove(invocation);

}catch(Throwablevar6){

logger.error("....",var6);

邏輯比較簡單，大致就是當(dāng)調(diào)用錯誤時返回空結(jié)果，并記錄下本次失敗調(diào)用到failedinvocation,router中，并且會創(chuàng)建一個定時任務(wù)定時地去調(diào)用failed中記錄的失敗調(diào)用，如果調(diào)用成功了就從failed中移除這個調(diào)用。

ForkingClusterInvoker

ForkingClusterInvoker運行時，會將所有Invoker都放入線程池中并發(fā)調(diào)用，只要有一個Invoker調(diào)用成功了就返回結(jié)果，doInvoker方法立即停止運行。

適用于對實時性比較高的讀寫操作。

publicResultdoInvoke(finalInvocationinvocation,ListInvokerTinvokers,LoadBalanceloadbalance)throwsRpcException{

Resultvar19;

try{

this.checkInvokers(invokers,invocation);

intforks=this.getUrl().getParameter("forks",2);

inttimeout=this.getUrl().getParameter("timeout",1000);

finalObjectselected;

if(forks0forksinvokers.size()){

selected=newArrayList();

for(inti=0;iforks;++i){

InvokerTinvoker=this.select(loadbalance,invocation,invokers,(List)selected);

if(!((List)selected).contains(invoker)){

//選擇好的Invoker放入這個selected列表

((List)selected).add(invoker);

}else{

selected=invokers;

RpcContext.getContext().setInvokers((List)selected);

finalAtomicIntegercount=newAtomicInteger();

//阻塞隊列

finalBlockingQueueObjectref=newLinkedBlockingQueue();

Iteratori$=((List)selected).iterator();

while(i$.hasNext()){

finalInvokerTinvoker=(Invoker)i$.next();

this.executor.execute(newRunnable(){

publicvoidrun(){

try{

Resultresult=invoker.invoke(invocation);

ref.offer(result);

}catch(Throwablevar3){

intvalue=count.incrementAndGet();

if(value=((List)selected).size()){//等待所有調(diào)用都產(chǎn)生異常才入隊

ref.offer(var3);

try{

//阻塞獲取結(jié)果

Objectret=ref.poll((long)timeout,TimeUnit.MILLISECONDS);

if(retinstanceofThrowable){

Throwablee=(Throwable)ret;

thrownewRpcException(....);

var19=(Result)ret;

}catch(InterruptedExceptionvar14){

thrownewRpcException(....);

}finally{

RpcContext.getContext().clearAttachments();

returnvar19;

BroadcastClusterInvoker

BroadcastClusterInvoker運行時會將所有Invoker逐個調(diào)用，在最后判斷中如果有一個調(diào)用產(chǎn)生錯誤，則拋出異常。

適用于通知所有提供者更新緩存或日志等本地資源的場景。

publicResultdoInvoke(Invocationinvocation,ListInvokerTinvokers,LoadBalanceloadbalance)throwsRpcException{

this.checkInvokers(invokers,invocation);

RpcContext.getContext().setInvokers(invokers);

RpcExceptionexception=null;

Resultresult=null;

Iteratori$=invokers.iterator();

while(i$.hasNext()){

Invokerinvoker=(Invoker)i$.next();

try{

result=invoker.invoke(invocation);

}catch(RpcExceptionvar9){

exception=var9;

logger.warn(var9.getMessage(),var9);

}catch(Throwablevar10){

exception=newRpcException(var10.getMessage(),var10);

logger.warn(var10.getMessage(),var10);

//如果調(diào)用過程中發(fā)生過錯誤拋出異常

if(exception!=null){

throwexception;

}else{

//返回調(diào)用結(jié)果

returnresult;

AbstractClusterInvoker

AbstractClusterInvoker是上述所有類的父類，內(nèi)部結(jié)構(gòu)較為簡單。AvailableCluster內(nèi)部返回結(jié)果就是AvailableClusterInvoker。

publicclassAvailableClusterimplementsCluster{

publicstaticfinalStringNAME="available";

publicAvailableCluster(){

publicTInvokerTjoin(DirectoryTdirectory)throwsRpcException{

returnnewAbstractClusterInvokerT(directory){

publicResultdoInvoke(Invocationinvocation,ListInvokerTinvokers,LoadBalanceloadbalance)throwsRpcException{

Iteratori$=invokers.iterator();

Invokerinvoker;

do{//循環(huán)判斷：哪個invoker能用就調(diào)用哪個

if(!i$.hasNext()){

thrownewRpcException("Noprovideravailablein"+invokers);

invoker=(Invoker)i$.next();

}while(!invoker.isAvailable());

returninvoker.invoke(invocation);

小結(jié)

上述中有很多種集群的實現(xiàn)，各適用于不同的場景，加了Cluster這個中間層，向服務(wù)消費者屏蔽了集群調(diào)用的細(xì)節(jié)，并且支持不同場景使用不同的模式。

負(fù)載均衡

Dubbo中的負(fù)載均衡，即LoadBalance，服務(wù)提供者一般都是集群分布，所以需要Dubbo選擇出合適的服務(wù)提供者來給服務(wù)消費者調(diào)用。

Dubbo中提供了多種負(fù)載均衡算法：

RandomLoadBalanceLeastActiveLoadBalanceConsistentHashLoadBalanceRoundRobinLoadBalance

AbstractLoadBalance

實現(xiàn)類都繼承了于這個類，該類實現(xiàn)了LoadBalance，使用模板方法模式，將一些公用的邏輯封裝好，而具體的實現(xiàn)由子類自定義。

publicTInvokerTselect(ListInvokerTinvokers,URLurl,Invocationinvocation){

if(invokers!=null!invokers.isEmpty()){

//子類實現(xiàn)

returninvokers.size()==1(Invoker)invokers.get(0):this.doSelect(invokers,url,invocation);

}else{

returnnull;

protectedabstractTInvokerTdoSelect(ListInvokerTvar1,URLvar2,Invocationvar3);

服務(wù)剛啟動需要預(yù)熱，不能突然讓服務(wù)負(fù)載過高，需要進(jìn)行服務(wù)的降權(quán)。

protectedintgetWeight(Invokerinvoker,Invocationinvocation){

intweight=invoker.getUrl().getMethodParameter(invocation.getMethodName(),"weight",100);//獲得權(quán)重

if(weight0){

longtimestamp=invoker.getUrl().getParameter("remote.timestamp",0L);//啟動時間

if(timestamp0L){

intuptime=(int)(System.currentTimeMillis()-timestamp);//計算已啟動時長

intwarmup=invoker.getUrl().getParameter("warmup",600000);

if(uptime0uptimewarmup){

weight=calculateWarmupWeight(uptime,warmup,weig