前言
本文繼續(xù)分析dubbo的cluster層，此層封裝多個(gè)提供者的路由及負(fù)載均衡，并橋接注冊(cè)中心，以Invoker為中心，擴(kuò)展接口為Cluster, Directory, Router, LoadBalance；
Cluster接口
整個(gè)cluster層可以使用如下圖片概括：
各節(jié)點(diǎn)關(guān)系：
這里的Invoker是Provider的一個(gè)可調(diào)用Service的抽象，Invoker封裝了Provider地址及Service接口信息；
Directory代表多個(gè)Invoker，可以把它看成List，但與List不同的是，它的值可能是動(dòng)態(tài)變化的，比如注冊(cè)中心推送變更；
Cluster將Directory中的多個(gè)Invoker偽裝成一個(gè) Invoker，對(duì)上層透明，偽裝過(guò)程包含了容錯(cuò)邏輯，調(diào)用失敗后，重試另一個(gè)；
Router負(fù)責(zé)從多個(gè)Invoker中按路由規(guī)則選出子集，比如讀寫(xiě)分離，應(yīng)用隔離等；
LoadBalance負(fù)責(zé)從多個(gè)Invoker中選出具體的一個(gè)用于本次調(diào)用，選的過(guò)程包含了負(fù)載均衡算法，調(diào)用失敗后，需要重選；
Cluster經(jīng)過(guò)目錄，路由，負(fù)載均衡獲取到一個(gè)可用的Invoker，交給上層調(diào)用，接口如下：
@SPI(FailoverCluster.NAME)
public interface Cluster {
? ? /**
? ? * Merge the directory invokers to a virtual invoker.
? ? *
? ? * @param <T>
? ? * @param directory
? ? * @return cluster invoker
? ? * @throws RpcException
? ? */
? ? @Adaptive
? ? <T> Invoker<T> join(Directory<T> directory) throws RpcException;
}
Cluster是一個(gè)集群容錯(cuò)接口，經(jīng)過(guò)路由，負(fù)載均衡之后獲取的Invoker，由容錯(cuò)機(jī)制來(lái)處理，dubbo提供了多種容錯(cuò)機(jī)制包括：
Failover Cluster：失敗自動(dòng)切換，當(dāng)出現(xiàn)失敗，重試其它服務(wù)器 [1]。通常用于讀操作，但重試會(huì)帶來(lái)更長(zhǎng)延遲?？赏ㄟ^(guò) retries=”2″ 來(lái)設(shè)置重試次數(shù)(不含第一次)。
Failfast Cluster：快速失敗，只發(fā)起一次調(diào)用，失敗立即報(bào)錯(cuò)。通常用于非冪等性的寫(xiě)操作，比如新增記錄。
Failsafe Cluster：失敗安全，出現(xiàn)異常時(shí)，直接忽略。通常用于寫(xiě)入審計(jì)日志等操作。
Failback Cluster：失敗自動(dòng)恢復(fù)，后臺(tái)記錄失敗請(qǐng)求，定時(shí)重發(fā)。通常用于消息通知操作。
Forking Cluster：并行調(diào)用多個(gè)服務(wù)器，只要一個(gè)成功即返回。通常用于實(shí)時(shí)性要求較高的讀操作，但需要浪費(fèi)更多服務(wù)資源?？赏ㄟ^(guò) forks=”2″ 來(lái)設(shè)置最大并行數(shù)。
Broadcast Cluster：廣播調(diào)用所有提供者，逐個(gè)調(diào)用，任意一臺(tái)報(bào)錯(cuò)則報(bào)錯(cuò) [2]。通常用于通知所有提供者更新緩存或日志等本地資源信息。
默認(rèn)使用了FailoverCluster，失敗的時(shí)候會(huì)默認(rèn)重試其他服務(wù)器，默認(rèn)為兩次；當(dāng)然也可以擴(kuò)展其他的容錯(cuò)機(jī)制；看一下默認(rèn)的FailoverCluster容錯(cuò)機(jī)制，具體源碼在FailoverClusterInvoker中：
public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
? ? ? ? List<Invoker<T>> copyinvokers = invokers;
? ? ? ? checkInvokers(copyinvokers, invocation);
? ? ? ? int len = getUrl().getMethodParameter(invocation.getMethodName(), Constants.RETRIES_KEY, Constants.DEFAULT_RETRIES) + 1;
? ? ? ? if (len <= 0) {
? ? ? ? ? ? len = 1;
? ? ? ? }
? ? ? ? // retry loop.
? ? ? ? RpcException le = null; // last exception.
? ? ? ? List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(copyinvokers.size()); // invoked invokers.
? ? ? ? Set<String> providers = new HashSet<String>(len);
? ? ? ? for (int i = 0; i < len; i++) {
? ? ? ? ? ? //Reselect before retry to avoid a change of candidate `invokers`.
? ? ? ? ? ? //NOTE: if `invokers` changed, then `invoked` also lose accuracy.
? ? ? ? ? ? if (i > 0) {
? ? ? ? ? ? ? ? checkWhetherDestroyed();
? ? ? ? ? ? ? ? copyinvokers = list(invocation);
? ? ? ? ? ? ? ? // check again
? ? ? ? ? ? ? ? checkInvokers(copyinvokers, invocation);
? ? ? ? ? ? }
? ? ? ? ? ? Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
? ? ? ? ? ? invoked.add(invoker);
? ? ? ? ? ? RpcContext.getContext().setInvokers((List) invoked);
? ? ? ? ? ? try {
? ? ? ? ? ? ? ? Result result = invoker.invoke(invocation);
? ? ? ? ? ? ? ? if (le != null && logger.isWarnEnabled()) {
? ? ? ? ? ? ? ? ? ? logger.warn("Although retry the method " + invocation.getMethodName()
? ? ? ? ? ? ? ? ? ? ? ? ? ? + " in the service " + getInterface().getName()
? ? ? ? ? ? ? ? ? ? ? ? ? ? + " was successful by the provider " + invoker.getUrl().getAddress()
? ? ? ? ? ? ? ? ? ? ? ? ? ? + ", but there have been failed providers " + providers
? ? ? ? ? ? ? ? ? ? ? ? ? ? + " (" + providers.size() + "/" + copyinvokers.size()
? ? ? ? ? ? ? ? ? ? ? ? ? ? + ") from the registry " + directory.getUrl().getAddress()
? ? ? ? ? ? ? ? ? ? ? ? ? ? + " on the consumer " + NetUtils.getLocalHost()
? ? ? ? ? ? ? ? ? ? ? ? ? ? + " using the dubbo version " + Version.getVersion() + ". Last error is: "
? ? ? ? ? ? ? ? ? ? ? ? ? ? + le.getMessage(), le);
? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? return result;
? ? ? ? ? ? } catch (RpcException e) {
? ? ? ? ? ? ? ? if (e.isBiz()) { // biz exception.
? ? ? ? ? ? ? ? ? ? throw e;
? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? le = e;
? ? ? ? ? ? } catch (Throwable e) {
? ? ? ? ? ? ? ? le = new RpcException(e.getMessage(), e);
? ? ? ? ? ? } finally {
? ? ? ? ? ? ? ? providers.add(invoker.getUrl().getAddress());
? ? ? ? ? ? }
? ? ? ? }
? ? ? ? throw new RpcException(le != null ? le.getCode() : 0, "Failed to invoke the method "
? ? ? ? ? ? ? ? + invocation.getMethodName() + " in the service " + getInterface().getName()
? ? ? ? ? ? ? ? + ". Tried " + len + " times of the providers " + providers
? ? ? ? ? ? ? ? + " (" + providers.size() + "/" + copyinvokers.size()
? ? ? ? ? ? ? ? + ") from the registry " + directory.getUrl().getAddress()
? ? ? ? ? ? ? ? + " on the consumer " + NetUtils.getLocalHost() + " using the dubbo version "
? ? ? ? ? ? ? ? + Version.getVersion() + ". Last error is: "
? ? ? ? ? ? ? ? + (le != null ? le.getMessage() : ""), le != null && le.getCause() != null ? le.getCause() : le);
? ? }
invocation是客戶端傳給服務(wù)器的相關(guān)參數(shù)包括(方法名稱，方法參數(shù)，參數(shù)值，附件信息)，invokers是經(jīng)過(guò)路由之后的服務(wù)器列表，loadbalance是指定的負(fù)載均衡策略；首先檢查invokers是否為空，為空直接拋異常，然后獲取重試的次數(shù)默認(rèn)為2次，接下來(lái)就是循環(huán)調(diào)用指定次數(shù)，如果不是第一次調(diào)用(表示第一次調(diào)用失敗)，會(huì)重新加載服務(wù)器列表，然后通過(guò)負(fù)載均衡策略獲取唯一的Invoker，最后就是通過(guò)Invoker把invocation發(fā)送給服務(wù)器，返回結(jié)果Result；
具體的doInvoke方法是在抽象類AbstractClusterInvoker中被調(diào)用的：
public Result invoke(final Invocation invocation) throws RpcException {
? ? ? ? checkWhetherDestroyed();
? ? ? ? LoadBalance loadbalance = null;
? ? ? ? List<Invoker<T>> invokers = list(invocation);
? ? ? ? if (invokers != null && !invokers.isEmpty()) {
? ? ? ? ? ? loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()
? ? ? ? ? ? ? ? ? ? .getMethodParameter(RpcUtils.getMethodName(invocation), Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE));
? ? ? ? }
? ? ? ? RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
? ? ? ? return doInvoke(invocation, invokers, loadbalance);
? ? }
protected List<Invoker<T>> list(Invocation invocation) throws RpcException {
? ? ? ? List<Invoker<T>> invokers = directory.list(invocation);
? ? ? ? return invokers;
? ? }
首先通過(guò)Directory獲取Invoker列表，同時(shí)在Directory中也會(huì)做路由處理，然后獲取負(fù)載均衡策略，最后調(diào)用具體的容錯(cuò)策略；下面具體看一下Directory；
Directory接口
接口定義如下：
public interface Directory<T> extends Node {
? ? /**
? ? * get service type.
? ? *
? ? * @return service type.
? ? */
? ? Class<T> getInterface();
? ? /**
? ? * list invokers.
? ? *
? ? * @return invokers
? ? */
? ? List<Invoker<T>> list(Invocation invocation) throws RpcException;
}
目錄服務(wù)作用就是獲取指定接口的服務(wù)列表，具體實(shí)現(xiàn)有兩個(gè)：StaticDirectory和RegistryDirectory，同時(shí)都繼承于AbstractDirectory；從名字可以大致知道StaticDirectory是一個(gè)固定的目錄服務(wù)，表示里面的Invoker列表不會(huì)動(dòng)態(tài)改變；RegistryDirectory是一個(gè)動(dòng)態(tài)的目錄服務(wù)，通過(guò)注冊(cè)中心動(dòng)態(tài)更新服務(wù)列表；list實(shí)現(xiàn)在抽象類中：
public List<Invoker<T>> list(Invocation invocation) throws RpcException {
? ? ? ? if (destroyed) {
? ? ? ? ? ? throw new RpcException("Directory already destroyed .url: " + getUrl());
? ? ? ? }
? ? ? ? List<Invoker<T>> invokers = doList(invocation);
? ? ? ? List<Router> localRouters = this.routers; // local reference
? ? ? ? if (localRouters != null && !localRouters.isEmpty()) {
? ? ? ? ? ? for (Router router : localRouters) {
? ? ? ? ? ? ? ? try {
? ? ? ? ? ? ? ? ? ? if (router.getUrl() == null || router.getUrl().getParameter(Constants.RUNTIME_KEY, false)) {
? ? ? ? ? ? ? ? ? ? ? ? invokers = router.route(invokers, getConsumerUrl(), invocation);
? ? ? ? ? ? ? ? ? ? }
? ? ? ? ? ? ? ? } catch (Throwable t) {
? ? ? ? ? ? ? ? ? ? logger.error("Failed to execute router: " + getUrl() + ", cause: " + t.getMessage(), t);
? ? ? ? ? ? ? ? }
? ? ? ? ? ? }
? ? ? ? }
? ? ? ? return invokers;
? ? }
首先檢查目錄是否被銷毀，然后調(diào)用doList，具體在實(shí)現(xiàn)類中定義，最后調(diào)用路由功能，下面重點(diǎn)看一下StaticDirectory和RegistryDirectory中的doList方法
1.RegistryDirectory
是一個(gè)動(dòng)態(tài)的目錄服務(wù)，所有可以看到RegistryDirectory同時(shí)也繼承了NotifyListener接口，是一個(gè)通知接口，注冊(cè)中心有服務(wù)列表更新的時(shí)候，同時(shí)通知RegistryDirectory，通知邏輯如下：
public synchronized void notify(List<URL> urls) {
? ? ? ? List<URL> invokerUrls = new ArrayList<URL>();
? ? ? ? List<URL> routerUrls = new ArrayList<URL>();
? ? ? ? List<URL> configuratorUrls = new ArrayList<URL>();
? ? ? ? for (URL url : urls) {
? ? ? ? ? ? String protocol = url.getProtocol();
? ? ? ? ? ? String category = url.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
? ? ? ? ? ? if (Constants.ROUTERS_CATEGORY.equals(category)
? ? ? ? ? ? ? ? ? ? || Constants.ROUTE_PROTOCOL.equals(protocol)) {
? ? ? ? ? ? ? ? routerUrls.add(url);
? ? ? ? ? ? } else if (Constants.CONFIGURATORS_CATEGORY.equals(category)
? ? ? ? ? ? ? ? ? ? || Constants.OVERRIDE_PROTOCOL.equals(protocol)) {
? ? ? ? ? ? ? ? configuratorUrls.add(url);
? ? ? ? ? ? } else if (Constants.PROVIDERS_CATEGORY.equals(category)) {
? ? ? ? ? ? ? ? invokerUrls.add(url);
? ? ? ? ? ? } else {
? ? ? ? ? ? ? ? logger.warn("Unsupported category " + category + " in notified url: " + url + " from registry " + getUrl().getAddress() + " to consumer " + NetUtils.getLocalHost());
? ? ? ? ? ? }
? ? ? ? }
? ? ? ? // configurators
? ? ? ? if (configuratorUrls != null && !configuratorUrls.isEmpty()) {
? ? ? ? ? ? this.configurators = toConfigurators(configuratorUrls);
? ? ? ? }
? ? ? ? // routers
? ? ? ? if (routerUrls != null && !routerUrls.isEmpty()) {
? ? ? ? ? ? List<Router> routers = toRouters(routerUrls);
? ? ? ? ? ? if (routers != null) { // null - do nothing
? ? ? ? ? ? ? ? setRouters(routers);
? ? ? ? ? ? }
? ? ? ? }
? ? ? ? List<Configurator> localConfigurators = this.configurators; // local reference
? ? ? ? // merge override parameters
? ? ? ? this.overrideDirectoryUrl = directoryUrl;
? ? ? ? if (localConfigurators != null && !localConfigurators.isEmpty()) {
? ? ? ? ? ? for (Configurator configurator : localConfigurators) {
? ? ? ? ? ? ? ? this.overrideDirectoryUrl = configurator.configure(overrideDirectoryUrl);
? ? ? ? ? ? }
? ? ? ? }
? ? ? ? // providers
? ? ? ? refreshInvoker(invokerUrls);
? ? }
此通知接口會(huì)接受三種類別的url包括：router(路由)，configurator(配置)，provider(服務(wù)提供方)；
路由規(guī)則：決定一次dubbo服務(wù)調(diào)用的目標(biāo)服務(wù)器，分為條件路由規(guī)則和腳本路由規(guī)則，并且支持可擴(kuò)展，向注冊(cè)中心寫(xiě)入路由規(guī)則的操作通常由監(jiān)控中心或治理中心的頁(yè)面完成；
配置規(guī)則：向注冊(cè)中心寫(xiě)入動(dòng)態(tài)配置覆蓋規(guī)則 [1]。該功能通常由監(jiān)控中心或治理中心的頁(yè)面完成；
provider：動(dòng)態(tài)提供的服務(wù)列表
路由規(guī)則和配置規(guī)則其實(shí)就是對(duì)provider服務(wù)列表更新和過(guò)濾處理，refreshInvoker方法就是根據(jù)三種url類別刷新本地的invoker列表，下面看一下RegistryDirectory實(shí)現(xiàn)的doList接口：
public List<Invoker<T>> doList(Invocation invocation) {
? ? ? ? if (forbidden) {
? ? ? ? ? ? // 1. No service provider 2. Service providers are disabled
? ? ? ? ? ? throw new RpcException(RpcException.FORBIDDEN_EXCEPTION,
? ? ? ? ? ? ? ? "No provider available from registry " + getUrl().getAddress() + " for service " + getConsumerUrl().getServiceKey() + " on consumer " +? NetUtils.getLocalHost()
? ? ? ? ? ? ? ? ? ? ? ? + " use dubbo version " + Version.getVersion() + ", please check status of providers(disabled, not registered or in blacklist).");
? ? ? ? }
? ? ? ? List<Invoker<T>> invokers = null;
? ? ? ? Map<String, List<Invoker<T>>> localMethodInvokerMap = this.methodInvokerMap; // local reference
? ? ? ? if (localMethodInvokerMap != null && localMethodInvokerMap.size() > 0) {
? ? ? ? ? ? String methodName = RpcUtils.getMethodName(invocation);
? ? ? ? ? ? Object[] args = RpcUtils.getArguments(invocation);
? ? ? ? ? ? if (args != null && args.length > 0 && args[0] != null
? ? ? ? ? ? ? ? ? ? && (args[0] instanceof String || args[0].getClass().isEnum())) {
? ? ? ? ? ? ? ? invokers = localMethodInvokerMap.get(methodName + "." + args[0]); // The routing can be enumerated according to the first parameter
? ? ? ? ? ? }
? ? ? ? ? ? if (invokers == null) {
? ? ? ? ? ? ? ? invokers = localMethodInvokerMap.get(methodName);
? ? ? ? ? ? }
? ? ? ? ? ? if (invokers == null) {
? ? ? ? ? ? ? ? invokers = localMethodInvokerMap.get(Constants.ANY_VALUE);
? ? ? ? ? ? }
? ? ? ? ? ? if (invokers == null) {
? ? ? ? ? ? ? ? Iterator<List<Invoker<T>>> iterator = localMethodInvokerMap.values().iterator();
? ? ? ? ? ? ? ? if (iterator.hasNext()) {
? ? ? ? ? ? ? ? ? ? invokers = iterator.next();
? ? ? ? ? ? ? ? }
? ? ? ? ? ? }
? ? ? ? }
? ? ? ? return invokers == null ? new ArrayList<Invoker<T>>(0) : invokers;
? ? }
refreshInvoker處理之后，服務(wù)列表已methodInvokerMap存在，一個(gè)方法對(duì)應(yīng)服務(wù)列表Map>>；
通過(guò)Invocation中指定的方法獲取對(duì)應(yīng)的服務(wù)列表，如果具體的方法沒(méi)有對(duì)應(yīng)的服務(wù)列表，則獲取”*”對(duì)應(yīng)的服務(wù)列表；處理完之后就在父類中進(jìn)行路由處理，路由規(guī)則同樣是通過(guò)通知接口獲取的，路由規(guī)則在下章介紹；
2.StaticDirectory
這是一個(gè)靜態(tài)的目錄服務(wù)，里面的服務(wù)列表在初始化的時(shí)候就已經(jīng)存在，并且不會(huì)改變；StaticDirectory用得比較少,主要用在服務(wù)對(duì)多注冊(cè)中心的引用；
protected List<Invoker<T>> doList(Invocation invocation) throws RpcException {
? ? ? ? return invokers;
? ? }
因?yàn)槭庆o態(tài)的，所有doList方法也很簡(jiǎn)單，直接返回內(nèi)存中的服務(wù)列表即可；
Router接口
路由規(guī)則決定一次dubbo服務(wù)調(diào)用的目標(biāo)服務(wù)器，分為條件路由規(guī)則和腳本路由規(guī)則，并且支持可擴(kuò)展，接口如下：
public interface Router extends Comparable<Router> {
? ? /**
? ? * get the router url.
? ? *
? ? * @return url
? ? */
? ? URL getUrl();
? ? /**
? ? * route.
? ? *
? ? * @param invokers
? ? * @param url? ? ? ? refer url
? ? * @param invocation
? ? * @return routed invokers
? ? * @throws RpcException
? ? */
? ? <T> List<Invoker<T>> route(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException;
}
接口中提供的route方法通過(guò)一定的規(guī)則過(guò)濾出invokers的一個(gè)子集；提供了三個(gè)實(shí)現(xiàn)類：ScriptRouter，ConditionRouter和MockInvokersSelector
ScriptRouter：腳本路由規(guī)則支持 JDK 腳本引擎的所有腳本，比如：javascript, jruby, groovy 等，通過(guò)type=javascript參數(shù)設(shè)置腳本類型，缺省為javascript；
ConditionRouter：基于條件表達(dá)式的路由規(guī)則，如：host = 10.20.153.10 => host = 10.20.153.11；=> 之前的為消費(fèi)者匹配條件，所有參數(shù)和消費(fèi)者的 URL 進(jìn)行對(duì)比，=> 之后為提供者地址列表的過(guò)濾條件，所有參數(shù)和提供者的 URL 進(jìn)行對(duì)比；
MockInvokersSelector：是否被配置為使用mock，此路由器保證只有具有協(xié)議MOCK的調(diào)用者出現(xiàn)在最終的調(diào)用者列表中，所有其他調(diào)用者將被排除；
下面重點(diǎn)看一下ScriptRouter源碼
public ScriptRouter(URL url) {
? ? ? ? this.url = url;
? ? ? ? String type = url.getParameter(Constants.TYPE_KEY);
? ? ? ? this.priority = url.getParameter(Constants.PRIORITY_KEY, 0);
? ? ? ? String rule = url.getParameterAndDecoded(Constants.RULE_KEY);
? ? ? ? if (type == null || type.length() == 0) {
? ? ? ? ? ? type = Constants.DEFAULT_SCRIPT_TYPE_KEY;
? ? ? ? }
? ? ? ? if (rule == null || rule.length() == 0) {
? ? ? ? ? ? throw new IllegalStateException(new IllegalStateException("route rule can not be empty. rule:" + rule));
? ? ? ? }
? ? ? ? ScriptEngine engine = engines.get(type);
? ? ? ? if (engine == null) {
? ? ? ? ? ? engine = new ScriptEngineManager().getEngineByName(type);
? ? ? ? ? ? if (engine == null) {
? ? ? ? ? ? ? ? throw new IllegalStateException(new IllegalStateException("Unsupported route rule type: " + type + ", rule: " + rule));
? ? ? ? ? ? }
? ? ? ? ? ? engines.put(type, engine);
? ? ? ? }
? ? ? ? this.engine = engine;
? ? ? ? this.rule = rule;
? ? }
構(gòu)造器分別初始化腳本引擎(engine)和腳本代碼(rule)，默認(rèn)的腳本引擎是javascript；看一個(gè)具體的url：
"script://0.0.0.0/com.foo.BarService?category=routers&dynamic=false&rule=" + URL.encode("（function route(invokers) { ... } (invokers)）")
script協(xié)議表示一個(gè)腳本協(xié)議，rule后面是一段javascript腳本，傳入的參數(shù)是invokers；
（function route(invokers) {
? ? var result = new java.util.ArrayList(invokers.size());
? ? for (i = 0; i < invokers.size(); i ++) {
? ? ? ? if ("10.20.153.10".equals(invokers.get(i).getUrl().getHost())) {
? ? ? ? ? ? result.add(invokers.get(i));
? ? ? ? }
? ? }
? ? return result;
} (invokers)）; // 表示立即執(zhí)行方法
如上這段腳本過(guò)濾出host為10.20.153.10，具體是如何執(zhí)行這段腳本的，在route方法中：
public <T> List<Invoker<T>> route(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException {
? ? ? ? try {
? ? ? ? ? ? List<Invoker<T>> invokersCopy = new ArrayList<Invoker<T>>(invokers);
? ? ? ? ? ? Compilable compilable = (Compilable) engine;
? ? ? ? ? ? Bindings bindings = engine.createBindings();
? ? ? ? ? ? bindings.put("invokers", invokersCopy);
? ? ? ? ? ? bindings.put("invocation", invocation);
? ? ? ? ? ? bindings.put("context", RpcContext.getContext());
? ? ? ? ? ? CompiledScript function = compilable.compile(rule);
? ? ? ? ? ? Object obj = function.eval(bindings);
? ? ? ? ? ? if (obj instanceof Invoker[]) {
? ? ? ? ? ? ? ? invokersCopy = Arrays.asList((Invoker<T>[]) obj);
? ? ? ? ? ? } else if (obj instanceof Object[]) {
? ? ? ? ? ? ? ? invokersCopy = new ArrayList<Invoker<T>>();
? ? ? ? ? ? ? ? for (Object inv : (Object[]) obj) {
? ? ? ? ? ? ? ? ? ? invokersCopy.add((Invoker<T>) inv);
? ? ? ? ? ? ? ? }
? ? ? ? ? ? } else {
? ? ? ? ? ? ? ? invokersCopy = (List<Invoker<T>>) obj;
? ? ? ? ? ? }
? ? ? ? ? ? return invokersCopy;
? ? ? ? } catch (ScriptException e) {
? ? ? ? ? ? //fail then ignore rule .invokers.
? ? ? ? ? ? logger.error("route error , rule has been ignored. rule: " + rule + ", method:" + invocation.getMethodName() + ", url: " + RpcContext.getContext().getUrl(), e);
? ? ? ? ? ? return invokers;
? ? ? ? }
? ? }
首先通過(guò)腳本引擎編譯腳本，然后執(zhí)行腳本，同時(shí)傳入Bindings參數(shù)，這樣在腳本中就可以獲取invokers，然后進(jìn)行過(guò)濾；最后來(lái)看一下負(fù)載均衡策略
LoadBalance接口
在集群負(fù)載均衡時(shí)，Dubbo提供了多種均衡策略，缺省為random隨機(jī)調(diào)用，可以自行擴(kuò)展負(fù)載均衡策略；接口類如下：
@SPI(RandomLoadBalance.NAME)
public interface LoadBalance {
? ? /**
? ? * select one invoker in list.
? ? *
? ? * @param invokers? invokers.
? ? * @param url? ? ? ? refer url
? ? * @param invocation invocation.
? ? * @return selected invoker.
? ? */
? ? @Adaptive("loadbalance")
? ? <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException;
}
SPI定義了默認(rèn)的策略為RandomLoadBalance，提供了一個(gè)select方法，通過(guò)策略從服務(wù)列表中選擇一個(gè)invoker；dubbo默認(rèn)提供了多種策略：
Random LoadBalance：隨機(jī)，按權(quán)重設(shè)置隨機(jī)概率，在一個(gè)截面上碰撞的概率高，但調(diào)用量越大分布越均勻，而且按概率使用權(quán)重后也比較均勻，有利于動(dòng)態(tài)調(diào)整提供者權(quán)重；
RoundRobin LoadBalance：輪詢，按公約后的權(quán)重設(shè)置輪詢比率；存在慢的提供者累積請(qǐng)求的問(wèn)題，比如：第二臺(tái)機(jī)器很慢，但沒(méi)掛，當(dāng)請(qǐng)求調(diào)到第二臺(tái)時(shí)就卡在那，
久而久之，所有請(qǐng)求都卡在調(diào)到第二臺(tái)上；
LeastActive LoadBalance：最少活躍調(diào)用數(shù)，相同活躍數(shù)的隨機(jī)，活躍數(shù)指調(diào)用前后計(jì)數(shù)差；使慢的提供者收到更少請(qǐng)求，因?yàn)樵铰奶峁┱叩恼{(diào)用前后計(jì)數(shù)差會(huì)越大；
ConsistentHash LoadBalance：一致性 Hash，相同參數(shù)的請(qǐng)求總是發(fā)到同一提供者；當(dāng)某一臺(tái)提供者掛時(shí)，原本發(fā)往該提供者的請(qǐng)求，基于虛擬節(jié)點(diǎn)，平攤到其它提供者，不會(huì)引起劇烈變動(dòng)；
下面重點(diǎn)看一下默認(rèn)的RandomLoadBalance源碼
protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
? ? ? ? int length = invokers.size(); // Number of invokers
? ? ? ? int totalWeight = 0; // The sum of weights
? ? ? ? boolean sameWeight = true; // Every invoker has the same weight?
? ? ? ? for (int i = 0; i < length; i++) {
? ? ? ? ? ? int weight = getWeight(invokers.get(i), invocation);
? ? ? ? ? ? totalWeight += weight; // Sum
? ? ? ? ? ? if (sameWeight && i > 0
? ? ? ? ? ? ? ? ? ? && weight != getWeight(invokers.get(i - 1), invocation)) {
? ? ? ? ? ? ? ? sameWeight = false;
? ? ? ? ? ? }
? ? ? ? }
? ? ? ? if (totalWeight > 0 && !sameWeight) {
? ? ? ? ? ? // If (not every invoker has the same weight & at least one invoker's weight>0), select randomly based on totalWeight.
? ? ? ? ? ? int offset = random.nextInt(totalWeight);
? ? ? ? ? ? // Return a invoker based on the random value.
? ? ? ? ? ? for (int i = 0; i < length; i++) {
? ? ? ? ? ? ? ? offset -= getWeight(invokers.get(i), invocation);
? ? ? ? ? ? ? ? if (offset < 0) {
? ? ? ? ? ? ? ? ? ? return invokers.get(i);
? ? ? ? ? ? ? ? }
? ? ? ? ? ? }
? ? ? ? }
? ? ? ? // If all invokers have the same weight value or totalWeight=0, return evenly.
? ? ? ? return invokers.get(random.nextInt(length));
? ? }
首先計(jì)算總權(quán)重，同時(shí)檢查是否每一個(gè)服務(wù)都有相同的權(quán)重；如果總權(quán)重大于0并且服務(wù)的權(quán)重都不相同，則通過(guò)權(quán)重來(lái)隨機(jī)選擇，否則直接通過(guò)Random函數(shù)來(lái)隨機(jī)；
總結(jié)
本文圍繞Cluster層中的幾個(gè)重要的接口從上到下來(lái)分別介紹，并重點(diǎn)介紹了其中的某些實(shí)現(xiàn)類；結(jié)合官方提供的調(diào)用圖，還是很容易理解此層的。
歡迎工作一到五年的Java工程師朋友們加入Java架構(gòu)開(kāi)發(fā)： 855835163
群內(nèi)提供免費(fèi)的Java架構(gòu)學(xué)習(xí)資料（里面有高可用、高并發(fā)、高性能及分布式、Jvm性能調(diào)優(yōu)、Spring源碼，MyBatis，Netty,Redis,Kafka,Mysql,Zookeeper,Tomcat,Docker,Dubbo,Nginx等多個(gè)知識(shí)點(diǎn)的架構(gòu)資料）合理利用自己每一分每一秒的時(shí)間來(lái)學(xué)習(xí)提升自己，不要再用"沒(méi)有時(shí)間“來(lái)掩飾自己思想上的懶惰！趁年輕，使勁拼，給未來(lái)的自己一個(gè)交代！
