2.1 服务调用方式

Dubbo 支持同步和异步两种调用方式，其中异步调用还可细分为“有返回值”的异步调用和“无返回值”的异步调用。所谓“无返回值”异步调用是指服务消费方只管调用，但不关心调用结果，此时 Dubbo 会直接返回一个空的 RpcResult。若要使用异步特性，需要服务消费方手动进行配置。默认情况下，Dubbo 使用同步调用方式。

本节以及其他章节将会使用 Dubbo 官方提供的 Demo 分析整个调用过程，下面我们从 DemoService 接口的代理类开始进行分析。Dubbo 默认使用 Javassist 框架为服务接口生成动态代理类，因此我们需要先将代理类进行反编译才能看到源码。这里使用阿里开源 Java 应用诊断工具 Arthas 反编译代理类，结果如下：

  1. /**
  2.  * Arthas 反编译步骤：
  3.  * 1. 启动 Arthas
  4.  *    java -jar arthas-boot.jar
  5.  *
  6.  * 2. 输入编号选择进程
  7.  *    Arthas 启动后，会打印 Java 应用进程列表，如下：
  8.  *    [1]: 11232 org.jetbrains.jps.cmdline.Launcher
  9.  *    [2]: 22370 org.jetbrains.jps.cmdline.Launcher
  10.  *    [3]: 22371 com.alibaba.dubbo.demo.consumer.Consumer
  11.  *    [4]: 22362 com.alibaba.dubbo.demo.provider.Provider
  12.  *    [5]: 2074 org.apache.zookeeper.server.quorum.QuorumPeerMain
  13.  * 这里输入编号 3，让 Arthas 关联到启动类为 com.....Consumer 的 Java 进程上
  14.  *
  15.  * 3. 由于 Demo 项目中只有一个服务接口，因此此接口的代理类类名为 proxy0，此时使用 sc 命令搜索这个类名。
  16.  *    $ sc *.proxy0
  17.  *    com.alibaba.dubbo.common.bytecode.proxy0
  18.  *
  19.  * 4. 使用 jad 命令反编译 com.alibaba.dubbo.common.bytecode.proxy0
  20.  *    $ jad com.alibaba.dubbo.common.bytecode.proxy0
  21.  *
  22.  * 更多使用方法请参考 Arthas 官方文档：
  23.  *   https://alibaba.github.io/arthas/quick-start.html
  24.  */
  25. public class proxy0 implements ClassGenerator.DC, EchoService, DemoService {
  26.     // 方法数组
  27.     public static Method[] methods;
  28.     private InvocationHandler handler;
  30.     public proxy0(InvocationHandler invocationHandler) {
  31.         this.handler = invocationHandler;
  32.     }
  34.     public proxy0() {
  35.     }
  37.     public String sayHello(String string) {
  38.         // 将参数存储到 Object 数组中
  39.         Object[] arrobject = new Object[]{string};
  40.         // 调用 InvocationHandler 实现类的 invoke 方法得到调用结果
  41.         Object object = this.handler.invoke(this, methods[0], arrobject);
  42.         // 返回调用结果
  43.         return (String)object;
  44.     }
  46.     /** 回声测试方法 */
  47.     public Object $echo(Object object) {
  48.         Object[] arrobject = new Object[]{object};
  49.         Object object2 = this.handler.invoke(this, methods[1], arrobject);
  50.         return object2;
  51.     }
  52. }

如上，代理类的逻辑比较简单。首先将运行时参数存储到数组中，然后调用 InvocationHandler 接口实现类的 invoke 方法，得到调用结果，最后将结果转型并返回给调用方。关于代理类的逻辑就说这么多，继续向下分析。

  1. public class InvokerInvocationHandler implements InvocationHandler {
  3.     private final Invoker<?> invoker;
  5.     public InvokerInvocationHandler(Invoker<?> handler) {
  6.         this.invoker = handler;
  7.     }
  9.     @Override
  10.     public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
  11.         String methodName = method.getName();
  12.         Class<?>[] parameterTypes = method.getParameterTypes();
  14.         // 拦截定义在 Object 类中的方法（未被子类重写），比如 wait/notify
  15.         if (method.getDeclaringClass() == Object.class) {
  16.             return method.invoke(invoker, args);
  17.         }
  19.         // 如果 toString、hashCode 和 equals 等方法被子类重写了，这里也直接调用
  20.         if ("toString".equals(methodName) && parameterTypes.length == 0) {
  21.             return invoker.toString();
  22.         }
  23.         if ("hashCode".equals(methodName) && parameterTypes.length == 0) {
  24.             return invoker.hashCode();
  25.         }
  26.         if ("equals".equals(methodName) && parameterTypes.length == 1) {
  27.             return invoker.equals(args[0]);
  28.         }
  30.         // 将 method 和 args 封装到 RpcInvocation 中，并执行后续的调用
  31.         return invoker.invoke(new RpcInvocation(method, args)).recreate();
  32.     }
  33. }

InvokerInvocationHandler 中的 invoker 成员变量类型为 MockClusterInvoker，MockClusterInvoker 内部封装了服务降级逻辑。下面简单看一下：

  1. public class MockClusterInvoker<T> implements Invoker<T> {
  3.     private final Invoker<T> invoker;
  5.     public Result invoke(Invocation invocation) throws RpcException {
  6.         Result result = null;
  8.         // 获取 mock 配置值
  9.         String value = directory.getUrl().getMethodParameter(invocation.getMethodName(), Constants.MOCK_KEY, Boolean.FALSE.toString()).trim();
  10.         if (value.length() == 0 || value.equalsIgnoreCase("false")) {
  11.             // 无 mock 逻辑，直接调用其他 Invoker 对象的 invoke 方法，
  12.             // 比如 FailoverClusterInvoker
  13.             result = this.invoker.invoke(invocation);
  14.         } else if (value.startsWith("force")) {
  15.             // force:xxx 直接执行 mock 逻辑，不发起远程调用
  16.             result = doMockInvoke(invocation, null);
  17.         } else {
  18.             // fail:xxx 表示消费方对调用服务失败后，再执行 mock 逻辑，不抛出异常
  19.             try {
  20.                 // 调用其他 Invoker 对象的 invoke 方法
  21.                 result = this.invoker.invoke(invocation);
  22.             } catch (RpcException e) {
  23.                 if (e.isBiz()) {
  24.                     throw e;
  25.                 } else {
  26.                     // 调用失败，执行 mock 逻辑
  27.                     result = doMockInvoke(invocation, e);
  28.                 }
  29.             }
  30.         }
  31.         return result;
  32.     }
  34.     // 省略其他方法
  35. }

服务降级不是本文重点，因此这里就不分析 doMockInvoke 方法了。考虑到前文已经详细分析过 FailoverClusterInvoker，因此本节略过 FailoverClusterInvoker，直接分析 DubboInvoker。

  1. public abstract class AbstractInvoker<T> implements Invoker<T> {
  3.     public Result invoke(Invocation inv) throws RpcException {
  4.         if (destroyed.get()) {
  5.             throw new RpcException("Rpc invoker for service ...");
  6.         }
  7.         RpcInvocation invocation = (RpcInvocation) inv;
  8.         // 设置 Invoker
  9.         invocation.setInvoker(this);
  10.         if (attachment != null && attachment.size() > 0) {
  11.             // 设置 attachment
  12.             invocation.addAttachmentsIfAbsent(attachment);
  13.         }
  14.         Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();
  15.         if (contextAttachments != null && contextAttachments.size() != 0) {
  16.             // 添加 contextAttachments 到 RpcInvocation#attachment 变量中
  17.             invocation.addAttachments(contextAttachments);
  18.         }
  19.         if (getUrl().getMethodParameter(invocation.getMethodName(), Constants.ASYNC_KEY, false)) {
  20.             // 设置异步信息到 RpcInvocation#attachment 中
  21.             invocation.setAttachment(Constants.ASYNC_KEY, Boolean.TRUE.toString());
  22.         }
  23.         RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
  25.         try {
  26.             // 抽象方法，由子类实现
  27.             return doInvoke(invocation);
  28.         } catch (InvocationTargetException e) {
  29.             // ...
  30.         } catch (RpcException e) {
  31.             // ...
  32.         } catch (Throwable e) {
  33.             return new RpcResult(e);
  34.         }
  35.     }
  37.     protected abstract Result doInvoke(Invocation invocation) throws Throwable;
  39.     // 省略其他方法
  40. }

上面的代码来自 AbstractInvoker 类，其中大部分代码用于添加信息到 RpcInvocation#attachment 变量中，添加完毕后，调用 doInvoke 执行后续的调用。doInvoke 是一个抽象方法，需要由子类实现，下面到 DubboInvoker 中看一下。

  1. public class DubboInvoker<T> extends AbstractInvoker<T> {
  3.     private final ExchangeClient[] clients;
  5.     protected Result doInvoke(final Invocation invocation) throws Throwable {
  6.         RpcInvocation inv = (RpcInvocation) invocation;
  7.         final String methodName = RpcUtils.getMethodName(invocation);
  8.         // 设置 path 和 version 到 attachment 中
  9.         inv.setAttachment(Constants.PATH_KEY, getUrl().getPath());
  10.         inv.setAttachment(Constants.VERSION_KEY, version);
  12.         ExchangeClient currentClient;
  13.         if (clients.length == 1) {
  14.             // 从 clients 数组中获取 ExchangeClient
  15.             currentClient = clients[0];
  16.         } else {
  17.             currentClient = clients[index.getAndIncrement() % clients.length];
  18.         }
  19.         try {
  20.             // 获取异步配置
  21.             boolean isAsync = RpcUtils.isAsync(getUrl(), invocation);
  22.             // isOneway 为 true，表示“单向”通信
  23.             boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);
  24.             int timeout = getUrl().getMethodParameter(methodName, Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
  26.             // 异步无返回值
  27.             if (isOneway) {
  28.                 boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
  29.                 // 发送请求
  30.                 currentClient.send(inv, isSent);
  31.                 // 设置上下文中的 future 字段为 null
  32.                 RpcContext.getContext().setFuture(null);
  33.                 // 返回一个空的 RpcResult
  34.                 return new RpcResult();
  35.             } 
  37.             // 异步有返回值
  38.             else if (isAsync) {
  39.                 // 发送请求，并得到一个 ResponseFuture 实例
  40.                 ResponseFuture future = currentClient.request(inv, timeout);
  41.                 // 设置 future 到上下文中
  42.                 RpcContext.getContext().setFuture(new FutureAdapter<Object>(future));
  43.                 // 暂时返回一个空结果
  44.                 return new RpcResult();
  45.             } 
  47.             // 同步调用
  48.             else {
  49.                 RpcContext.getContext().setFuture(null);
  50.                 // 发送请求，得到一个 ResponseFuture 实例，并调用该实例的 get 方法进行等待
  51.                 return (Result) currentClient.request(inv, timeout).get();
  52.             }
  53.         } catch (TimeoutException e) {
  54.             throw new RpcException(..., "Invoke remote method timeout....");
  55.         } catch (RemotingException e) {
  56.             throw new RpcException(..., "Failed to invoke remote method: ...");
  57.         }
  58.     }
  60.     // 省略其他方法
  61. }

上面的代码包含了 Dubbo 对同步和异步调用的处理逻辑，搞懂了上面的代码，会对 Dubbo 的同步和异步调用方式有更深入的了解。Dubbo 实现同步和异步调用比较关键的一点就在于由谁调用 ResponseFuture 的 get 方法。同步调用模式下，由框架自身调用 ResponseFuture 的 get 方法。异步调用模式下，则由用户调用该方法。ResponseFuture 是一个接口，下面我们来看一下它的默认实现类 DefaultFuture 的源码。

  1. public class DefaultFuture implements ResponseFuture {
  3.     private static final Map<Long, Channel> CHANNELS = 
  4.         new ConcurrentHashMap<Long, Channel>();
  6.     private static final Map<Long, DefaultFuture> FUTURES = 
  7.         new ConcurrentHashMap<Long, DefaultFuture>();
  9.     private final long id;
  10.     private final Channel channel;
  11.     private final Request request;
  12.     private final int timeout;
  13.     private final Lock lock = new ReentrantLock();
  14.     private final Condition done = lock.newCondition();
  15.     private volatile Response response;
  17.     public DefaultFuture(Channel channel, Request request, int timeout) {
  18.         this.channel = channel;
  19.         this.request = request;
  21.         // 获取请求 id，这个 id 很重要，后面还会见到
  22.         this.id = request.getId();
  23.         this.timeout = timeout > 0 ? timeout : channel.getUrl().getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
  24.         // 存储 <requestId, DefaultFuture> 映射关系到 FUTURES 中
  25.         FUTURES.put(id, this);
  26.         CHANNELS.put(id, channel);
  27.     }
  29.     @Override
  30.     public Object get() throws RemotingException {
  31.         return get(timeout);
  32.     }
  34.     @Override
  35.     public Object get(int timeout) throws RemotingException {
  36.         if (timeout <= 0) {
  37.             timeout = Constants.DEFAULT_TIMEOUT;
  38.         }
  40.         // 检测服务提供方是否成功返回了调用结果
  41.         if (!isDone()) {
  42.             long start = System.currentTimeMillis();
  43.             lock.lock();
  44.             try {
  45.                 // 循环检测服务提供方是否成功返回了调用结果
  46.                 while (!isDone()) {
  47.                     // 如果调用结果尚未返回，这里等待一段时间
  48.                     done.await(timeout, TimeUnit.MILLISECONDS);
  49.                     // 如果调用结果成功返回，或等待超时，此时跳出 while 循环，执行后续的逻辑
  50.                     if (isDone() || System.currentTimeMillis() - start > timeout) {
  51.                         break;
  52.                     }
  53.                 }
  54.             } catch (InterruptedException e) {
  55.                 throw new RuntimeException(e);
  56.             } finally {
  57.                 lock.unlock();
  58.             }
  60.             // 如果调用结果仍未返回，则抛出超时异常
  61.             if (!isDone()) {
  62.                 throw new TimeoutException(sent > 0, channel, getTimeoutMessage(false));
  63.             }
  64.         }
  66.         // 返回调用结果
  67.         return returnFromResponse();
  68.     }
  70.     @Override
  71.     public boolean isDone() {
  72.         // 通过检测 response 字段为空与否，判断是否收到了调用结果
  73.         return response != null;
  74.     }
  76.     private Object returnFromResponse() throws RemotingException {
  77.         Response res = response;
  78.         if (res == null) {
  79.             throw new IllegalStateException("response cannot be null");
  80.         }
  82.         // 如果调用结果的状态为 Response.OK，则表示调用过程正常，服务提供方成功返回了调用结果
  83.         if (res.getStatus() == Response.OK) {
  84.             return res.getResult();
  85.         }
  87.         // 抛出异常
  88.         if (res.getStatus() == Response.CLIENT_TIMEOUT || res.getStatus() == Response.SERVER_TIMEOUT) {
  89.             throw new TimeoutException(res.getStatus() == Response.SERVER_TIMEOUT, channel, res.getErrorMessage());
  90.         }
  91.         throw new RemotingException(channel, res.getErrorMessage());
  92.     }
  94.     // 省略其他方法
  95. }

如上，当服务消费者还未接收到调用结果时，用户线程调用 get 方法会被阻塞住。同步调用模式下，框架获得 DefaultFuture 对象后，会立即调用 get 方法进行等待。而异步模式下则是将该对象封装到 FutureAdapter 实例中，并将 FutureAdapter 实例设置到 RpcContext 中，供用户使用。FutureAdapter 是一个适配器，用于将 Dubbo 中的 ResponseFuture 与 JDK 中的 Future 进行适配。这样当用户线程调用 Future 的 get 方法时，经过 FutureAdapter 适配，最终会调用 ResponseFuture 实现类对象的 get 方法，也就是 DefaultFuture 的 get 方法。

到这里关于 Dubbo 几种调用方式的代码逻辑就分析完了，下面来分析请求数据的发送与接收，以及响应数据的发送与接收过程。