Skip to content

21-调用流程:消费方的调用流程体系,你知道多少? ​

你好,我是何辉。今天我们深入研究Dubbo源码的第十篇,调用流程。

在消费方这样的代码你一定见过很多了。

java
@DubboReference
private DemoFacade demoFacade;

在 Spring Bean 中定义 Dubbo 接口为成员变量时,用 @DubboReference 注解修饰 DemoFacade。

你有没有好奇过,我们现在看到的这个 demoFacade 变量,在内存运行时,值类型还属于 DemoFacade 这个类型么?如果不是,那拿着 demoFacade 变量去调用里面的方法时,在消费方到底会经历怎样的调用流程呢?

这个问题,我以前也探索过,不过因为源码太复杂,最后都徒劳无功。直到有一天,遇到了一位资深大佬,指点了我看源码的 12 字方针:不钻细节:只看流程;不看过程:只看结论;再看细节:再看过程。参考他的经验,我经过一番简单的调试后,就轻松梳理出了调用流程的大体框架。

今天我就带你按照这12字方针的思路,感受一下高手是如何研究源码流程的。每个环节,我会用图片总结,加强你对流程的形象化理解。今天的内容稍微多一些,做好准备,我们马上开始。

sayHello 调试 ​

先来看下我们的消费方调用代码。

java
///////////////////////////////////////////////////
// 消费方的一个 Spring Bean 类
// 1、里面定义了下游 Dubbo 接口的成员变量,并且还用 @DubboReference 修饰了一下。
// 2、还定义了一个 invokeDemo 方法被外部调用,但其重点是该方法可以调用下游的 Dubbo 接口
///////////////////////////////////////////////////
@Component
public class InvokeDemoService {

    // 定义调用下游接口的成员变量,
    // 并且用注解修饰
    @DubboReference
    private DemoFacade demoFacade;
    
    // invokeDemo 是被外部触发调用的,不过这不是重点,
    // 重点的是该 invokeDemo 逻辑中能调用下游的接口,
    // 这里调用的是下游 DemoFacade 接口的 sayHello 方法
    public String invokeDemo(){
        return demoFacade.sayHello("Geek");
    }
}

代码在这个类中定义了一个 invokeDemo 的方法,然后,在方法体中对下游 DemoFacade 接口的 sayHello 方法发起了远程调用。

是不是发现消费方真的非常简单且普通。那怎么展开探索呢?我们以 Debug 的方式调试消费方,在 sayHello 方法打个断点,通过断点调试,钻进调用流程的各个环节,同时参考12字方针,深入研究调用流程的底层逻辑。

1. JDK代理 ​

我们在调用 sayHello 方法的这行打上一个断点,先运行提供方,再 Debug 运行消费方,很快断点就到来了。

图片

还记得前面提出的一个疑问么,demoFacade 在内存运行时,值类型是什么?

从图中,我们可以清楚地看到,demoFacade 的类型,是一个随机生成的代理类名,不再属于 DemoFacade 这个类型了,而且结合 $Proxy 类名、代理类中的 h 成员变量属于 JdkDynamicAopProxy 类型,综合判断,这是采用 JDK 代理动态,生成了一个继承 Proxy 的代理类。

如果你再展开 h 属性看看它里面的成员变量。

图片

有 2 个比较重要的字段,targetSource 和 interfaces,我们挨个看下。

先来看看 targetSource 变量(ReferenceBean$DubboReferenceLazyInitTargetSource)。

看到它的类名,有没有似曾相识的感觉,类名上包含了 DubboReference 关键字,想必跟订阅流程中的引用有关。而且类名的构成,由 $ 符号隔开了,可以说明 DubboReferenceLazyInitTargetSource 是 ReferenceBean 的一个内部类。

进去瞄一眼,看看猜想对不对。

java
///////////////////////////////////////////////////                  
// org.apache.dubbo.config.spring.ReferenceBean.DubboReferenceLazyInitTargetSource
// 懒加载初始化的一种操作
///////////////////////////////////////////////////
private class DubboReferenceLazyInitTargetSource extends AbstractLazyCreationTargetSource {
    @Override
    protected Object createObject() throws Exception {
        // 创建对象,其实就是创建代理对象,拿到远程引用
        return getCallProxy();
    }
    @Override
    public synchronized Class<?> getTargetClass() {
        return getInterfaceClass();
    }
}
                  ↓
///////////////////////////////////////////////////                  
// org.apache.dubbo.config.spring.ReferenceBean#getCallProxy
// 直接通过 referenceConfig 的 get 方法获取接口引用的代理对象
///////////////////////////////////////////////////                  
private Object getCallProxy() throws Exception {
    if (referenceConfig == null) {
        throw new IllegalStateException("ReferenceBean is not ready yet, please make sure to call reference interface method after dubbo is started.");
    }
    //get reference proxy
    return referenceConfig.get();
}

进去之后才发现,原来内部类是一层壳,核心创建对象的逻辑还是 ReferenceConfig 的 get 方法。

我们再来看 interfacaces 变量。

图片

这里除了有我们调用下游的接口 DemoFacade,还有一个回声测试接口(EchoService)和一个销毁引用的接口(Destroyable)。

这俩接口我们没见过,却被 Dubbo 在创建代理的过程中追加进来了,可以说明一点,我们可以拿着 demoFacade 变量强转为 EchoService,或者强转为 Destroyable,强转后就可以调用对应接口的方法,实现了一个代理类拥有多态功能的效果。

这个多态功能的支持,得归功于 JDK 的 Proxy 类,它在创建代理对象的 newProxyInstance 方法中,支持入参传入一个接口数组,Dubbo 框架在创建代理时,不但传入了 DemoFacade 接口,还传入了 EchoService 和 Destroyable 接口,生成的代理对象就有了另外两种接口的行为能力。

java
///////////////////////////////////////////////////                  
// java.lang.reflect.Proxy#newProxyInstance
// JDK 的 Proxy 类中,创建代理对象的方法
///////////////////////////////////////////////////

@CallerSensitive
public static Object newProxyInstance(ClassLoader loader,
                                      Class<?>[] interfaces,
                                      InvocationHandler h)
    throws IllegalArgumentException
{ 省略部分实现代码... }

好,简单整理一下,刚开始断点探索,我们就接触到了“高深”的 JDK 代理对象,解开了开头的一个疑惑, demoFacade 在运行时并非我们所见的 DemoFacade 类型,而是由 JDK 动态代理生成的一个代理对象类型。

碰巧发现,在生成的代理对象中,targetSource 成员变量创建的底层核心逻辑还是 ReferenceConfig 的 get 方法,不得不说 ReferenceConfig 是消费方引用下游接口逻辑中非常重要的一个类。

同时还认识了 EchoService 和 Destroyable 两个接口,让我们使用 demoFacade 时不仅可以调用 sayHello 方法,还可以强转为这两个接口调用不同的方法,使得一个 demoFacade 变量拥有三能能力,这就是代理增强的魅力所在。

图片

2. InvokerInvocationHandler ​

接下来,我们继续 Debug 进入 sayHello 方法中,发现了一个名字有点眼熟的 InvokerInvocationHandler 类,虽然不知道具体能做啥,但是发现它实现了 JDK 代理中的 InvocationHandler 接口,所以,我们可以认为,这个 InvokerInvocationHandler 类是 Dubbo 框架接收 JDK 代理触发调用的入口。

来看看 InvokerInvocationHandler 的 invoke 方法,验证一下。

java
///////////////////////////////////////////////////                  
// org.apache.dubbo.rpc.proxy.InvokerInvocationHandler#invoke
// JDK 代理被触发调用后紧接着就开始进入 Dubbo 框架的调用,
// 因此跟踪消费方调用的入口,一般直接搜索这个 InvokerInvocationHandler 即可,
// 再说一点,这个 InvokerInvocationHandler 继承了 InvocationHandler 接口。
///////////////////////////////////////////////////
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
    // 如果方法所在的类是 Object 类型的话,则不做任何处理
    // 毕竟 Object 类型是 JDK 源码中的类,也不是 Dubbo 框架处理的重点
    if (method.getDeclaringClass() == Object.class) {
        return method.invoke(invoker, args);
    }
    
    // 获取方法名,方法参数类型
    String methodName = method.getName();
    Class<?>[] parameterTypes = method.getParameterTypes();
    // 如果参数类型个数是 0 个的话,则表示是无参数方法
    // 既然是无参数方法的话,对于一些特殊的方法需要尽可能的在入口处解决
    // 因此对于 Object 中的 toString、hashCode 方法,Destroyable 接口实现类的 $destroy 方法
    if (parameterTypes.length == 0) {
        if ("toString".equals(methodName)) {
            return invoker.toString();
        } else if ("$destroy".equals(methodName)) {
            invoker.destroy();
            return null;
        } else if ("hashCode".equals(methodName)) {
            return invoker.hashCode();
        }
    } 
    // 如果方法参数个数为 1 个,并且还是 equals 方法,则也不是 Dubbo 框架处理的重点
    else if (parameterTypes.length == 1 && "equals".equals(methodName)) {
        return invoker.equals(args[0]);
    }
    
    // 还能来到这里,说明这才是 Dubbo 框架需要介入处理的方法
    // 前面那么多提前返回的逻辑,在前几次阅读时,根本不需要细看,可以直接跳过不看
    RpcInvocation rpcInvocation = new RpcInvocation(serviceModel, method, invoker.getInterface().getName(), protocolServiceKey, args);
    if (serviceModel instanceof ConsumerModel) {
        rpcInvocation.put(Constants.CONSUMER_MODEL, serviceModel);
        rpcInvocation.put(Constants.METHOD_MODEL, ((ConsumerModel) serviceModel).getMethodModel(method));
    }
    // 然后转手就把逻辑全部收口到一个 InvocationUtil 类中,
    // 从命名也看得出,就是一个调用的工具类
    return InvocationUtil.invoke(invoker, rpcInvocation);
}

按12字方针,从上往下大致浏览一遍。

  • 从代码表面的流程看,一些 toString、$destroy、hashCode 等方法做了特殊的逻辑提前返回了,最终调用 InvocationUtil 的 invoke 方法进行了收口处理。
  • 从方法的入参和返回值看,有方法对象、入参数据、代理对象,满足了反射调用的基本要素,调用后返回的数据,自然就是我们需要的结果。
  • 再看方法实现体的一些细节,发现最终并没有使用 proxy 代理对象,而是使用了 invoker + rpcInvocation 传入 InvocationUtil 工具类,完成了逻辑收口。

再进入 InvocationUtil,忽略与入参无关的逻辑,最终发现把 rpcInvocation 对象传入了 invoker 的 invoke 方法中,继续走后续调用逻辑。

图片

3. MigrationInvoker ​

Debug 一直往后走,来到了一个 MigrationInvoker 的调用类,从类的名字上看是“迁移”的意思,有点 Dubbo2 与 Dubbo3 新老兼容迁移的味道。

那来看看 MigrationInvoker 的 invoke 方法代码逻辑。

java
///////////////////////////////////////////////////                  
// org.apache.dubbo.registry.client.migration.MigrationInvoker#invoke
// 迁移兼容调用器
///////////////////////////////////////////////////
@Override
public Result invoke(Invocation invocation) throws RpcException {
    if (currentAvailableInvoker != null) {
        if (step == APPLICATION_FIRST) {
            // call ratio calculation based on random value
            if (promotion < 100 && ThreadLocalRandom.current().nextDouble(100) > promotion) {
                return invoker.invoke(invocation);
            }
        }
        return currentAvailableInvoker.invoke(invocation);
    }

    switch (step) {
        case APPLICATION_FIRST:
            if (checkInvokerAvailable(serviceDiscoveryInvoker)) {
                currentAvailableInvoker = serviceDiscoveryInvoker;
            } else if (checkInvokerAvailable(invoker)) {
                currentAvailableInvoker = invoker;
            } else {
                currentAvailableInvoker = serviceDiscoveryInvoker;
            }
            break;
        case FORCE_APPLICATION:
            currentAvailableInvoker = serviceDiscoveryInvoker;
            break;
        case FORCE_INTERFACE:
        default:
            currentAvailableInvoker = invoker;
    }

    return currentAvailableInvoker.invoke(invocation);
}

继续按 12 字方针分析。

  • 从代码流程看,从上到下都在针对 step 变量值,进行 if…else 判断走不走分支处理,说明这里针对 step 进行了逻辑分发处理。
  • 从方法的入参和返回值来看,入参 invocation 是一个富含所有请求信息的接口,返参也是一个抽象的 Result 接口,意思很直白,根据参数拿到结果,只是入参和返参是接口,体现了抽象的概念。
  • 再看看方法实现体的一些细节,通过针对 APPLICATION_FIRST、FORCE_APPLICATION、FORCE_INTERFACE 来进行分发处理,这不就是消费方设置 dubbo.application.service-discovery.migration 属性,进行新老订阅方案兼容的值么?根据不同的属性值,选择对应可用的 invoker 进行继续后续调用。

图片

4. MockClusterInvoker ​

然后走进”step == APPLICATION_FIRST“的分支逻辑,进入 currentAvailableInvoker 的 invoke 方法,来到了 MockClusterInvoker 这个类,看名字是“模拟集群调用”的意思。

本意是干啥的呢,我们继续看它的 invoke 方法逻辑。

java
///////////////////////////////////////////////////                  
// org.apache.dubbo.rpc.cluster.support.wrapper.MockClusterInvoker#invoke
// 调用异常时进行使用mock逻辑来处理数据的返回
///////////////////////////////////////////////////
@Override
public Result invoke(Invocation invocation) throws RpcException {
    Result result;
    // 从远程引用的url中看看有没有 mock 属性
    String value = getUrl().getMethodParameter(invocation.getMethodName(), "mock", Boolean.FALSE.toString()).trim();
    // mock 属性值为空的话,相当于没有 mock 逻辑,则直接继续后续逻辑调用
    if (ConfigUtils.isEmpty(value)) {
        //no mock
        result = this.invoker.invoke(invocation);
    } 
    // 如果 mock 属性值是以 force 开头的话
    else if (value.startsWith("force")) {
        // 那么就直接执行 mock 调用逻辑,
        // 用事先准备好的模拟逻辑或者模拟数据返回
        //force:direct mock
        result = doMockInvoke(invocation, null);
    } 
    // 还能来到这说明只是想在调用失败的时候尝试一下 mock 逻辑
    else {
        //fail-mock
        try {
            // 先正常执行业务逻辑调用
            result = this.invoker.invoke(invocation);
            //fix:#4585
            // 当业务逻辑执行有异常时,并且这个异常类属于RpcException或RpcException子类时,
            // 还有异常的原因如果是 Dubbo 框架层面的业务异常时,则不做任何处理。
            // 如果不是业务异常的话,则会继续尝试执行 mock 业务逻辑
            if(result.getException() != null && result.getException() instanceof RpcException){
                RpcException rpcException= (RpcException)result.getException();
                // 如果异常是 Dubbo 系统层面所认为的业务异常时,就不错任何处理
                if(rpcException.isBiz()){
                    throw  rpcException;
                }else {
                    // 能来到这里说明是不是业务异常的话,那就执行模拟逻辑
                    result = doMockInvoke(invocation, rpcException);
                }
            }
        } catch (RpcException e) {
            // 业务异常直接往上拋
            if (e.isBiz()) {
                throw e;
            }
            // 不是 Dubbo 层面所和认为的异常信息时代,
            // 直接
            result = doMockInvoke(invocation, e);
        }
    }
    return result;
}

照例按12字方针分析。

  • 从代码流程看,整体上就三个 if 逻辑分支,no mock、force mock 和 fail mock,分别进行了不同逻辑的调用。
  • 从方法的入参和返回值看,入参和返参和 MigrationInvoker 一样,都是重写了 invoke 方法,只是内部逻辑不一样,不同实现类做各自的事情。
  • 再看看方法实现体的一些细节,mock 属性值(除了空值),当以 force 开头时,会直接执行 Mock 业务逻辑,其他情况还是先尝试正常后续调用,如果出现了非业务异常时,就尝试执行 Mock 业务逻辑。

图片

5. 过滤器链 ​

这里,看到 Cluster 这个关键字,想必你也想到了它是一个 SPI 接口,那在“发布流程”中我们也学过,远程导出和远程引用的时候,会用过滤器链把 invoker 层层包装起来。

那么我们就接着断点下去,发现确实进入 FutureFilter、MonitorFilter 等过滤器,这也证明了过滤器链包裹消费方 invoker 的存在。

图片

6. FailoverClusterInvoker ​

断点一层层走完了所有的过滤器,接着又来到了 FailoverClusterInvoker 这个类,从名字上一看就是在“温故知新”中接触的故障转移策略,是不是有点好奇故障到底是怎么转移的?

我们不妨继续断点,进入它的 doInvoke 方法看看。

java
///////////////////////////////////////////////////                  
// org.apache.dubbo.rpc.cluster.support.FailoverClusterInvoker#doInvoke
// 故障转移策略的核心逻辑实现类
///////////////////////////////////////////////////
@Override
@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);
    // 获取此次调用的方法名
    String methodName = RpcUtils.getMethodName(invocation);
    // 通过方法名计算获取重试次数
    int len = calculateInvokeTimes(methodName);
    // retry loop.
    // 循环计算得到的 len 次数
    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.
        // 从第2次循环开始,会有一段特殊的逻辑处理
        if (i > 0) {
            // 检测 invoker 是否被销毁了
            checkWhetherDestroyed();
            // 重新拿到调用接口的所有提供者列表集合,
            // 粗俗理解,就是提供该接口服务的每个提供方节点就是一个 invoker 对象
            copyInvokers = list(invocation);
            // check again
            // 再次检查所有拿到的 invokes 的一些可用状态
            checkInvokers(copyInvokers, invocation);
        }
        // 选择其中一个,即采用了负载均衡策略从众多 invokers 集合中挑选出一个合适可用的
        Invoker<T> invoker = select(loadbalance, invocation, copyInvokers, invoked);
        invoked.add(invoker);
        // 设置 RpcContext 上下文
        RpcContext.getServiceContext().setInvokers((List) invoked);
        boolean success = false;
        try {
            // 得到最终的 invoker 后也就明确了需要调用哪个提供方节点了
            // 反正继续走后续调用流程就是了
            Result result = invokeWithContext(invoker, invocation);
            // 如果没有抛出异常的话,则认为正常拿到的返回数据
            // 那么设置调用成功标识,然后直接返回 result 结果
            success = true;
            return result;
        } catch (RpcException e) {
            // 如果是 Dubbo 框架层面认为的业务异常,那么就直接抛出异常
            if (e.isBiz()) { // biz exception.
                throw e;
            }
            // 其他异常的话,则不继续抛出异常,那么就意味着还可以有机会再次循环调用
            le = e;
        } catch (Throwable e) {
            le = new RpcException(e.getMessage(), e);
        } finally {
            // 如果没有正常返回拿到结果的话,那么把调用异常的提供方地址信息记录起来
            if (!success) {
                providers.add(invoker.getUrl().getAddress());
            }
        }
    }
    
    // 如果 len 次循环仍然还没有正常拿到调用结果的话,
    // 那么也不再继续尝试调用了,直接索性把一些需要开发人员关注的一些信息写到异常描述信息中,通过异常方式拋出去
    throw new RpcException(le.getCode(), "Failed to invoke the method "
            + methodName + " 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.getMessage(), le.getCause() != null ? le.getCause() : le);
}

代码流程分析也比较简单。

  • 从代码流程看,主要就是一个大的 for 循环,循环体中进行了 select 操作,拿到了一个合适的 invoker,发起后续逻辑调用。
  • 从方法的入参和返回值看,入参是 invocation、invokers、loadbalance 三个参数,猜测应该就是利用 loadbalance 负载均衡器,从 invokers 集合中,选择一个 invoker来发送 invocation 数据,发送完成后得到了返参的 Result 结果。
  • 再看看方法实现体的一些细节,通过计算 retries 属性值得到重试次数并循环,每次循环都是利用负载均衡器选择一个进行调用,如果出现非业务异常,继续循环调用,直到所有次数循环完,还是没能拿到结果的话就会抛出 RpcException 异常了。

图片

7. DubboInvoker ​

现在你应该知道“故障转移策略”的深层含义了吧,就是根据预先设置好的重试次数,当调用发生非业务异常时,按照负载均衡策略继续选择一个 invoker,再次发起调用。

了解完故障转移策略,我们继续 Debug,结果来到了 DubboInvoker 这个类。

java
///////////////////////////////////////////////////                  
// org.apache.dubbo.rpc.protocol.dubbo.DubboInvoker#doInvoke
// 按照dubbo协议发起调用实现类
///////////////////////////////////////////////////
@Override
protected Result doInvoke(final Invocation invocation) throws Throwable {
    // 获取此次调用的方法名
    RpcInvocation inv = (RpcInvocation) invocation;
    final String methodName = RpcUtils.getMethodName(invocation);
    inv.setAttachment(PATH_KEY, getUrl().getPath());
    inv.setAttachment(VERSION_KEY, version);
    // 获取发送数据的客户端
    ExchangeClient currentClient;
    if (clients.length == 1) {
        currentClient = clients[0];
    } else {
        currentClient = clients[index.getAndIncrement() % clients.length];
    }
    try {
        // 看看是单程发送不需要等待响应,还是发送完了后需要等待响应
        boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);
        // 获取超时时间,这个之前在“配置加载顺序”接触过这个方法
        int timeout = calculateTimeout(invocation, methodName);
        invocation.setAttachment(TIMEOUT_KEY, timeout);
        // 单程发送不需要等待响应
        if (isOneway) {
            boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
            currentClient.send(inv, isSent);
            return AsyncRpcResult.newDefaultAsyncResult(invocation);
        } else {
            // 发送完了之后需要等待响应
            ExecutorService executor = getCallbackExecutor(getUrl(), inv);
            // 操作 currentClient 发送了一个 request 请求,
            // 然后接收了一个 CompletableFuture 对象,说明这里存在异步操作
            CompletableFuture<AppResponse> appResponseFuture =
                    currentClient.request(inv, timeout, executor).thenApply(obj -> (AppResponse) obj);
            // save for 2.6.x compatibility, for example, TraceFilter in Zipkin uses com.alibaba.xxx.FutureAdapter
            FutureContext.getContext().setCompatibleFuture(appResponseFuture);
            AsyncRpcResult result = new AsyncRpcResult(appResponseFuture, inv);
            result.setExecutor(executor);
            return result;
        }
    } catch (TimeoutException e) {
        throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
    } catch (RemotingException e) {
        throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
    }
}
  • 从代码流程看,拿到了一个交换数据的客户端类,然后走两个发送数据的分支,一条分支逻辑单程调用不需要响应,一条有响应。
  • 从方法的入参和返回值来看,没有什么特别,根据请求数据想办法发起调用,拿到 Result 结果。

但是当前的类名是处理 Dubbo 协议的调用?有点好奇,我们看看这个 doInvoke 有哪些实现类,发现了有 InjvmInvoker、GrpcInvoker 等实现类,也证实了当前的 DubboInvoker 是专门处理 Dubbo 协议调用的类。

  • 再看看方法实现体的一些细节,计算了超时时间值,单程发送,不需要响应好说,直接调用交互数据客户端类的 send 方法就行,需要有响应的分支逻辑还定义了线程池,调用的是 request 方法进行发送。

不过,两条分支最终都返回了一个异步结果对象,可以看出当前的 DubboInvoker 旨在屏蔽协议之间的差异,让上层调用方不感知协议间的不同。

图片

8. ReferenceCountExchangeClient ​

因为我们没有单独设置调用不需要响应,就继续断点进入 currentClient 的 request 方法,看看到底是怎么发送的,来到了 ReferenceCountExchangeClient 类。

java
///////////////////////////////////////////////////                  
// org.apache.dubbo.rpc.protocol.dubbo.ReferenceCountExchangeClient#request(java.lang.Object, int, java.util.concurrent.ExecutorService)
// 这里将构建好的 request 对象发送出去,然后拿到了一个 CompletableFuture 异步化的对象
///////////////////////////////////////////////////
@Override
public CompletableFuture<Object> request(Object request, int timeout, ExecutorService executor) throws RemotingException {
    // client为:HeaderExchangeClient
    return client.request(request, timeout, executor);
}
  • 从代码流程来看,这个类中的 request 方法也很简单,把入参数据直接全部交给了当前类的成员变量 client 处理。
  • 从方法的入参和返回值看,入参是请求对象、超时时间、线程池对象,返参是 CompletableFuture 对象。从线程池和 Future 对象,充分说明了这个 request 方法进行了同步转异步的操作。
  • 再看看方法实现体的一些细节,从断点展示的变量值细节看,只是发现了 client 的类型属于 HeaderExchangeClient,也就意味着 HeaderExchangeClient 完成了同步转异步的逻辑操作。

图片

9. NettyClient ​

仍然没有看到数据是怎么发送的,我们继续深入 HeaderExchangeClient 的 request 方法中,Debug 几步后发现,进入了抽象类 AbstractPeer 的 send 方法。

java
///////////////////////////////////////////////////                  
// org.apache.dubbo.remoting.transport.AbstractPeer#send
// this:NettyClient
// NettyClient 负责和提供方服务建立连接、发送数据等操作
///////////////////////////////////////////////////
@Override
public void send(Object message) throws RemotingException {
    send(message, url.getParameter(Constants.SENT_KEY, false));
}

逻辑也挺简单。

  • 从代码流程来看,没有过多的逻辑包装,直接把入参交给了另外一个重载的 send 方法。
  • 从方法的入参和返回值来看,入参是请求对象,没有返参,并且方法上声明了 RemotingException 异常,说明这里离数据的发送应该很近了。至于没有返回值,主要是前面由同步转了异步,没有返回值也正常。
  • 再看看方法实现体的一些细节,抽象类看不出什么,但是从堆栈变量的引用值可以发现,当前的抽象类是 NettyClient 的父类。

那我们从 NettyClient 的 send 方法细看,结果发现了最终调用了 NioSocketChannel 的 writeAndFlush 方法,这个不就是 Netty 网络通信框架的 API 么?

到这里,我们在代码层面解释了数据原来是通过 Netty 框架的 NioSocketChannel 发送出去的。

图片

10. NettyCodecAdapter ​

一旦进入到 Netty 框架,再通过断点一步步跟踪数据就有点难了,因为 Netty 框架内部处处都是“线程+队列”的异步操作方式,这里我们走个捷径,进入 NettyClient 类,找找初始化 NettyClient 的相关 Netty 层面的配置。

你会找到一个 NettyCodecAdapter 类,对数据进行编解码,直接在类中的 encode 方法打个断点,等断点过来。

java
///////////////////////////////////////////////////                  
// org.apache.dubbo.remoting.transport.netty4.NettyCodecAdapter.InternalEncoder#encode
// 将对象按照一定的序列化方式发送出去
///////////////////////////////////////////////////
private class InternalEncoder extends MessageToByteEncoder {

    @Override
    protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception {
        ChannelBuffer buffer = new NettyBackedChannelBuffer(out);
        Channel ch = ctx.channel();
        NettyChannel channel = NettyChannel.getOrAddChannel(ch, url, handler);
        codec.encode(channel, buffer, msg);
    }
}

可以看到,编码的方法简单干脆。

  • 从代码流程看,调用了一个 codec 编码器的变量,对入参编码处理。
  • 从方法的入参和返回值看,msg 是请求数据对象,out 变量是 ByteBuf 缓冲区,应该是将 msg 编码之后的数据流。
  • 再看看方法实现体的一些细节,没什么特别之处,无非就是走后续编解码器的编码方法,编码完成后,再通过 Netty 框架把数据流发送出去。

图片

调用流程的其他案例 ​

经过漫长的十步断点,我们把消费方的调用流程大致串了起来,虽说过程有点曲折,但最终的艰辛是值得的,对于你理解框架调用的来龙去脉很重要,而且未来你会体会到,当调用环节发生了异常,你只要认真回忆一下异常在哪个环节,然后结合代码简单分析一下,很快就能分析出原因来。

消费方的调用流程就是这么一步步调试分析出来的,那还有哪些典型的流程,可以按这个思路一步步调试分析出调用流程呢?我就列举三个比较典型的,你课后可以练练手。

第一,Tomcat 容器接收请求流程,通过在 HttpServlet 的 service 方法打个断点,从调用堆栈的最下面开始,可以分析出 Tomcat 的整体架构。

第二,SpringMvc 处理请求的流程,通过在 Controller 的任意方法打个断点,就可以逐步分析出一个 SpringMvc 处理请求的大致流程框架。

第三,Spring 的 getBean 方法,通过这个方法的层层深入,可以分析出一个庞大的 Spring 对象生成体系,也能挖掘出非常多 Spring 各种操控对象的扩展机制。

总结 ​

今天,从一个被 @DubboReference 标识的常规成员变量开始,我们抛出了两个问题,变量的值类型在运行时是否是见到的类型,在消费方发起调用时会经过怎样的调用流程体系。

带着这两个问题,我们用最普通的 Debug 调试方式,借鉴 12 字方针深入跟踪调用流程的源码,边分析、边绘图,步步为营,最终梳理出了消费方调用流程的大体框架。你还可以和“源码框架”中学过的分层模块联系到一起复习。

图片

总结一下跟踪源码的 12 字方针。

  • “不钻细节:只看流程。”代码虽然多,但我们不必研究每个细节,要先捡方法中的重要分支流程,一些提前返回的边边角角的流程一概忽略不看。
  • “不看过程:只看结论。”每个方法的代码逻辑可长可短,我们可以重点研究这个方法需要什么入参,又能给出什么返参,以此推测方法在干什么,到底要完成一件什么样的事情,搞清楚并得出结论。
  • “再看细节:再看过程。”当你按照前两点认真调试后,大概的调用流程体系就清楚了。在此基础之上,再来细看被遗忘的边角料代码,有助于你进一步丰富调用流程图,体会源码细节中那些缜密的思维逻辑。

思考题 ​

留个作业,消费方的普通调用会经历哪些流程你已经十分熟悉了,可以研究消费方进行泛化调用时会经历哪些流程,以及泛化调用的底层是怎么实现的?动手练习一下。

期待看到你的回答,如果觉得今天的内容对你有帮助,也欢迎分享给身边的朋友一起讨论。我们下一讲见。

20 思考题参考 ​

上一期留了个作业,研究下消费方接收 zookeeper 服务端事件变更的地方在哪里。

想要解答这个问题,我们得从添加监听的起源看起,只有知道添加监听干了些什么事情,预留了什么回调,我们才能进一步找到接收的地方。

我们直接进入 doSubscribe 方法中看看,到底添加了什么监听?监听对应的实现类是哪个?

java
///////////////////////////////////////////////////                  
// org.apache.dubbo.registry.zookeeper.ZookeeperRegistry#doSubscribe
// 订阅的核心逻辑,读取 zk 目录下的数据,然后通知刷新内存中的数据
///////////////////////////////////////////////////
@Override
public void doSubscribe(final URL url, final NotifyListener listener) {
    try {
        checkDestroyed();
        // 因为这里用 * 号匹配,我们在真实的线上环境也不可能将服务接口配置为 * 号
        // 因此这里的 * 号逻辑暂且跳过,直接看后面的具体接口的逻辑
        if ("*".equals(url.getServiceInterface())) {
            // 省略其他部分代码...
        } 
        // 能来到这里,说明 ServiceInterface 不是 * 号
        // url.getServiceInterface() = com.hmilyylimh.cloud.facade.demo.DemoFacade
        else {
            CountDownLatch latch = new CountDownLatch(1);
            try {
                List<URL> urls = new ArrayList<>();
                // toCategoriesPath(url) 得出来的集合有以下几种:
                // 1、/dubbo/com.hmilyylimh.cloud.facade.demo.DemoFacade/providers
                // 2、/dubbo/com.hmilyylimh.cloud.facade.demo.DemoFacade/configurators
                // 3、/dubbo/com.hmilyylimh.cloud.facade.demo.DemoFacade/routers
                for (String path : toCategoriesPath(url)) {
                    ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.computeIfAbsent(url, k -> new ConcurrentHashMap<>());
                    ChildListener zkListener = listeners.computeIfAbsent(listener, k -> new RegistryChildListenerImpl(url, path, k, latch));
                    if (zkListener instanceof RegistryChildListenerImpl) {
                        ((RegistryChildListenerImpl) zkListener).setLatch(latch);
                    }
                    // 向 zk 创建持久化目录,一种容错方式,担心目录被谁意外的干掉了
                    zkClient.create(path, false);
                    
                    // !!!!!!!!!!!!!!!!
                    // 这段逻辑很重要了,添加对 path 目录的监听,
                    // 添加监听完成后,还能拿到 path 路径下所有的信息
                    // 那就意味着监听一旦添加完成,那么就能立马获取到该 DemoFacade 接口到底有多少个提供方节点
                    List<String> children = zkClient.addChildListener(path, zkListener);
                    // 将返回的信息全部添加到 urls 集合中
                    if (children != null) {
                        urls.addAll(toUrlsWithEmpty(url, path, children));
                    }
                }
                
                // 从 zk 拿到了所有的信息后,然后调用 notify 方法
                // url.get(0) = dubbo://192.168.100.183:28200/com.hmilyylimh.cloud.facade.demo.DemoFacade?anyhost=true&application=dubbo-20-subscribe-consumer&background=false&check=false&deprecated=false&dubbo=2.0.2&dynamic=true&generic=false&interface=com.hmilyylimh.cloud.facade.demo.DemoFacade&methods=sayHello,say&register-mode=interface&release=3.0.7&side=provider
                // url.get(1) = empty://192.168.100.183/com.hmilyylimh.cloud.facade.demo.DemoFacade?application=dubbo-20-subscribe-consumer&background=false&category=configurators&dubbo=2.0.2&interface=com.hmilyylimh.cloud.facade.demo.DemoFacade&methods=sayHello,say&pid=11560&qos.enable=false&release=3.0.7&side=consumer&sticky=false&timestamp=1670846788876
                // url.get(2) = empty://192.168.100.183/com.hmilyylimh.cloud.facade.demo.DemoFacade?application=dubbo-20-subscribe-consumer&background=false&category=routers&dubbo=2.0.2&interface=com.hmilyylimh.cloud.facade.demo.DemoFacade&methods=sayHello,say&pid=11560&qos.enable=false&release=3.0.7&side=consumer&sticky=false&timestamp=1670846788876
                notify(url, listener, urls);
            } finally {
                // tells the listener to run only after the sync notification of main thread finishes.
                latch.countDown();
            }
        }
    } catch (Throwable e) {
        throw new RpcException("Failed to subscribe " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
    }
}

再次回到了这段代码中,这次我们是有针对性地寻找监听器类了,眼尖的你,相信很快就看到了 RegistryChildListenerImpl 这个类,从名字上看,是子节点的监听实现类,既然能在这段实现全流程逻辑,那毫无疑问得进入 RegistryChildListenerImpl 这个类去看看。

java
///////////////////////////////////////////////////                  
// org.apache.dubbo.registry.zookeeper.ZookeeperRegistry.RegistryChildListenerImpl
// 子节点目录监听变更类
///////////////////////////////////////////////////
private class RegistryChildListenerImpl implements ChildListener {
    private RegistryNotifier notifier;
    private long lastExecuteTime;
    private volatile CountDownLatch latch;

    public RegistryChildListenerImpl(URL consumerUrl, String path, NotifyListener listener, CountDownLatch latch) {
        this.latch = latch;
        notifier = new RegistryNotifier(getUrl(), ZookeeperRegistry.this.getDelay()) {
            @Override
            public void notify(Object rawAddresses) {
                // 获取延时时间
                long delayTime = getDelayTime();
                // 若延时时间小于或等于 0 的话,就直接执行 RegistryNotifier 内部类的 doNotify 方法
                if (delayTime <= 0) {
                    this.doNotify(rawAddresses);
                } else {
                    // 否则就计算【当前时间】减去【上一次执行时间】得到一个差值
                    // 若 delayTime 大于这个差值的话,那就继续睡眠下
                    // 否则又立马执行 RegistryNotifier 内部类的 doNotify 方法
                    long interval = delayTime - (System.currentTimeMillis() - lastExecuteTime);
                    if (interval > 0) {
                        try {
                            Thread.sleep(interval);
                        } catch (InterruptedException e) {
                            // ignore
                        }
                    }
                    lastExecuteTime = System.currentTimeMillis();
                    this.doNotify(rawAddresses);
                }
            }

            @Override
            protected void doNotify(Object rawAddresses) {
                ZookeeperRegistry.this.notify(consumerUrl, listener, ZookeeperRegistry.this.toUrlsWithEmpty(consumerUrl, path, (List<String>) rawAddresses));
            }
        };
    }
    // 设置计数器概念的同步工具类
    public void setLatch(CountDownLatch latch) {
        this.latch = latch;
    }
    // 感知到节点有变更,该方法是别其他地方回调的
    @Override
    public void childChanged(String path, List<String> children) {
        // path = /dubbo/com.hmilyylimh.cloud.facade.demo.DemoFacade/providers
        // children.get(0) = dubbo://192.168.100.183:28200/com.hmilyylimh.cloud.facade.demo.DemoFacade?anyhost=true&application=dubbo-20-subscribe-consumer&background=false&deprecated=false&dubbo=2.0.2&dynamic=true&generic=false&interface=com.hmilyylimh.cloud.facade.demo.DemoFacade&methods=sayHello,say&pid=8552&register-mode=interface&release=3.0.7&side=provider&timestamp=1670859852464
        // children.get(1) = dubbo://192.168.100.183:28200/com.hmilyylimh.cloud.facade.demo.DemoFacade?anyhost=true&application=dubbo-20-subscribe-consumer&background=false&deprecated=false&dubbo=2.0.2&dynamic=true&generic=false&interface=com.hmilyylimh.cloud.facade.demo.DemoFacade&methods=sayHello,say&pid=10568&register-mode=interface&release=3.0.7&side=provider&timestamp=1670859874187
        // children.get(2) = dubbo://192.168.100.183:28200/com.hmilyylimh.cloud.facade.demo.DemoFacade?anyhost=true&application=dubbo-20-subscribe-consumer&background=false&deprecated=false&dubbo=2.0.2&dynamic=true&generic=false&interface=com.hmilyylimh.cloud.facade.demo.DemoFacade&methods=sayHello,say&pid=12988&register-mode=interface&release=3.0.7&side=provider&timestamp=1670859893317
        try {
            latch.await();
        } catch (InterruptedException e) {
            logger.warn("Zookeeper children listener thread was interrupted unexpectedly, may cause race condition with the main thread.");
        }
        // 节点变更通知,直接调用的是上面 RegistryNotifier 内部类中的 notify 方法
        notifier.notify(children);
    }
}

通过这段代码我们看到了若 childChanged 方法发生了回调,那么最终会调用到 ZookeeperRegistry 的 notify 方法。虽然我们找到了 Dubbo 针对 zookeeper 目录监听回调的地方,但是到底是谁调用了这个 childChanged 方法呢?

这个时候,又得使出我们的 Debug 绝招了,你可以按步骤操作。

  • 步骤一,启动提供方。
  • 步骤二,启动消费方。
  • 步骤三,在 RegistryChildListenerImpl 类中的 childChanged 方法打个断点。
  • 步骤四,关闭提供方。
  • 步骤五,启动提供方。

就能看到断点的到来了。

图片

通过对断点调用栈的分析,真相也就清晰了,原来 dubbo 框架引用了 curator 插件来与 zookeeper 进行交互,curator 插件会实时感知 zookeeper 服务端发送的事件变更,然后 curator 再通过一定的回调处理,就调用到了 RegistryChildListenerImpl 类中的 childChanged 方法。

最近更新