Cluster接口的作用是将RegistryDirectory目录中的List 包装为Invoker接口,让Invoker接口调用方不需要关注负载均衡的细节。
@SPI(FailoverCluster.NAME) public interface Cluster { @Adaptive <T> Invoker<T> join(Directory<T> directory) throws RpcException; } @SPI(RandomLoadBalance.NAME) public interface LoadBalance { @Adaptive("loadbalance") <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException; } 复制代码
Cluster SPI接口默认使用FailoverCluster作为实现类,Failover就是当前服务节点调用发送网络异常时,重试其它节点。
Cluster实现会使用LoadBalance来实现多个Invoker的动态选择。默认使用随机算法实现。
/** * 失败转移,当出现失败,重试其它服务器,通常用于读操作,但重试会带来更长延迟。 * * <a href="http://en.wikipedia.org/wiki/Failover">Failover</a> * * @author william.liangf */ public class FailoverCluster implements Cluster { public final static String NAME = "failover"; public <T> Invoker<T> join(Directory<T> directory) throws RpcException { return new FailoverClusterInvoker<T>(directory); } } public abstract class AbstractClusterInvoker<T> implements Invoker<T> { private static final Logger logger = LoggerFactory .getLogger(AbstractClusterInvoker.class); protected final Directory<T> directory; protected final boolean availablecheck; private volatile boolean destroyed = false; private volatile Invoker<T> stickyInvoker = null; public AbstractClusterInvoker(Directory<T> directory) { this(directory, directory.getUrl()); } public AbstractClusterInvoker(Directory<T> directory, URL url) { if (directory == null) throw new IllegalArgumentException("service directory == null"); this.directory = directory ; //sticky 需要检测 avaliablecheck this.availablecheck = url.getParameter(Constants.CLUSTER_AVAILABLE_CHECK_KEY, Constants.DEFAULT_CLUSTER_AVAILABLE_CHECK) ; } public Class<T> getInterface() { return directory.getInterface(); } public URL getUrl() { return directory.getUrl(); } public boolean isAvailable() { Invoker<T> invoker = stickyInvoker; if (invoker != null) { return invoker.isAvailable(); } return directory.isAvailable(); } public void destroy() { directory.destroy(); destroyed = true; } public Result invoke(final Invocation invocation) throws RpcException { checkWheatherDestoried(); LoadBalance loadbalance; //调用注册目录获取可用invoker列表 List<Invoker<T>> invokers = list(invocation); if (invokers != null && invokers.size() > 0) { //获取配置好的LoadBalance实现,如果没有配置使用默认的随机算法LoadBalance实现类 loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl() .getMethodParameter(invocation.getMethodName(),Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE)); } else { loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE); } RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation); return doInvoke(invocation, invokers, loadbalance); } //抽象类,之类提供实现 protected abstract Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException; protected List<Invoker<T>> list(Invocation invocation) throws RpcException { List<Invoker<T>> invokers = directory.list(invocation); return invokers; } /** * 使用loadbalance选择invoker.</br> * a)先lb选择,如果在selected列表中 或者 不可用且做检验时,进入下一步(重选),否则直接返回</br> * b)重选验证规则:selected > available .保证重选出的结果尽量不在select中,并且是可用的 * * @param availablecheck 如果设置true,在选择的时候先选invoker.available == true * @param selected 已选过的invoker.注意:输入保证不重复 * */ protected Invoker<T> select(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException { if (invokers == null || invokers.size() == 0) return null; String methodName = invocation == null ? "" : invocation.getMethodName(); //dubbo可以基于Service方法级别配置调用亲和性,将该方法请求都固定在一台服务器节点上。默认关闭,一般情况下用不到这个特性。 boolean sticky = invokers.get(0).getUrl().getMethodParameter(methodName,Constants.CLUSTER_STICKY_KEY, Constants.DEFAULT_CLUSTER_STICKY) ; { //ignore overloaded method if ( stickyInvoker != null && !invokers.contains(stickyInvoker) ){ stickyInvoker = null; } //ignore cucurrent problem if (sticky && stickyInvoker != null && (selected == null || !selected.contains(stickyInvoker))){ if (availablecheck && stickyInvoker.isAvailable()){ return stickyInvoker; } } } Invoker<T> invoker = doselect(loadbalance, invocation, invokers, selected); if (sticky){ stickyInvoker = invoker; } return invoker; } private Invoker<T> doselect(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException { if (invokers == null || invokers.size() == 0) return null; if (invokers.size() == 1) return invokers.get(0); // 如果只有两个invoker,退化成轮循 if (invokers.size() == 2 && selected != null && selected.size() > 0) { return selected.get(0) == invokers.get(0) ? invokers.get(1) : invokers.get(0); } Invoker<T> invoker = loadbalance.select(invokers, getUrl(), invocation); //如果 selected中包含(优先判断) 或者 不可用&&availablecheck=true 则重试. if( (selected != null && selected.contains(invoker)) ||(!invoker.isAvailable() && getUrl()!=null && availablecheck)){ try{ Invoker<T> rinvoker = reselect(loadbalance, invocation, invokers, selected, availablecheck); if(rinvoker != null){ invoker = rinvoker; }else{ //看下第一次选的位置,如果不是最后,选+1位置. int index = invokers.indexOf(invoker); try{ //最后在避免碰撞 invoker = index <invokers.size()-1?invokers.get(index+1) :invoker; }catch (Exception e) { logger.warn(e.getMessage()+" may because invokers list dynamic change, ignore.",e); } } }catch (Throwable t){ logger.error("clustor relselect fail reason is :"+t.getMessage() +" if can not slove ,you can set cluster.availablecheck=false in url",t); } } return invoker; } /** * 重选,先从非selected的列表中选择,没有在从selected列表中选择. * @param loadbalance * @param invocation * @param invokers * @param selected * @return * @throws RpcException */ private Invoker<T> reselect(LoadBalance loadbalance,Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected ,boolean availablecheck) throws RpcException { //预先分配一个,这个列表是一定会用到的. List<Invoker<T>> reselectInvokers = new ArrayList<Invoker<T>>(invokers.size()>1?(invokers.size()-1):invokers.size()); //先从非select中选 if( availablecheck ){ //选isAvailable 的非select for(Invoker<T> invoker : invokers){ if(invoker.isAvailable()){ if(selected ==null || !selected.contains(invoker)){ reselectInvokers.add(invoker); } } } if(reselectInvokers.size()>0){ return loadbalance.select(reselectInvokers, getUrl(), invocation); } }else{ //选全部非select for(Invoker<T> invoker : invokers){ if(selected ==null || !selected.contains(invoker)){ reselectInvokers.add(invoker); } } if(reselectInvokers.size()>0){ return loadbalance.select(reselectInvokers, getUrl(), invocation); } } //最后从select中选可用的. { if(selected != null){ for(Invoker<T> invoker : selected){ if((invoker.isAvailable()) //优先选available && !reselectInvokers.contains(invoker)){ reselectInvokers.add(invoker); } } } if(reselectInvokers.size()>0){ return loadbalance.select(reselectInvokers, getUrl(), invocation); } } return null; } protected void checkWheatherDestoried() { if(destroyed){ throw new RpcException("Rpc cluster invoker for " + getInterface() + " on consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + " is now destroyed! Can not invoke any more."); } } @Override public String toString() { return getInterface() + " -> " + getUrl().toString(); } protected void checkInvokers(List<Invoker<T>> invokers, Invocation invocation) { if (invokers == null || invokers.size() == 0) { throw new RpcException("Failed to invoke the method " + invocation.getMethodName() + " in the service " + getInterface().getName() + ". No provider available for the service " + directory.getUrl().getServiceKey() + " from registry " + directory.getUrl().getAddress() + " on the consumer " + NetUtils.getLocalHost() + " using the dubbo version " + Version.getVersion() + ". Please check if the providers have been started and registered."); } } } public class FailoverClusterInvoker<T> extends AbstractClusterInvoker<T> { private static final Logger logger = LoggerFactory.getLogger(FailoverClusterInvoker.class); public FailoverClusterInvoker(Directory<T> directory) { super(directory); } @SuppressWarnings({ "unchecked", "rawtypes" }) 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); //如果不是抛出业务异常,默认重试2次,总计调用3次,如果只有1个服务提供者,可能会调用相同节点3次 for (int i = 0; i < len; i++) { //重试时,进行重新选择,避免重试时invoker列表已发生变化. //注意:如果列表发生了变化,那么invoked判断会失效,因为invoker示例已经改变 if (i > 0) { checkWheatherDestoried(); copyinvokers = list(invocation); //重新检查一下 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); } } /** * 快速失败,只发起一次调用,失败立即报错,通常用于非幂等性的写操作。 * * <a href="http://en.wikipedia.org/wiki/Fail-fast">Fail-fast</a> * * @author william.liangf * @author chao.liuc */ public class FailfastClusterInvoker<T> extends AbstractClusterInvoker<T>{ public FailfastClusterInvoker(Directory<T> directory) { super(directory); } public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException { checkInvokers(invokers, invocation); Invoker<T> invoker = select(loadbalance, invocation, invokers, null); try { return invoker.invoke(invocation); } catch (Throwable e) { if (e instanceof RpcException && ((RpcException)e).isBiz()) { // biz exception. throw (RpcException) e; } throw new RpcException(e instanceof RpcException ? ((RpcException)e).getCode() : 0, "Failfast invoke providers " + invoker.getUrl() + " " + loadbalance.getClass().getSimpleName() + " select from all providers " + invokers + " for service " + getInterface().getName() + " method " + invocation.getMethodName() + " on consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + ", but no luck to perform the invocation. Last error is: " + e.getMessage(), e.getCause() != null ? e.getCause() : e); } } } public abstract class AbstractLoadBalance implements LoadBalance { public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) { if (invokers == null || invokers.size() == 0) return null; if (invokers.size() == 1) return invokers.get(0); return doSelect(invokers, url, invocation); } //获取服务提供者权重 protected abstract <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation); protected int getWeight(Invoker<?> invoker, Invocation invocation) { int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT); if (weight > 0) { //dubbo调用方会根据服务提供方服务上线时间来调整节点权重,默认要达到10分钟才会分配100%权重,10分钟以内按照时间比例来调整权重,比如服务发布1分钟分配的权重比例是10%,利用这种方式可以让java代码有足够长的热身时间,JVM每段代码调用超过1万次才会转化为机器指令,否则就是解释执行。但是该段代码有bug,在官方dubbo版本以及修复。 long timestamp = invoker.getUrl().getParameter(Constants.TIMESTAMP_KEY, 0L); if (timestamp > 0L) { int uptime = (int) (System.currentTimeMillis() - timestamp); int warmup = invoker.getUrl().getParameter(Constants.WARMUP_KEY, Constants.DEFAULT_WARMUP); if (uptime > 0 && uptime < warmup) { weight = calculateWarmupWeight(uptime, warmup, weight); } } } return weight; } static int calculateWarmupWeight(int uptime, int warmup, int weight) { int ww = (int) ( (float) uptime / ( (float) warmup / (float) weight ) ); return ww < 1 ? 1 : (ww > weight ? weight : ww); } } //按照每个服务提供者节点权重随机返回提供者 public class RandomLoadBalance extends AbstractLoadBalance { public static final String NAME = "random"; private final Random random = new Random(); protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) { int length = invokers.size(); // 总个数 int totalWeight = 0; // 总权重 boolean sameWeight = true; // 权重是否都一样 for (int i = 0; i < length; i++) { int weight = getWeight(invokers.get(i), invocation); totalWeight += weight; // 累计总权重 if (sameWeight && i > 0 && weight != getWeight(invokers.get(i - 1), invocation)) { sameWeight = false; // 计算所有权重是否一样 } } if (totalWeight > 0 && ! sameWeight) { // 如果权重不相同且权重大于0则按总权重数随机 int offset = random.nextInt(totalWeight); // 并确定随机值落在哪个片断上 for (int i = 0; i < length; i++) { offset -= getWeight(invokers.get(i), invocation); if (offset < 0) { return invokers.get(i); } } } // 如果权重相同或权重为0则均等随机 return invokers.get(random.nextInt(length)); } } 复制代码