Dubbo源码解析—网络调用

Dubbo网络调用

背景

咱们知道Dubbo远程调用（消费过程）的大体流程以下：

• 从Dirctory中获取该方法的invoker列表
• 通过router路由的筛选，获得知足条件的invoker列表
• 通过Cluster容错调用invoker
• 通过loadBalance筛选出最终执行的invoker
• 通过消费端的filter链
• 网络请求及序列化
• .....提供者方执行请求，返回结果
• 用户线程获取结果

网络调用方式

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

Dubbo里面经过参数isOneway、isAsync来控制调用方式：

• isOneway=true 表示异步不带回调
• isAsync=true 表示异步带回调
• 上述两种状况都不知足，使用同步API

同步调用

概念

同步调用适用在大部分环境，通讯方式简单、可靠，客户端发起调用，等待服务端处理，调用结果同步返回。

这种方式下，在高吞吐、高性能（响应时间很快）的服务接口场景中最为适用，能够减小异步带来的额外的消耗，也方便客户端作一致性保证。

源码

同步状况下，客户端发起请求，并经过get()方法阻塞等待服务端的响应结果:

RpcContext.getContext().setFuture(null);
return (Result) currentClient.request(inv, timeout).get();

// 代码位置: org.apache.dubbo.rpc.protocol.dubbo.DubboInvoker#doInvoke

咱们进入get()方法中：

public Object get(int timeout) throws RemotingException {
  if (timeout <= 0) {
    timeout = Constants.DEFAULT_TIMEOUT;
  }

  // 若是没有收到响应，则进行循环内，循环进行判断
  if (!isDone()) {

    long start = System.currentTimeMillis();
    lock.lock();
    try {
      while (!isDone()) {
        // 在等待队列中阻塞，等待被唤醒
        done.await(timeout, TimeUnit.MILLISECONDS);
        if (isDone() || System.currentTimeMillis() - start > timeout) {
          break;
        }
      }
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    } finally {
      lock.unlock();
    }

    if (!isDone()) {
      throw new TimeoutException(sent > 0, channel, getTimeoutMessage(false));
    }
  }

  // 从response对象中获取结果返回
  return returnFromResponse();
}

// 代码位置： org.apache.dubbo.remoting.exchange.support.DefaultFuture#get(int)

用户线程何时被唤醒呢？

private void doReceived(Response res) {
  lock.lock();
  try {
    response = res;
    if (done != null) {
      // 当收到响应的时候，唤醒正在等待的客户端线程
      done.signal();
    }
  } finally {
    lock.unlock();
  }
  if (callback != null) {
    invokeCallback(callback);
  }
}

// 代码位置：org.apache.dubbo.remoting.exchange.support.DefaultFuture#doReceived

流程

• 用户线程发起网络请求
• 用户线程调用ResponseFuture.get()方法进入阻塞状态
• 提供方执行请求，返回结果
• 唤醒用户线程
• 用户线程从await()方法返回，获得结果

异步调用有返回值

概念

异步调用有返回值，用在任务处理时间较长，客户端应用线程不肯阻塞等待，而是为了提升自身处理能力但愿服务端处理完成后能够异步通知应用线程。这种方式能够大大提高客户端的吞吐量，避免由于服务端的耗时问题拖死客户端。

实战

首先设置referenceConfig为async参数为异步调用：

<dubbo:reference
            id="demoService"
            interface="com.huang.yuan.api.service.DemoService"
            version="1.0"
            async="true"   
            timeout="1000000">
</dubbo:reference>

提供方代码以下:

@Override
public ModelResult<String> test(String param) {

  spLogger.warn("远程方法被执行了，进入睡眠");

  try {
    Thread.sleep(1000);
  } catch (Exception e) {
    e.printStackTrace();
  }

  spLogger.warn("远程方法结束了");

  return new ModelResult<>(param);
}

消费方代码以下：

public void test() throws Exception {
  ModelResult<String> modelResult = demoService.test("huangyuan");
  System.out.println("当即获取到结果= " + modelResult);

  System.out.println("用户线程先作点别的事...");

  Future future = RpcContext.getContext().getFuture();
  System.out.println("用户线程经过future获取到结果= " + future.get());
}

结果以下：

源码

异步请求的状况下，用户线程发起请求后，放置一个Future到RpcContext中，返回当即返回一个空的结果。

ResponseFuture future = currentClient.request(inv, timeout);
RpcContext.getContext().setFuture(new FutureAdapter<Object>(future));
return new RpcResult();
// 代码位置: com.alibaba.dubbo.rpc.protocol.dubbo.DubboInvoker#doInvoke

用户线程这时能够去干点别的事，当用户线程想要获取结果的时候，能够调用Future.get()方法尝试获取结果，代码会进入 com.alibaba.dubbo.remoting.exchange.support.DefaultFuture#get(int)，若是这时候提供方尚未返回结果，则用户线程进入阻塞状态

一样的，接收到提供方的结果之后，回调com.alibaba.dubbo.remoting.exchange.support.DefaultFuture#doReceived，这时候用户线程就能够从阻塞状态中返回，获取到结果。

流程

引用官网的图解，大概流程以下：

• 一、用户线程调用接口
• 二、发出网络请求
• 三、设置future到RpcContext上下文中
• 四、用户线程从RpcContext上下文中获取future
• 五、用户线程调用get()方法等待结果，进入阻塞状态
• 六、远程调用返回响应
• 七、唤醒用户线程，获取结果

---

能够看到，Dubbo目前虽然实现了异步调用，可是获取结果仍是须要同步阻塞等待，这个问题在apache dubbo中经过CompletableFuture获得解决，用户线程能够真正的不用管结果什么时候返回，只要dubbo回调用户线程，用户线程去拿结果便可

异步调用不带返回值

概念

异步调用不带返回值，一些场景为了进一步提高客户端的吞吐能力，只需发起一次服务端调用，不需关心调用结果，可使用此种通讯方式。

通常在不须要严格保证数据一致性或者有其余补偿措施的状况下，选用这种，能够最小化远程调用带来的性能损耗。

实战

这种调用方式目前Dubbo的配置文件彷佛还不支持，能够经过自定义FIlter，改写Dubbo参数的方式使用这种调用方式：

<dubbo:reference
  id="demoService"
  interface="com.huang.yuan.api.service.DemoService"
  version="1.0"
  timeout="1000000"
  filter="testFilter">  // 自定义filter
</dubbo:reference>

自定义Filter以下：

public class Filter2 implements Filter {

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        
        Map<String, String> attachments = invocation.getAttachments();

        // 经过RETURN_KEY这个参数，表示调用不须要返回值
        attachments.put(Constants.RETURN_KEY, "false");

        return invoker.invoke(invocation);
    }
}

输出结果：

源码

异步不带回调接口的调用方式，源码很是简单，就是在发起请求以后，当即返回一个空结果

boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
currentClient.send(inv, isSent);
RpcContext.getContext().setFuture(null);
return new RpcResult();

参考

非阻塞通信下的同步API实现原理