dubbo源码分析系列——dubbo-cluster模块源码分析
模块功能介绍
该模块的使用介绍请参考dubbo官方用户手册以下章节内容。java
其中注册中心实际上是对于目录服务的一种实现方式，本文不会对注册中心进行详细讲解。算法
核心类图
核心源码分析
核心接口概念及关系
各节点关系：express
- 这里的Invoker是Provider的一个可调用Service的抽象，Invoker封装了Provider地址及Service接口信息。
- Directory表明多个Invoker，能够把它当作List<Invoker>，但与List不一样的是，它的值多是动态变化的，好比注册中心推送变动。
- Cluster将Directory中的多个Invoker假装成一个Invoker，对上层透明，假装过程包含了容错逻辑，调用失败后，重试另外一个。
- Router负责从多个Invoker中按路由规则选出子集，好比读写分离，应用隔离等。
- LoadBalance负责从多个Invoker中选出具体的一个用于本次调用，选的过程包含了负载均衡算法，调用失败后，须要重选。
因为每种接口都有多种实现类，篇幅和时间有限，咱们选择其中最为典型的一种来进行源码分析。apache
Cluster
扩展接口介绍
集群的源码以下。api
package com.alibaba.dubbo.rpc.cluster;
import com.alibaba.dubbo.common.extension.Adaptive;
import com.alibaba.dubbo.common.extension.SPI;
import com.alibaba.dubbo.rpc.Invoker;
import com.alibaba.dubbo.rpc.RpcException;
import com.alibaba.dubbo.rpc.cluster.support.FailoverCluster;
/**
* Cluster. (SPI, Singleton, ThreadSafe)
*
* <a href="http://en.wikipedia.org/wiki/Computer_cluster">Cluster</a>
* <a href="http://en.wikipedia.org/wiki/Fault-tolerant_system">Fault-Tolerant</a>
*
* @author william.liangf
*/
@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;
}
该接口只有一个方法，就是将directory对象中的多个invoker的集合整合成一个invoker对象。该方法被ReferenceConfig类的createProxy方法调用，调用它的代码以下。缓存
// 对有注册中心的Cluster 只用 AvailableCluster
URL u = registryURL.addParameter(Constants.CLUSTER_KEY, AvailableCluster.NAME);
invoker = cluster.join(new StaticDirectory(u, invokers));
Cluster内置有9个扩展实现类，都实现了不一样的集群容错策略，咱们只分析默认的自动故障转移的扩展实现FailoverCluster。服务器
FailoverCluster
源码以下，只是构造了一个类型为FailoverClusterInvoker的invoker对象。app
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);
}
}
咱们进入看看FailoverClusterInvoker的源码。
负载均衡
/**
* 失败转移，当出现失败，重试其它服务器，一般用于读操做，但重试会带来更长延迟。
*
* <a href="http://en.wikipedia.org/wiki/Failover">Failover</a>
*
* @author william.liangf
* @author chao.liuc
*/
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);
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);
}
}
该类又继承自抽象实现类AbstractClusterInvoker，使用该类的一些方法，所以也要结合该类的源码一块儿看。less
/*
* Copyright 1999-2011 Alibaba Group.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.alibaba.dubbo.rpc.cluster.support;
import java.util.ArrayList;
import java.util.List;
import com.alibaba.dubbo.common.Constants;
import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.common.Version;
import com.alibaba.dubbo.common.extension.ExtensionLoader;
import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.alibaba.dubbo.common.utils.NetUtils;
import com.alibaba.dubbo.rpc.Invocation;
import com.alibaba.dubbo.rpc.Invoker;
import com.alibaba.dubbo.rpc.Result;
import com.alibaba.dubbo.rpc.RpcException;
import com.alibaba.dubbo.rpc.cluster.Directory;
import com.alibaba.dubbo.rpc.cluster.LoadBalance;
import com.alibaba.dubbo.rpc.support.RpcUtils;
/**
* AbstractClusterInvoker
*
* @author william.liangf
* @author chao.liuc
*/
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;
}
/**
* 使用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();
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;
}
public Result invoke(final Invocation invocation) throws RpcException {
checkWheatherDestoried();
LoadBalance loadbalance;
List<Invoker<T>> invokers = list(invocation);
if (invokers != null && invokers.size() > 0) {
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 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.");
}
}
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;
}
}
源码实现分析。
- AbstractClusterInvoker的invoke方法提供了一个骨架实现。逻辑是检查对象是否销毁状态，从directory得到invoker列表，得到loadbalance扩展实现对象，而后调用抽象方法doInvoke去执行真正的逻辑，交由具体子类实现。
- AbstractClusterInvoker实现了一个公共的protected的方法select，该方法实现了使用loadbalance选择合适的invoker对象。在选择方法的实现中支持 粘滞链接 特性。做为一个公共特性，全部的子类都支持。最后再调用private方法doselect实现进一步选择逻辑。
- AbstractClusterInvoker的doselect方法实现了真正的选择invoker逻辑。首先检查可选invoker，若没有则返回null；若是有2个可选invoker则退化为轮询；不然继续调用loadbalance的select方法选择一个invoker；而后在检查选中的invoker是否已经使用过或者不可用，若是不可用则会调用reselect从新选择，若从新选择成功则使用它，不然则使用invoker列表中当前index+1的invoker，若是已是最后一个则直接使用当前的invoker。
- AbstractClusterInvoker的reselect方法的实现逻辑是：若是availablecheck标志为true，则只将未被selected的可用状态的invoker交给loadbalance进行选择，不然将全部的未被selected的invoker交给loadbalance选择，若可从新选择的invoker为空，则将selected的invoker列表交给loadbalance进行选择。
- FailoverClusterInvoker的doInvoke方法的实现逻辑为：检查invoker列表状态；得到参数中重试次数，默认次数是2；从directory得到invoker列表；调用select方法选择一个invoker；将选择的invoker加入到invoked集合，表示已经选择和使用的；调用invoker.invoke()方法，若成功则返回result，若抛出的是业务异常则抛出，不然继续重试选择并调用下一个invoker；
LoadBalance
负载均衡器
扩展接口定义
@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;
}
上述源码所示，负载均衡只定义了一个方法，就是在候选的invokers中选择一个invoker对象出来。默认的扩展实现是random。那我么就分析RandomLoadBalance的源码。
RandomLoadBalance
/*
* Copyright 1999-2011 Alibaba Group.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.alibaba.dubbo.rpc.cluster.loadbalance;
import java.util.List;
import java.util.Random;
import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.rpc.Invocation;
import com.alibaba.dubbo.rpc.Invoker;
/**
* random load balance.
*
* @author qianlei
* @author william.liangf
*/
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));
}
}
该类继承了抽象类AbstractLoadBalance，所以咱们也要结合该类一块儿分析。
/*
* Copyright 1999-2011 Alibaba Group.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.alibaba.dubbo.rpc.cluster.loadbalance;
import java.util.List;
import com.alibaba.dubbo.common.Constants;
import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.rpc.Invoker;
import com.alibaba.dubbo.rpc.Invocation;
import com.alibaba.dubbo.rpc.cluster.LoadBalance;
/**
* AbstractLoadBalance
*
* @author william.liangf
*/
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) {
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);
}
}
源码分析以下：
- AbstractLoadBalance的select方法实现，只是作了参数校验，invoker列表若0个则返回null，1个元素则直接返回；不然调用抽象方法doSelect交给子类实现。
- AbstractLoadBalance定义了公共方法getWeight。该方法是获取invoker的权重的方法，公式是：(int) ( (float) uptime / ( (float) warmup / (float) weight ) );
- 若是未设置权重或者权重值都同样，则直接调用random.nextInt()随机得到一个invoker；若设置了权重而且不同，则在总权重中随机，分布在哪一个invoker的分片上，则选择该invoker对象，实现了按照权重随机。
Router
接口定义
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;
}
路由器就定义了上述2个方法，核心方法是route，从大的invoker列表结合中根据规则过滤出一个子集合。咱们这里只分析实现类ConditionRouter的源码。
ConditionRouter
/*
* Copyright 1999-2012 Alibaba Group.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.alibaba.dubbo.rpc.cluster.router.condition;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import com.alibaba.dubbo.common.Constants;
import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.alibaba.dubbo.common.utils.NetUtils;
import com.alibaba.dubbo.common.utils.StringUtils;
import com.alibaba.dubbo.common.utils.UrlUtils;
import com.alibaba.dubbo.rpc.Invocation;
import com.alibaba.dubbo.rpc.Invoker;
import com.alibaba.dubbo.rpc.RpcException;
import com.alibaba.dubbo.rpc.cluster.Router;
/**
* ConditionRouter
*
* @author william.liangf
*/
public class ConditionRouter implements Router, Comparable<Router> {
private static final Logger logger = LoggerFactory.getLogger(ConditionRouter.class);
private final URL url;
private final int priority;
private final boolean force;
private final Map<String, MatchPair> whenCondition;
private final Map<String, MatchPair> thenCondition;
public ConditionRouter(URL url) {
this.url = url;
this.priority = url.getParameter(Constants.PRIORITY_KEY, 0);
this.force = url.getParameter(Constants.FORCE_KEY, false);
try {
String rule = url.getParameterAndDecoded(Constants.RULE_KEY);
if (rule == null || rule.trim().length() == 0) {
throw new IllegalArgumentException("Illegal route rule!");
}
rule = rule.replace("consumer.", "").replace("provider.", "");
int i = rule.indexOf("=>");
String whenRule = i < 0 ? null : rule.substring(0, i).trim();
String thenRule = i < 0 ? rule.trim() : rule.substring(i + 2).trim();
Map<String, MatchPair> when = StringUtils.isBlank(whenRule) || "true".equals(whenRule) ? new HashMap<String, MatchPair>() : parseRule(whenRule);
Map<String, MatchPair> then = StringUtils.isBlank(thenRule) || "false".equals(thenRule) ? null : parseRule(thenRule);
// NOTE: When条件是容许为空的，外部业务来保证相似的约束条件
this.whenCondition = when;
this.thenCondition = then;
} catch (ParseException e) {
throw new IllegalStateException(e.getMessage(), e);
}
}
public <T> List<Invoker<T>> route(List<Invoker<T>> invokers, URL url, Invocation invocation)
throws RpcException {
if (invokers == null || invokers.size() == 0) {
return invokers;
}
try {
if (! matchWhen(url)) {
return invokers;
}
List<Invoker<T>> result = new ArrayList<Invoker<T>>();
if (thenCondition == null) {
logger.warn("The current consumer in the service blacklist. consumer: " + NetUtils.getLocalHost() + ", service: " + url.getServiceKey());
return result;
}
for (Invoker<T> invoker : invokers) {
if (matchThen(invoker.getUrl(), url)) {
result.add(invoker);
}
}
if (result.size() > 0) {
return result;
} else if (force) {
logger.warn("The route result is empty and force execute. consumer: " + NetUtils.getLocalHost() + ", service: " + url.getServiceKey() + ", router: " + url.getParameterAndDecoded(Constants.RULE_KEY));
return result;
}
} catch (Throwable t) {
logger.error("Failed to execute condition router rule: " + getUrl() + ", invokers: " + invokers + ", cause: " + t.getMessage(), t);
}
return invokers;
}
public URL getUrl() {
return url;
}
public int compareTo(Router o) {
if (o == null || o.getClass() != ConditionRouter.class) {
return 1;
}
ConditionRouter c = (ConditionRouter) o;
return this.priority == c.priority ? url.toFullString().compareTo(c.url.toFullString()) : (this.priority > c.priority ? 1 : -1);
}
public boolean matchWhen(URL url) {
return matchCondition(whenCondition, url, null);
}
public boolean matchThen(URL url, URL param) {
return thenCondition != null && matchCondition(thenCondition, url, param);
}
private boolean matchCondition(Map<String, MatchPair> condition, URL url, URL param) {
Map<String, String> sample = url.toMap();
for (Map.Entry<String, String> entry : sample.entrySet()) {
String key = entry.getKey();
MatchPair pair = condition.get(key);
if (pair != null && ! pair.isMatch(entry.getValue(), param)) {
return false;
}
}
return true;
}
private static Pattern ROUTE_PATTERN = Pattern.compile("([&!=,]*)\\s*([^&!=,\\s]+)");
private static Map<String, MatchPair> parseRule(String rule)
throws ParseException {
Map<String, MatchPair> condition = new HashMap<String, MatchPair>();
if(StringUtils.isBlank(rule)) {
return condition;
}
// 匹配或不匹配Key-Value对
MatchPair pair = null;
// 多个Value值
Set<String> values = null;
final Matcher matcher = ROUTE_PATTERN.matcher(rule);
while (matcher.find()) { // 逐个匹配
String separator = matcher.group(1);
String content = matcher.group(2);
// 表达式开始
if (separator == null || separator.length() == 0) {
pair = new MatchPair();
condition.put(content, pair);
}
// KV开始
else if ("&".equals(separator)) {
if (condition.get(content) == null) {
pair = new MatchPair();
condition.put(content, pair);
} else {
condition.put(content, pair);
}
}
// KV的Value部分开始
else if ("=".equals(separator)) {
if (pair == null)
throw new ParseException("Illegal route rule \""
+ rule + "\", The error char '" + separator
+ "' at index " + matcher.start() + " before \""
+ content + "\".", matcher.start());
values = pair.matches;
values.add(content);
}
// KV的Value部分开始
else if ("!=".equals(separator)) {
if (pair == null)
throw new ParseException("Illegal route rule \""
+ rule + "\", The error char '" + separator
+ "' at index " + matcher.start() + " before \""
+ content + "\".", matcher.start());
values = pair.mismatches;
values.add(content);
}
// KV的Value部分的多个条目
else if (",".equals(separator)) { // 若是为逗号表示
if (values == null || values.size() == 0)
throw new ParseException("Illegal route rule \""
+ rule + "\", The error char '" + separator
+ "' at index " + matcher.start() + " before \""
+ content + "\".", matcher.start());
values.add(content);
} else {
throw new ParseException("Illegal route rule \"" + rule
+ "\", The error char '" + separator + "' at index "
+ matcher.start() + " before \"" + content + "\".", matcher.start());
}
}
return condition;
}
private static final class MatchPair {
final Set<String> matches = new HashSet<String>();
final Set<String> mismatches = new HashSet<String>();
public boolean isMatch(String value, URL param) {
for (String match : matches) {
if (! UrlUtils.isMatchGlobPattern(match, value, param)) {
return false;
}
}
for (String mismatch : mismatches) {
if (UrlUtils.isMatchGlobPattern(mismatch, value, param)) {
return false;
}
}
return true;
}
}
}
该源码实现了以下条件路由器功能。
基于条件表达式的路由规则，如：
|
|
规则：
- "=>"以前的为消费者匹配条件，全部参数和消费者的URL进行对比，当消费者知足匹配条件时，对该消费者执行后面的过滤规则。
- "=>"以后为提供者地址列表的过滤条件，全部参数和提供者的URL进行对比，消费者最终只拿到过滤后的地址列表。
- 若是匹配条件为空，表示对全部消费方应用，如：=> host != 10.20.153.11
- 若是过滤条件为空，表示禁止访问，如：host = 10.20.153.10 =>
表达式：
- 参数支持：
- 服务调用信息，如：method, argument 等 (暂不支持参数路由)
- URL自己的字段，如：protocol, host, port 等
- 以及URL上的全部参数，如：application, organization 等
- 条件支持：
- 等号"="表示"匹配"，如：host = 10.20.153.10
- 不等号"!="表示"不匹配"，如：host != 10.20.153.10
- 值支持：
- 以逗号","分隔多个值，如：host != 10.20.153.10,10.20.153.11
- 以星号"*"结尾，表示通配，如：host != 10.20.*
- 以美圆符"$"开头，表示引用消费者参数，如：host = $host
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;
}
目录服务定义了一个核心接口list，就是列举出某个接口在目录中的全部服务列表。
抽象实现AbstractDirectory
提供了一个抽象的目录实现类，源码以下。
/**
* 增长router的Directory
*
* @author chao.liuc
*/
public abstract class AbstractDirectory<T> implements Directory<T> {
// 日志输出
private static final Logger logger = LoggerFactory.getLogger(AbstractDirectory.class);
private final URL url ;
private volatile boolean destroyed = false;
private volatile URL consumerUrl ;
private volatile List<Router> routers;
public AbstractDirectory(URL url) {
this(url, null);
}
public AbstractDirectory(URL url, List<Router> routers) {
this(url, url, routers);
}
public AbstractDirectory(URL url, URL consumerUrl, List<Router> routers) {
if (url == null)
throw new IllegalArgumentException("url == null");
this.url = url;
this.consumerUrl = consumerUrl;
setRouters(routers);
}
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.size() > 0) {
for (Router router: localRouters){
try {
if (router.getUrl() == null || router.getUrl().getParameter(Constants.RUNTIME_KEY, true)) {
invokers = router.route(invokers, getConsumerUrl(), invocation);
}
} catch (Throwable t) {
logger.error("Failed to execute router: " + getUrl() + ", cause: " + t.getMessage(), t);
}
}
}
return invokers;
}
public URL getUrl() {
return url;
}
public List<Router> getRouters(){
return routers;
}
public URL getConsumerUrl() {
return consumerUrl;
}
public void setConsumerUrl(URL consumerUrl) {
this.consumerUrl = consumerUrl;
}
protected void setRouters(List<Router> routers){
// copy list
routers = routers == null ? new ArrayList<Router>() : new ArrayList<Router>(routers);
// append url router
String routerkey = url.getParameter(Constants.ROUTER_KEY);
if (routerkey != null && routerkey.length() > 0) {
RouterFactory routerFactory = ExtensionLoader.getExtensionLoader(RouterFactory.class).getExtension(routerkey);
routers.add(routerFactory.getRouter(url));
}
// append mock invoker selector
routers.add(new MockInvokersSelector());
Collections.sort(routers);
this.routers = routers;
}
public boolean isDestroyed() {
return destroyed;
}
public void destroy(){
destroyed = true;
}
protected abstract List<Invoker<T>> doList(Invocation invocation) throws RpcException ;
}
list方法的实现逻辑是：先检查目录是否销毁状态，若已经销毁则抛出异常；调用抽象方法doList实现真正的从目录服务中获取invoker列表，该方法须要子类实现；循环对象中的路由器列表，若路由器url为null或者参数runtime为true则调用该路由器的route方法进行路由，将返回的invoker列表替换为路由后的结果； 返回最终的invoker列表。
setRouters方法是设置路由器列表，除了参数参入的routers以外，还会追加2个默认的路由器，一个是参数router指定的routerFactory得到的router，另一个是MockInvokersSelector对象；
默认实现StaticDirectory
模块还提供了一个默认目录实现类StaticDirectory，它是一个静态的内存缓存目录服务实现。源码以下：
public class StaticDirectory<T> extends AbstractDirectory<T> {
private final List<Invoker<T>> invokers;
public StaticDirectory(List<Invoker<T>> invokers){
this(null, invokers, null);
}
public StaticDirectory(List<Invoker<T>> invokers, List<Router> routers){
this(null, invokers, routers);
}
public StaticDirectory(URL url, List<Invoker<T>> invokers) {
this(url, invokers, null);
}
public StaticDirectory(URL url, List<Invoker<T>> invokers, List<Router> routers) {
super(url == null && invokers != null && invokers.size() > 0 ? invokers.get(0).getUrl() : url, routers);
if (invokers == null || invokers.size() == 0)
throw new IllegalArgumentException("invokers == null");
this.invokers = invokers;
}
public Class<T> getInterface() {
return invokers.get(0).getInterface();
}
public boolean isAvailable() {
if (isDestroyed()) {
return false;
}
for (Invoker<T> invoker : invokers) {
if (invoker.isAvailable()) {
return true;
}
}
return false;
}
public void destroy() {
if(isDestroyed()) {
return;
}
super.destroy();
for (Invoker<T> invoker : invokers) {
invoker.destroy();
}
invokers.clear();
}
@Override
protected List<Invoker<T>> doList(Invocation invocation) throws RpcException {
return invokers;
}
}
它的doList方法的实现是直接将属性invokers的值返回，很是简单。
此外还有一个RegistryDirectory的实现类，该类是整合了注册中心和目录服务。
NEXT
由于考虑到本模块与dubbo-registry相关性较大，接下来咱们将研究dubbo-registry-api和dubbo-registry-default模块的源码。
相关文章
- 1. dubbo源码分析系列——dubbo-rpc-default模块源码分析
- 2. dubbo源码分析系列——dubbo-rpc-api模块源码分析
- 3. dubbo源码分析系列——dubbo-register-api模块源码分析
- 4. Dubbo系列源码分析
- 5. dubbo源码分析系列——dubbo的SPI机制源码分析
- 6. 小强源码分析系列-Dubbo-源码分析2---模块整体介绍
- 7. Dubbo 源码分析
- 8. dubbo源码分析
- 9. 源码分析Dubbo系列文章
- 10. jQuery源码分析系列 jQuery源码分析系列
- 更多相关文章...
- • Eclipse 代码模板 - Eclipse 教程
- • 互联网系统应用架构基础分析 - 红包项目实战
- • 互联网组织的未来：剖析GitHub员工的任性之源
- • Java Agent入门实战（二）-Instrumentation源码概述
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
欢迎关注本站公众号,获取更多信息
