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實現原理