dubbo源碼解析——服務調用過程

2020 年 1 月 16 日
筆記

本文中，將進入消費端源碼解析（具體邏輯會放到程式碼的注釋中）。本文先是對消費過程的總體程式碼邏輯理一遍，個別需要細講的點，後面會專門的文章進行解析。

首先，把完整的流程給出來:

服務消費方發送請求

應用啟動的時候，消費者會訂閱服務，並且拉取所有訂閱的提供者節點資訊到Directory中
正式調用開始，在Directory中找出本次集群中的全部invokers
在Router中，將上一步的全部invokers進行篩選，得到滿足條件的invokers
利用cluster集群容錯來調用invoker
在LoadBalance中，根據負載均衡策略，挑選出需要執行的invoker
執行消費者filter鏈
通過Exchange封裝請求-響應對象
對請求對象進行編碼（序列化），通過網路框架（如Netty）發送數據

服務提供方接收請求

對請求進行解碼（反序列化）
將這個請求對象封裝成runable對象，由派發器決定這個請求由IO執行緒或業務執行緒池執行
- 請求解碼可在 IO 執行緒上執行，也可在執行緒池中執行，這個取決於運行時配置。
獲取invoker實例，執行invoke方法
執行提供者filter鏈
invoker反射調用真正的實現類方法
將結果進行編碼
服務提供方返回對應的結果

服務消費方接收調用結果

對響應數據進行解碼
向用戶執行緒傳遞調用結果
- 響應數據解碼完成後，Dubbo 會將響應對象派發到執行緒池上（默認）。
- 然後將響應對象從執行緒池執行緒傳遞到用戶執行緒上。
- 當響應對象到來後，用戶執行緒會被喚醒，並通過調用編號獲取屬於自己的響應對象。

源碼解析

開頭進入InvokerInvocationHandler

通過實現InvocationHandler，我們知道dubbo生成代理使用的是JDK動態代理。這個類中主要是對特殊方法進行處理。由於在生成代理實例的時候，在構造函數中賦值了invoker，因此可以只用該invoker進行invoke方法的調用。

/**   * dubbo使用JDK動態代理，對介面對象進行注入   * InvokerHandler   *   * 程式啟動的過程中，在構造函數中，賦值下一個需要調用的invoker，從而形成執行鏈   */  public class InvokerInvocationHandler implements InvocationHandler {        private final Invoker<?> invoker;        public InvokerInvocationHandler(Invoker<?> handler) {          this.invoker = handler;      }        @Override      public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {          // 獲取方法名稱          String methodName = method.getName();          // 獲取參數類型          Class<?>[] parameterTypes = method.getParameterTypes();          // 方法所處的類 是 Object類，則直接調用          if (method.getDeclaringClass() == Object.class) {              return method.invoke(invoker, args);          }            /*           * toString、hashCode、equals方法比較特殊，如果interface裡面定義了這幾個方法，並且進行實現，           * 通過dubbo遠程調用是不會執行這些程式碼實現的。           */            /*           * 方法調用是toString，依次執行MockClusterInvoker、AbstractClusterInvoker的toString方法           */          if ("toString".equals(methodName) && parameterTypes.length == 0) {              return invoker.toString();          }            /*           * interface中含有hashCode方法，直接調用invoker的hashCode           */          if ("hashCode".equals(methodName) && parameterTypes.length == 0) {              return invoker.hashCode();          }            /*           * interface中含有equals方法，直接調用invoker的equals           */          if ("equals".equals(methodName) && parameterTypes.length == 1) {              return invoker.equals(args[0]);          }            /*           * invocationv包含了遠程調用的參數、方法資訊           */          RpcInvocation invocation;            /*           * todo這段程式碼在最新的dubbo版本中沒有           */          if (RpcUtils.hasGeneratedFuture(method)) {              Class<?> clazz = method.getDeclaringClass();              String syncMethodName = methodName.substring(0, methodName.length() - Constants.ASYNC_SUFFIX.length());              Method syncMethod = clazz.getMethod(syncMethodName, method.getParameterTypes());              invocation = new RpcInvocation(syncMethod, args);              invocation.setAttachment(Constants.FUTURE_GENERATED_KEY, "true");              invocation.setAttachment(Constants.ASYNC_KEY, "true");          } else {              invocation = new RpcInvocation(method, args);              if (RpcUtils.hasFutureReturnType(method)) {                  invocation.setAttachment(Constants.FUTURE_RETURNTYPE_KEY, "true");                  invocation.setAttachment(Constants.ASYNC_KEY, "true");              }          }            // 繼續invoker鏈式調用          return invoker.invoke(invocation).recreate();      }      }

進入MockClusterInvoker

這段程式碼主要是判斷是否需要進行mock調用

@Override      public Result invoke(Invocation invocation) throws RpcException {          Result result = null;            // 獲取mock參數，從而判斷是否需要mock          String value = directory.getUrl().getMethodParameter(invocation.getMethodName(), Constants.MOCK_KEY, Boolean.FALSE.toString()).trim();            if (value.length() == 0 || value.equalsIgnoreCase("false")) {              // 不需要mock，繼續往下調用              result = this.invoker.invoke(invocation);          } else if (value.startsWith("force")) {              if (logger.isWarnEnabled()) {                  logger.warn("force-mock: " + invocation.getMethodName() + " force-mock enabled , url : " + directory.getUrl());              }              // 選擇mock的invoker              result = doMockInvoke(invocation, null);          } else {              // 正常調用失敗，則調用mock              try {                  result = this.invoker.invoke(invocation);              } catch (RpcException e) {                  if (e.isBiz()) {                      throw e;                  } else {                      if (logger.isWarnEnabled()) {                          logger.warn("fail-mock: " + invocation.getMethodName() + " fail-mock enabled , url : " + directory.getUrl(), e);                      }                      result = doMockInvoke(invocation, e);                  }              }          }          return result;      }

進入AbstractClusterInvoker

進入這段程式碼，表明開始進入到集群

@Override      public Result invoke(final Invocation invocation) throws RpcException {          // 檢查消費端invoker是否銷毀了          checkWhetherDestroyed();            // 將參數綁定到invocation          Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();          if (contextAttachments != null && contextAttachments.size() != 0) {              ((RpcInvocation) invocation).addAttachments(contextAttachments);          }            // 獲取滿足條件的invoker（從Directory獲取，並且經過router過濾）          List<Invoker<T>> invokers = list(invocation);            // 初始化loadBalance          LoadBalance loadbalance = initLoadBalance(invokers, invocation);            // invocation ID將被添加在非同步操作          RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);            return doInvoke(invocation, invokers, loadbalance);      }

進入AbstractDirectory

@Override      public List<Invoker<T>> list(Invocation invocation) throws RpcException {          // 判斷Directory是否銷毀          if (destroyed) {              throw new RpcException("Directory already destroyed .url: " + getUrl());          }          // 從methodInvokerMap中取出滿足條件的invoker          List<Invoker<T>> invokers = doList(invocation);            // 根據路由列表，篩選出滿足條件的invoker          List<Router> localRouters = this.routers; // local reference          if (localRouters != null && !localRouters.isEmpty()) {              for (Router router : localRouters) {                  try {                      if (router.getUrl() == null || router.getUrl().getParameter(Constants.RUNTIME_KEY, false)) {                          invokers = router.route(invokers, getConsumerUrl(), invocation);                      }                  } catch (Throwable t) {                      logger.error("Failed to execute router: " + getUrl() + ", cause: " + t.getMessage(), t);                  }              }          }          return invokers;      }

這裡主要是從Directory中獲取invoker，並且經過router路由的篩選，獲得滿足條件的invoker。在AbstractDirectory中，有一個關鍵的方法com.alibaba.dubbo.rpc.cluster.directory.AbstractDirectory#doList，這是一個抽象方法，子類RegistryDirectory的有具體實現，並且調用RegistryDirectory的doList方法。（這裡應該是用到了模板方法模式）。後面的文字中會詳細講下doList方法中做了啥。

進入FailoverClusterInvoker

經過從Directory中獲取invoker，然後router篩選出滿足條件的invoker之後，進入到FailoverClusterInvoker。為什麼會到這裡呢？

根據官網的描述：在集群調用失敗時，Dubbo 提供了多種容錯方案，預設為 failover 重試。所以這個時候是到了FailoverClusterInvoker類,但是如果你配置的是Failfast Cluster(快速失敗),Failsafe Cluster(失敗安全),Failback Cluster(失敗自動恢復),Forking Cluster(並行調用多個伺服器，只要一個成功即返回),Broadcast Cluster(廣播調用所有提供者，逐個調用，任意一台報錯則報錯)，他也會到達相應的類。

@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 = getUrl().getMethodParameter(methodName, Constants.RETRIES_KEY, Constants.DEFAULT_RETRIES) + 1;          if (len <= 0) {              // 最少要調用1次              len = 1;          }            // 局部引用          RpcException le = null;          List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(copyinvokers.size()); // invoked invokers.          Set<String> providers = new HashSet<String>(len);            // i < len 作為循環條件，說明len是多少就循環多少次          for (int i = 0; i < len; i++) {              // 在重試之前，需要重新選擇，以避免候選invoker的改變              if (i > 0) {                  // 檢查invoker是否被銷毀                  checkWhetherDestroyed();                  // 重新選擇invoker                  copyinvokers = list(invocation);                  // 參數檢查                  checkInvokers(copyinvokers, invocation);              }                /*               * 這一步就是進入loadBalance負載均衡               * 因為上述步驟可能篩選出invoker數量大於1，所以再次經過loadBalance的篩選               */              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 " + methodName                              + " 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 "                  + 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 != null ? le.getMessage() : ""), le != null && le.getCause() != null ? le.getCause() : le);      }

到達終點站.我們回憶總結一下,文初提到的三個關鍵詞,在這個集群容錯的整體架構過程中,dubbo究竟做了什麼.其實也就是三件事

（1）在Directory中找出本次集群中的全部invokers （2）在Router中,將上一步的全部invokers挑選出能正常執行的invokers （3）在LoadBalance中,將上一步的能正常的執行invokers中,根據配置的負載均衡策略,挑選出需要執行的invoker

然後就是編碼、網路傳輸、提供方處理數據的過程。後面會補充。

Tags: