dubbo源碼解析（四十六）消費端發送請求過程

時間 2019-11-07

原文原文鏈接

2.7大揭祕——消費端發送請求過程

目標：從源碼的角度分析一個服務方法調用經歷怎麼樣的磨難之後到達服務端。

前言

前一篇文章講到的是引用服務的過程，引用服務無非就是建立出一個代理。供消費者調用服務的相關方法。本節將從調用方法開始講解內部的整個調用鏈。咱們就拿dubbo內部的例子講。java

ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("spring/dubbo-consumer.xml");
context.start();
DemoService demoService = context.getBean("demoService", DemoService.class);
String hello = demoService.sayHello("world");
System.out.println("result: " + hello);

這是dubbo-demo-xml-consumer內的實例代碼。接下來咱們就開始來看調用demoService.sayHello方法的時候，dubbo執行了哪些操做。spring

執行過程

（一）InvokerInvocationHandler的invoke

public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
    // 得到方法名稱
    String methodName = method.getName();
    // 得到方法參數類型
    Class<?>[] parameterTypes = method.getParameterTypes();
    // 若是該方法所在的類是Object類型，則直接調用invoke。
    if (method.getDeclaringClass() == Object.class) {
        return method.invoke(invoker, args);
    }
    // 若是這個方法是toString，則直接調用invoker.toString()
    if ("toString".equals(methodName) && parameterTypes.length == 0) {
        return invoker.toString();
    }
    // 若是這個方法是hashCode直接調用invoker.hashCode()
    if ("hashCode".equals(methodName) && parameterTypes.length == 0) {
        return invoker.hashCode();
    }
    // 若是這個方法是equals，直接調用invoker.equals(args[0])
    if ("equals".equals(methodName) && parameterTypes.length == 1) {
        return invoker.equals(args[0]);
    }

    // 調用invoke
    return invoker.invoke(new RpcInvocation(method, args)).recreate();
}

能夠看到上面的源碼，首先對Object的方法進行了處理，若是調用的方法不是這些方法，則先會建立RpcInvocation，而後再調用invoke。segmentfault

RpcInvocation的構造方法

public RpcInvocation(Method method, Object[] arguments) {
    this(method.getName(), method.getParameterTypes(), arguments, null, null);
}

public RpcInvocation(String methodName, Class<?>[] parameterTypes, Object[] arguments, Map<String, String> attachments, Invoker<?> invoker) {
    // 設置方法名
    this.methodName = methodName;
    // 設置參數類型
    this.parameterTypes = parameterTypes == null ? new Class<?>[0] : parameterTypes;
    // 設置參數
    this.arguments = arguments == null ? new Object[0] : arguments;
    // 設置附加值
    this.attachments = attachments == null ? new HashMap<String, String>() : attachments;
    // 設置invoker實體
    this.invoker = invoker;
}

建立完RpcInvocation後，就是調用invoke。先進入的是ListenerInvokerWrapper的invoke。數組

（二）MockClusterInvoker的invoke

能夠參考《dubbo源碼解析（四十一）集羣——Mock》的（二）MockClusterInvoker，降級後的返回策略的實現，根據配置的不一樣來決定不用降級仍是強制服務降級仍是失敗後再服務降級。app

（三）AbstractClusterInvoker的invoke

能夠參考《dubbo源碼解析（三十五）集羣——cluster》的（一）AbstractClusterInvoker，該類是一個抽象類，其中封裝了一些公用的方法，AbstractClusterInvoker的invoke也只是作了一些公用操做。主要的邏輯在doInvoke中。異步

（四）FailoverClusterInvoker的doInvoke

能夠參考《dubbo源碼解析（三十五）集羣——cluster》的（十二）FailoverClusterInvoker，該類實現了失敗重試的容錯策略。async

（五）InvokerWrapper的invoke

能夠參考《dubbo源碼解析（二十二）遠程調用——Protocol》的（五）InvokerWrapper。該類用了裝飾模式，不過並無實現實際的功能加強。ide

（六）ProtocolFilterWrapper的內部類CallbackRegistrationInvoker的invoke

public Result invoke(Invocation invocation) throws RpcException {
    // 調用攔截器鏈的invoke
    Result asyncResult = filterInvoker.invoke(invocation);

    // 把異步返回的結果加入到上下文中
    asyncResult.thenApplyWithContext(r -> {
        // 循環各個過濾器
        for (int i = filters.size() - 1; i >= 0; i--) {
            Filter filter = filters.get(i);
            // onResponse callback
            // 若是該過濾器是ListenableFilter類型的
            if (filter instanceof ListenableFilter) {
                // 強制類型轉化
                Filter.Listener listener = ((ListenableFilter) filter).listener();
                if (listener != null) {
                    // 若是內部類listener不爲空，則調用回調方法onResponse
                    listener.onResponse(r, filterInvoker, invocation);
                }
            } else {
                // 不然，直接調用filter的onResponse，作兼容。
                filter.onResponse(r, filterInvoker, invocation);
            }
        }
        // 返回異步結果
        return r;
    });

    // 返回異步結果
    return asyncResult;
}

這裏看到先是調用攔截器鏈的invoke方法。下面的邏輯是把異步返回的結果放到上下文中，具體的ListenableFilter以及內部類的設計，還有thenApplyWithContext等方法我會在異步的實現中講到。ui

（七）ProtocolFilterWrapper的buildInvokerChain方法中的invoker實例的invoke方法。

public Result invoke(Invocation invocation) throws RpcException {
    Result asyncResult;
    try {
        // 依次調用各個過濾器，得到最終的返回結果
        asyncResult = filter.invoke(next, invocation);
    } catch (Exception e) {
        // onError callback
        // 捕獲異常，若是該過濾器是ListenableFilter類型的
        if (filter instanceof ListenableFilter) {
            // 得到內部類Listener
            Filter.Listener listener = ((ListenableFilter) filter).listener();
            if (listener != null) {
                //調用onError，回調錯誤信息
                listener.onError(e, invoker, invocation);
            }
        }
        // 拋出異常
        throw e;
    }
    // 返回結果
    return asyncResult;
}

該方法中是對異常的捕獲，調用內部類Listener的onError來回調錯誤信息。接下來看它通過了哪些攔截器。this

（八）ConsumerContextFilter的invoke

public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
    // 得到上下文，設置invoker，會話域，本地地址和原創地址
    RpcContext.getContext()
            .setInvoker(invoker)
            .setInvocation(invocation)
            .setLocalAddress(NetUtils.getLocalHost(), 0)
            .setRemoteAddress(invoker.getUrl().getHost(),
                    invoker.getUrl().getPort());
    // 若是會話域是RpcInvocation，則設置invoker
    if (invocation instanceof RpcInvocation) {
        ((RpcInvocation) invocation).setInvoker(invoker);
    }
    try {
        // 移除服務端的上下文
        RpcContext.removeServerContext();
        // 調用下一個過濾器
        return invoker.invoke(invocation);
    } finally {
        // 清空上下文
        RpcContext.removeContext();
    }
}

static class ConsumerContextListener implements Listener {
    @Override
    public void onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation) {
        // 把結果中的附加值放入到上下文中
        RpcContext.getServerContext().setAttachments(appResponse.getAttachments());
    }

    @Override
    public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
        // 不作任何處理
    }
}

能夠參考《dubbo源碼解析（二十）遠程調用——Filter》，不過上面的源碼是最新的，而連接內的源碼是2.6.x的，雖然作了一些變化，好比內部類的的設計，後續的過濾器也有一樣的實現，可是ConsumerContextFilter做用沒有變化，它依舊是在當前的RpcContext中記錄本地調用的一次狀態信息。該過濾器執行完成後，會回到ProtocolFilterWrapper的invoke中的

Result result = filter.invoke(next, invocation);

而後繼續調用下一個過濾器FutureFilter。

（九）FutureFilter的invoke

public Result invoke(final Invoker<?> invoker, final Invocation invocation) throws RpcException {
    // 該方法是真正的調用方法的執行
    fireInvokeCallback(invoker, invocation);
    // need to configure if there's return value before the invocation in order to help invoker to judge if it's
    // necessary to return future.
    return invoker.invoke(invocation);
}

class FutureListener implements Listener {
    @Override
    public void onResponse(Result result, Invoker<?> invoker, Invocation invocation) {
        if (result.hasException()) {
            // 處理異常結果
            fireThrowCallback(invoker, invocation, result.getException());
        } else {
            // 處理正常結果
            fireReturnCallback(invoker, invocation, result.getValue());
        }
    }

    @Override
    public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {

    }
}

能夠參考《dubbo源碼解析（二十四）遠程調用——dubbo協議》中的（十四）FutureFilter，其中會有部分結構不同，跟ConsumerContextFilter同樣，由於後續版本對Filter接口進行了新的設計，增長了onResponse方法，把返回的執行邏輯放到onResponse中去了。其餘邏輯沒有很大變化。等該過濾器執行完成後，仍是回到ProtocolFilterWrapper的invoke中的，繼續調用下一個過濾器MonitorFilter。

（十）MonitorFilter的invoke

public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
    // 若是開啓監控
    if (invoker.getUrl().hasParameter(MONITOR_KEY)) {
        // 設置監控開始時間
        invocation.setAttachment(MONITOR_FILTER_START_TIME, String.valueOf(System.currentTimeMillis()));
        // 得到當前的調用數，而且增長
        getConcurrent(invoker, invocation).incrementAndGet(); // count up
    }
    return invoker.invoke(invocation); // proceed invocation chain
}

class MonitorListener implements Listener {

    @Override
    public void onResponse(Result result, Invoker<?> invoker, Invocation invocation) {
        // 若是開啓監控
        if (invoker.getUrl().hasParameter(MONITOR_KEY)) {
            // 執行監控，蒐集數據
            collect(invoker, invocation, result, RpcContext.getContext().getRemoteHost(), Long.valueOf(invocation.getAttachment(MONITOR_FILTER_START_TIME)), false);
            // 減小當前調用數
            getConcurrent(invoker, invocation).decrementAndGet(); // count down
        }
    }

    @Override
    public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
        // 若是開啓監控
        if (invoker.getUrl().hasParameter(MONITOR_KEY)) {
            // 執行監控，蒐集數據
            collect(invoker, invocation, null, RpcContext.getContext().getRemoteHost(), Long.valueOf(invocation.getAttachment(MONITOR_FILTER_START_TIME)), true);
            // 減小當前調用數
            getConcurrent(invoker, invocation).decrementAndGet(); // count down
        }
    }

能夠看到該過濾器實際用來作監控，監控服務的調用數量等。其中監控的邏輯不是本文重點，因此不細講。接下來調用的是ListenerInvokerWrapper的invoke。

（十一）ListenerInvokerWrapper的invoke

public Result invoke(Invocation invocation) throws RpcException {
    return invoker.invoke(invocation);
}

能夠參考《dubbo源碼解析（二十一）遠程調用——Listener》，這裏用到了裝飾者模式，直接調用了invoker。該類裏面作了服務啓動的監聽器。咱們直接關注下一個invoke。

（十二）AsyncToSyncInvoker的invoke

public Result invoke(Invocation invocation) throws RpcException {
    Result asyncResult = invoker.invoke(invocation);

    try {
        // 若是是同步的調用
        if (InvokeMode.SYNC == ((RpcInvocation)invocation).getInvokeMode()) {
            // 從異步結果中get結果
            asyncResult.get();
        }
    } catch (InterruptedException e) {
        throw new RpcException("Interrupted unexpectedly while waiting for remoting result to return!  method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
    } catch (ExecutionException e) {
        Throwable t = e.getCause();
        if (t instanceof TimeoutException) {
            throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
        } else if (t instanceof RemotingException) {
            throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
        }
    } catch (Throwable e) {
        throw new RpcException(e.getMessage(), e);
    }
    // 返回異步結果
    return asyncResult;
}

AsyncToSyncInvoker類從名字上就很好理解，它的做用是把異步結果轉化爲同步結果。新的改動中每一個調用只要不是oneway方式調用都會先以異步調用開始，而後根據配置的狀況若是是同步調用，則會在這個類中進行異步結果轉同步的處理。固然，這裏先是執行了invoke，而後就進入下一個AbstractInvoker的invoke了。

（十三）AbstractInvoker的invoke

public Result invoke(Invocation inv) throws RpcException {
    // if invoker is destroyed due to address refresh from registry, let's allow the current invoke to proceed
    // 若是服務引用銷燬，則打印告警日誌，可是經過
    if (destroyed.get()) {
        logger.warn("Invoker for service " + this + " on consumer " + NetUtils.getLocalHost() + " is destroyed, "
                + ", dubbo version is " + Version.getVersion() + ", this invoker should not be used any longer");
    }
    RpcInvocation invocation = (RpcInvocation) inv;
    // 會話域中加入該調用鏈
    invocation.setInvoker(this);
    // 把附加值放入會話域
    if (CollectionUtils.isNotEmptyMap(attachment)) {
        invocation.addAttachmentsIfAbsent(attachment);
    }
    // 把上下文的附加值放入會話域
    Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();
    if (CollectionUtils.isNotEmptyMap(contextAttachments)) {
        /**
         * invocation.addAttachmentsIfAbsent(context){@link RpcInvocation#addAttachmentsIfAbsent(Map)}should not be used here,
         * because the {@link RpcContext#setAttachment(String, String)} is passed in the Filter when the call is triggered
         * by the built-in retry mechanism of the Dubbo. The attachment to update RpcContext will no longer work, which is
         * a mistake in most cases (for example, through Filter to RpcContext output traceId and spanId and other information).
         */
        invocation.addAttachments(contextAttachments);
    }

    // 從配置中獲得是什麼模式的調用，一共有FUTURE、ASYNC和SYNC
    invocation.setInvokeMode(RpcUtils.getInvokeMode(url, invocation));
    // 加入編號
    RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);

    try {
        // 執行調用鏈
        return doInvoke(invocation);
    } catch (InvocationTargetException e) { // biz exception
        // 得到異常
        Throwable te = e.getTargetException();
        if (te == null) {
            // 建立默認的異常異步結果
            return AsyncRpcResult.newDefaultAsyncResult(null, e, invocation);
        } else {
            if (te instanceof RpcException) {
                // 設置異常碼
                ((RpcException) te).setCode(RpcException.BIZ_EXCEPTION);
            }
            // 建立默認的異常異步結果
            return AsyncRpcResult.newDefaultAsyncResult(null, te, invocation);
        }
    } catch (RpcException e) {
        if (e.isBiz()) {
            return AsyncRpcResult.newDefaultAsyncResult(null, e, invocation);
        } else {
            throw e;
        }
    } catch (Throwable e) {
        return AsyncRpcResult.newDefaultAsyncResult(null, e, invocation);
    }
}

能夠參考《dubbo源碼解析（二十二）遠程調用——Protocol》的（三）AbstractInvoker。該方法作了一些公共的操做，好比服務引用銷燬的檢測，加入附加值，加入調用鏈實體域到會話域中等。而後執行了doInvoke抽象方法。各協議本身去實現。而後就是執行到doInvoke方法了。使用的協議不同，doInvoke的邏輯也有所不一樣，我這裏舉的例子是使用dubbo協議，因此我就介紹DubboInvoker的doInvoke，其餘自行查看具體的實現。這次的異步改造加入了InvokeMode，我會在後續中介紹這個。

（十四）DubboInvoker的doInvoke

protected Result doInvoke(final Invocation invocation) throws Throwable {
    // rpc會話域
    RpcInvocation inv = (RpcInvocation) invocation;
    // 得到方法名
    final String methodName = RpcUtils.getMethodName(invocation);
    // 把path放入到附加值中
    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 = getUrl().getMethodParameter(methodName, TIMEOUT_KEY, DEFAULT_TIMEOUT);
        // 若是是單向發送
        if (isOneway) {
            // 是否等待消息發送，默認不等待消息發出，將消息放入 IO 隊列，即刻返回。
            boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
            // 單向發送只負責發送消息，不等待服務端應答，因此沒有返回值
            currentClient.send(inv, isSent);
            // 設置future爲null，由於單向發送沒有返回值
            RpcContext.getContext().setFuture(null);
            // 建立一個默認的AsyncRpcResult
            return AsyncRpcResult.newDefaultAsyncResult(invocation);
        } else {
            // 不然直接建立AsyncRpcResult
            AsyncRpcResult asyncRpcResult = new AsyncRpcResult(inv);
            // 異步調用，返回CompletableFuture類型的future
            CompletableFuture<Object> responseFuture = currentClient.request(inv, timeout);
            // 當調用結果完成時
            responseFuture.whenComplete((obj, t) -> {
                // 若是有異常
                if (t != null) {
                    // 拋出一個異常
                    asyncRpcResult.completeExceptionally(t);
                } else {
                    // 完成調用
                    asyncRpcResult.complete((AppResponse) obj);
                }
            });
            // 異步返回結果用CompletableFuture包裝，把future放到上下文中，
            RpcContext.getContext().setFuture(new FutureAdapter(asyncRpcResult));
            // 返回結果
            return asyncRpcResult;
        }
    } 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);
    }
}

能夠參考《dubbo源碼解析（二十四）遠程調用——dubbo協議》的（一）DubboInvoker，不過連接內的文章的源碼是2.6.x版本的，而上述的源碼是最新版本的，其中就有對於異步的改動，好比加入了異步返回結果、除了單向調用，一概都先處理成AsyncRpcResult等。具體的AsyncRpcResult以及其中用到的CompletableFuture我會在下文介紹。

上述源碼中執行currentClient.request或者currentClient.send，表明把請求放入channel中，交給channel來處理請求。最後來看一個currentClient.request，由於這其中涉及到了Future的構建。

（十五）ReferenceCountExchangeClient的request

public CompletableFuture<Object> request(Object request, int timeout) throws RemotingException {
    return client.request(request, timeout);
}

ReferenceCountExchangeClient是一個記錄請求數的類，用了適配器模式，對ExchangeClient作了功能加強。

能夠參考《dubbo源碼解析（二十四）遠程調用——dubbo協議》的（八）ReferenceCountExchangeClient。

（十六）HeaderExchangeClient的request

public CompletableFuture<Object> request(Object request, int timeout) throws RemotingException {
    return channel.request(request, timeout);
}

該類也是用了適配器模式，該類主要的做用就是增長了心跳功能，能夠參考《dubbo源碼解析（十）遠程通訊——Exchange層》的（四）HeaderExchangeClient。而後進入HeaderExchangeChannel的request。

（十七）HeaderExchangeChannel的request

能夠參考《dubbo源碼解析（十）遠程通訊——Exchange層》的（二）HeaderExchangeChannel，在這個request方法中就能夠看到

// 建立DefaultFuture對象，能夠從future中主動得到請求對應的響應信息
    DefaultFuture future = new DefaultFuture(channel, req, timeout);

生成了須要的future。異步請求結果就是從這個future中獲取。關於DefaultFuture也能夠參考《dubbo源碼解析（十）遠程通訊——Exchange層》的（七）DefaultFuture。

後面channel.send方法就是跟遠程通訊有關了，例如使用netty做爲通訊實現，則會使用netty實現的客戶端進行通訊。

（十八）AbstractPeer的send

能夠參考《dubbo源碼解析（九）遠程通訊——Transport層》的（一）AbstractPeer，其中send方法比較簡單，根據sent配置項去作消息發送。接下來看AbstractClient的send

（十九）AbstractClient的send

能夠參考《dubbo源碼解析（九）遠程通訊——Transport層》的（四）AbstractClient。

public void send(Object message, boolean sent) throws RemotingException {
    // 若是須要重連或者沒有連接，則鏈接
    if (needReconnect && !isConnected()) {
        connect();
    }
    // 得到通道
    Channel channel = getChannel();
    //TODO Can the value returned by getChannel() be null? need improvement.
    if (channel == null || !channel.isConnected()) {
        throw new RemotingException(this, "message can not send, because channel is closed . url:" + getUrl());
    }
    // 經過通道發送消息
    channel.send(message, sent);
}

該方法中作了重連的邏輯，而後就是經過通道發送消息，dubbo有幾種通訊的實現，我這裏就按照默認的netty4實現來說解，因此下一步走到了NettyChannel的send。

（二十）NettyChannel的send

能夠參考《dubbo源碼解析（十七）遠程通訊——Netty4》的（一）NettyChannel。這裏其中先執行了下面父類AbstractChannel的send，檢查了一下通道是否關閉，而後再走下面的邏輯。當執行writeAndFlush方法後，消息就被髮送。

dubbo數據包能夠查看《dubbo源碼解析（十）遠程通訊——Exchange層》的（二十五）ExchangeCodec，後續關於netty發送消息，以及netty出站數據在發出以前還須要進行編碼操做我就先不作介紹，主要是跟netty知識點強相關，只是dubbo作了一些本身的編碼，以及集成了各種序列化方式。