转载

原 荐 Dubbo服务调用——流程分析

我们根据官网的案例来分析Dubbo的调用过程是什么样的

1.首先粘贴下调用源头

public static void main(String[] args) throws Exception {
    ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(new String[]{"META-INF/spring/dubbo-demo-consumer.xml"});
    context.start();

    DemoService demoService = (DemoService) context.getBean("demoService"); // 获取远程服务代理
    String hello = demoService.sayHello("world"); // 执行远程方法

    System.out.println(hello); // 显示调用结果

    System.in.read();
}

断点根据调用堆栈得知,实际上调用了InvokerInvocationHandler的代理对象(动态代理)

public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
    String methodName = method.getName();
    Class<?>[] parameterTypes = method.getParameterTypes();
    if (method.getDeclaringClass() == Object.class) {
        return method.invoke(invoker, args);
    }
    if ("toString".equals(methodName) && parameterTypes.length == 0) {
        return invoker.toString();
    }
    if ("hashCode".equals(methodName) && parameterTypes.length == 0) {
        return invoker.hashCode();
    }
    if ("equals".equals(methodName) && parameterTypes.length == 1) {
        return invoker.equals(args[0]);
    }
    return invoker.invoke(new RpcInvocation(method, args)).recreate();
}
  1. 将入方法和入参封装为一个远程调用的实体:RpcInvocation 
  2. 调用MokeClusterInvoker 对象的 invoke方法 
public Result invoke(Invocation invocation) throws RpcException {
    Result result = null;

    String value = directory.getUrl().getMethodParameter(invocation.getMethodName(), Constants.MOCK_KEY, Boolean.FALSE.toString()).trim();
    if (value.length() == 0 || value.equalsIgnoreCase("false")) {
        //1. no mock
        result = this.invoker.invoke(invocation);
    } else if (value.startsWith("force")) {
        if (logger.isWarnEnabled()) {
            logger.info("force-mock: " + invocation.getMethodName() + " force-mock enabled , url : " + directory.getUrl());
        }
        //force:direct mock  强制走moke逻辑
        result = doMockInvoke(invocation, null);
    } else {
        //fail-mock 异常moke逻辑
        try {
            result = this.invoker.invoke(invocation);
        } catch (RpcException e) {
            if (e.isBiz()) {
                throw e;
            } else {
                if (logger.isWarnEnabled()) {
                    logger.info("fail-mock: " + invocation.getMethodName() + " fail-mock enabled , url : " + directory.getUrl(), e);
                }
                result = doMockInvoke(invocation, e);
            }
        }
    }
    return result;
}

MokeClusterInvoker   moke 为正常业务错误时的"备胎",上面代码执行分为三种逻辑

  1. 正常逻辑不走moke
  2. 强制 moke
  3. 出错 moke

moke 实现方式我们以后分析。 接下来调用 FailoverClusterInvoker 对象 invoke 方法

public Result invoke(final Invocation invocation) throws RpcException {

    checkWhetherDestroyed();
    LoadBalance loadbalance;
    // 从 directory 获得所有服务提供者 Invoker 集合
    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);
    }
    //异步的话,需要添加id
    RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
    return doInvoke(invocation, invokers, loadbalance);
}

FailoverClusterInvoker  只是ClusterInvoker 的一种扩展实现。( 失败自动切换,当出现失败,重试其它服务器)

  1. 检测是否禁用
  2. 从Directory 中获取可用的Invoker数组
  3. 获取负载策略 默认为RandomLoadBalance

原 荐 Dubbo服务调用——流程分析

我们查看 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++) {
        //重试时,进行重新选择,避免重试时invoker列表已发生变化.
        //注意:如果列表发生了变化,那么invoked判断会失效,因为invoker示例已经改变
        if (i > 0) {
            checkWhetherDestroyed();
            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);
}
  1. 检测copyinvokers合法性 (size > 0)
  2. 获取调用方法重试次数 len 默认为3次
  3. 声明 List<Invoker<T>> invoked 大小为invokers的大小 , 存放失败调用的invoke对象
  4. 声明 Set<String> providers 记录调用失败的远程连接地址 大小默认为len 大小
  5. 针对每次循环调用,都会重新获取一遍invoker列表,  防止服务提供端发生改变。
  6. 调用select方法从invokers列表中 按照指定策略获取一个invoker对象。并在invoked中添加 invoker对象,并设置到本线程的调用上下文中。
  7. 进行服务调用(经过过滤器链) 
  8. 调用失败则进入循环体,重新循环调用。

接下来看 6过程中 Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);

/**
 * 使用 loadbalance 选择 invoker.
 *
 * @param loadbalance Loadbalance 对象,提供负责均衡策略
 * @param invocation Invocation 对象
 * @param invokers  候选的 Invoker 集合
 * @param selected  已选过的 Invoker 集合. 注意:输入保证不重复
 * @return  最终的 Invoker 对象
 * @throws RpcException  RpcException 当发生 RpcException 时
 */
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;
    // 获得 sticky(粘滞)配置项,方法级
    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   // 若 stickyInvoker 不存在于 invokers 中,说明不在候选中,需要置空,重新选择
        if (stickyInvoker != null && !invokers.contains(stickyInvoker)) {
            stickyInvoker = null;
        }
        //ignore cucurrent problem  若开启粘滞连接的特性,且 stickyInvoker 不存在于 selected 中,则返回 stickyInvoker 这个 Invoker 对象
        if (sticky && stickyInvoker != null && (selected == null || !selected.contains(stickyInvoker))) {
            // 若开启排除非可用的 Invoker 的特性,则校验 stickyInvoker 是否可用。若可用,则进行返回
            if (availablecheck && stickyInvoker.isAvailable()) {
                return stickyInvoker;
            }
        }
    }
    // 执行选择
    Invoker<T> invoker = doselect(loadbalance, invocation, invokers, selected);
    // 若开启粘滞连接的特性,记录最终选择的 Invoker 到 stickyInvoker
    if (sticky) {
        stickyInvoker = invoker;
    }
    return invoker;
}

该方法作用是获取 invokers 中指定的调用对象,调用过程如下

  1. 如果服务提供端列表为空,则直接返回null
  2. 获取调用方法名称 , 并判断该方法是否支持粘滞 ,如果粘滞则默认从 stickyInvoker 中获取对象
  3. 如果stickyInvoker 不存在于invokers中说明该服务已经卸载, 则置空并重新选取并调用。
  4. 调用 doSelect 执行选择 

粘滞连接用于有状态服务,尽可能让客户端总是向同一提供者发起调用,除非该提供者挂了,再连另一台。粘滞连接将自动开启延迟连接,以减少长连接数。

<dubbo:protocol name="dubbo" sticky="true" />

接下来看4 中的 doselect 方法

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;
    // 一、如果只有一个 Invoker ,直接选择
    if (invokers.size() == 1)
        return invokers.get(0);
    // 二、如果只有两个invoker,退化成轮循
    // If we only have two invokers, use round-robin instead.
    if (invokers.size() == 2 && selected != null && selected.size() > 0) {
        return selected.get(0) == invokers.get(0) ? invokers.get(1) : invokers.get(0);
    }
    // 三、使用 Loadbalance ,选择一个 Invoker 对象。
    Invoker<T> invoker = loadbalance.select(invokers, getUrl(), invocation);

    // If the `invoker` is in the  `selected` or invoker is unavailable && availablecheck is true, reselect.
    // 如果 selected中包含(优先判断) 或者 不可用&&availablecheck=true 则重试.
    if ((selected != null && selected.contains(invoker))
            || (!invoker.isAvailable() && getUrl() != null && availablecheck)) {
        try {
            //四、重选一个 Invoker 对象
            Invoker<T> rinvoker = reselect(loadbalance, invocation, invokers, selected, availablecheck);
            if (rinvoker != null) {
                invoker = rinvoker;
            } else {
                // Check the index of current selected invoker, if it's not the last one, choose the one at index+1.
                // 五、看下第一次选的位置,如果不是最后,选+1位置.
                int index = invokers.indexOf(invoker);
                try {
                    // Avoid collision
                    //最后在避免碰撞
                    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;
}

方法的作用是 : 从候选的 Invoker 集合,选择一个最终调用的 Invoker 对象。我们看下它的调用过程

  1. 如果候选集合中只有一个invoker , 则直接返回该invoker
  2. 如果有两个invoker并且不是第一次调用,则退化为轮询。(round-robin)
  3. 否则利用loadbalance.select 获取一个invoker对象。
  4. 如果selected中包含(优先判断) 或者 不可用&&检测 则重新选择invoker  reselect(loadbalance, invocation, invokers, selected, availablecheck);
  5. 如果重选为空,获取第一次选择下一个或当前选择。

private Invoker<T> reselect(LoadBalance loadbalance, Invocation invocation,

List<Invoker<T>> invokers, List<Invoker<T>> selected, boolean availablecheck) 

重选逻辑 : 先从非selected的列表中选择,没有在从selected列表中选择.

最后进行Invoker过滤器链调用进行远程方法调用。

loadbalance 结构图如下

原 荐 Dubbo服务调用——流程分析

这几种扩展种类:

1.Random LoadBalance

随机,按权重设置随机概率

在一个截面上碰撞的概率高,但调用量越大,分布越均匀,而且按概率使用权重后也比较均匀,有利于动态调整提供者权重

2.RoundRobin LoadBalance

轮询,按公约后的权重设置轮询比例

存在慢的提供者累计请求的问题。

3.LeastActive LoadBalance

最少活跃调用数,相同活跃数的随机,活跃数指调用前后计数差

使慢的提供者受到更少的请求,因为越慢的提供者的调用前后计数差会越大。

4.ConsistentHash LoadBalance

一致性Hash,相同参数请求总数发到同一提供者。

当某一台提供者挂时,原本发往该提供者的请求,基于虚拟节点,平摊到其他提供者,不会引起剧烈变动。

配置参考如下:

<dubbo:service interface="..."loadbalance="roundrobin"/>
<dubbo:reference interface="..."loadbalance="roundrobin"/>
<dubbo:service interface="...">
    <dubbo:method name="..."loadbalance="roundrobin"/>
</dubbo:service>
<dubbo:reference interface="...">
    <dubbo:method name="..."loadbalance="roundrobin"/>
</dubbo:reference>
<dubbo:reference interface="..." loadbalance="consistenthash" />
缺省只对第一个参数Hash,如果要修改,请配置
<dubbo:parameter key="hash.arguments" value="0,1" />
缺省用160份虚拟节点,如果要修改,请配置
<dubbo:parameter key="hash.nodes" value="320" />
原文  https://my.oschina.net/LucasZhu/blog/1928494
正文到此结束
Loading...