dubbo的一次請求源碼分析

在這裏插入圖片描述
調用某個服務首先會進入到動態代理。 InvokerInvocationHandler#invoke(Object proxy, Method method, Object[] args)

public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        String methodName = method.getName();
        Class<?>[] parameterTypes = method.getParameterTypes();
        if (method.getDeclaringClass() == Object.class) {
            return method.invoke(invoker, args);
        }
        /*object#method mock*/
        if ("toString".equals(methodName) && parameterTypes.length == 0) {
            return invoker.toString();
        }
        /*...*/
        /*省略hashCode和equals代碼*/
        /*其餘實現方法invoke*/
        return invoker.invoke(new RpcInvocation(method, args)).recreate();
    }
複製代碼

MockClusterInvoker#invoke(Invocation invocation) 裝飾者模式,MockClusterInvoker裝飾了Invoker,主要看result = this.invoker.invoke(invocation);java

public Result invoke(Invocation invocation) throws RpcException {
        Result result = null;

        String value = directory.getUrl().getMethodParameter(invocation.getMethodName(), Constants.MOCK_KEY, Boolean.FALSE.toString()).trim();
        if (value.length() == 0 || value.equalsIgnoreCase("false")) {
            //no mock
            result = this.invoker.invoke(invocation);
        } else if (value.startsWith("force")) {
            /*省略代碼,log*/
            //force:direct mock
            result = doMockInvoke(invocation, null);
        } else {
            //fail-mock
            try {
                result = this.invoker.invoke(invocation);
            } catch (RpcException e) {
                if (e.isBiz()) {
                    throw e;
                } else {
                    /*省略代碼,log*/
                    result = doMockInvoke(invocation, e);
                }
            }
        }
        return result;
    }
複製代碼

AbstractClusterInvoker#invoke(final Invocation invocation),缺省實現FailoverClusterInvoker,首先關注list(invocation);緩存

public Result invoke(final Invocation invocation) throws RpcException {
        checkWhetherDestroyed();
        LoadBalance loadbalance = null;

        // binding attachments into invocation.
        Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();
        if (contextAttachments != null && contextAttachments.size() != 0) {
            ((RpcInvocation) invocation).addAttachments(contextAttachments);
        }
		/*先往下看list方法*/
        List<Invoker<T>> invokers = list(invocation);
        /*得到負載均衡的具體實現,doInvoke或用到該實例,缺省RandomLoadBalance*/
        if (invokers != null && !invokers.isEmpty()) {
            loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()
                    .getMethodParameter(RpcUtils.getMethodName(invocation), Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE));
        }
        /*跳過,默認狀況下,將在異步操做中添加調用ID,爲了冪等*/
        RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
        return doInvoke(invocation, invokers, loadbalance);
    }
複製代碼
protected List<Invoker<T>> list(Invocation invocation) throws RpcException {
        List<Invoker<T>> invokers = directory.list(invocation);
        return invokers;
    }
複製代碼

directory.list(invocation);實際調用AbstractDirectory#list(Invocation invocation),缺省實現爲RegistryDirectoryapp

public List<Invoker<T>> list(Invocation invocation) throws RpcException {
        if (destroyed) {
            throw new RpcException("Directory already destroyed .url: " + getUrl());
        }
        /*得到invokers,先往下看doList方法*/
        List<Invoker<T>> invokers = doList(invocation);
        /*得到routers*/
        List<Router> localRouters = this.routers; // local reference
        if (localRouters != null && !localRouters.isEmpty()) {
        	/*遍歷全部Router,得到正常的invokers*/
            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) {
                    /*省略代碼,log*/
                }
            }
        }
        return invokers;
    }
複製代碼

doList(invocation);調用了RegistryDirectory#doList(Invocation invocation)負載均衡

public List<Invoker<T>> doList(Invocation invocation) {
        if (forbidden) {
            // 1. No service provider 2. Service providers are disabled
            /*省略代碼,throw new RpcException*/
        }
        List<Invoker<T>> invokers = null;
        /*緩存了方法和invokers的mapping,Invoker就是具體調用的執行器,之後能夠分析怎麼得到的*/
        Map<String, List<Invoker<T>>> localMethodInvokerMap = this.methodInvokerMap; // local reference
        if (localMethodInvokerMap != null && localMethodInvokerMap.size() > 0) {
            String methodName = RpcUtils.getMethodName(invocation);
            /*得到入參*/
            Object[] args = RpcUtils.getArguments(invocation);
            /*得到具體的invokers*/
            if (args != null && args.length > 0 && args[0] != null
                    && (args[0] instanceof String || args[0].getClass().isEnum())) {
                invokers = localMethodInvokerMap.get(methodName + "." + args[0]); // The routing can be enumerated according to the first parameter
            }
            if (invokers == null) {
                invokers = localMethodInvokerMap.get(methodName);
            }
            if (invokers == null) {
                invokers = localMethodInvokerMap.get(Constants.ANY_VALUE);
            }
            if (invokers == null) {
                Iterator<List<Invoker<T>>> iterator = localMethodInvokerMap.values().iterator();
                if (iterator.hasNext()) {
                    invokers = iterator.next();
                }
            }
        }
        return invokers == null ? new ArrayList<Invoker<T>>(0) : invokers;
    }
複製代碼

得到了invokers,我再看上面的AbstractDirectory#list(Invocation invocation)方法,router.route()實際調用了MockInvokersSelector#route(final List<Invoker> invokers, URL url, final Invocation invocation)dom

public <T> List<Invoker<T>> route(final List<Invoker<T>> invokers,
                                      URL url, final Invocation invocation) throws RpcException {
        if (invocation.getAttachments() == null) {
            return getNormalInvokers(invokers);
        } else {
        	/*是否須要mock*/
            String value = invocation.getAttachments().get(Constants.INVOCATION_NEED_MOCK);
            if (value == null)
            	/*走這*/
                return getNormalInvokers(invokers);
            else if (Boolean.TRUE.toString().equalsIgnoreCase(value)) {
                return getMockedInvokers(invokers);
            }
        }
        return invokers;
    }
複製代碼
private <T> List<Invoker<T>> getNormalInvokers(final List<Invoker<T>> invokers) {
		/*若是沒有mock的Provider,作校驗,校驗經過返回全部invokers*/
        if (!hasMockProviders(invokers)) {
            return invokers;
        } else {
        	/*不然去掉mock的Provider*/
            List<Invoker<T>> sInvokers = new ArrayList<Invoker<T>>(invokers.size());
            for (Invoker<T> invoker : invokers) {
                if (!invoker.getUrl().getProtocol().equals(Constants.MOCK_PROTOCOL)) {
                    sInvokers.add(invoker);
                }
            }
            return sInvokers;
        }
    }
複製代碼

AbstractDirectory#list(Invocation invocation)方法終於結束了,主要就是得到了正常的invokers。 小結:首先Directory得到全部Invokers,而後Router得到全部非mock的Invokers。異步

接着回到AbstractClusterInvoker#invoke(final Invocation invocation),得到具體負載均衡的實例後,調用了FailoverClusterInvoker#doInvoke(Invocation invocation, final List<Invoker> invokers, LoadBalance loadbalance),下面這部分主要爲負載到某個Invoker,不想看的能夠直接跳到invokeide

public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
        List<Invoker<T>> copyinvokers = invokers;
        /*檢查是否爲空*/
        checkInvokers(copyinvokers, invocation);
        int len = getUrl().getMethodParameter(invocation.getMethodName(), Constants.RETRIES_KEY, Constants.DEFAULT_RETRIES) + 1;
        if (len <= 0) {
            len = 1;
        }
        // retry loop.
        RpcException le = null; // last exception.
        List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(copyinvokers.size()); // invoked invokers.
        Set<String> providers = new HashSet<String>(len);
        for (int i = 0; i < len; i++) {
            //在重試以前從新選擇以免Invokers發生變化。
            //注意:若是`invokers`改變了,那麼`invoked`也會失去準確性。
            if (i > 0) {
                checkWhetherDestroyed();
                copyinvokers = list(invocation);
                // check again
                checkInvokers(copyinvokers, invocation);
            }
            /*選擇具體某個Invoker,先往下看*/
            Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
            invoked.add(invoker);
            RpcContext.getContext().setInvokers((List) invoked);
            try {
            	/*invoke!!!*/
                Result result = invoker.invoke(invocation);
                if (le != null && logger.isWarnEnabled()) {
                    /*省略代碼,log.warn*/
                }
                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*/
    }
複製代碼

AbstractClusterInvoker#select(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected)oop

protected Invoker<T> select(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
        if (invokers == null || invokers.isEmpty())
            return null;
        String methodName = invocation == null ? "" : invocation.getMethodName();

        boolean sticky = invokers.get(0).getUrl().getMethodParameter(methodName, Constants.CLUSTER_STICKY_KEY, Constants.DEFAULT_CLUSTER_STICKY);
        {
            //ignore overloaded method
            if (stickyInvoker != null && !invokers.contains(stickyInvoker)) {
                stickyInvoker = null;
            }
            //ignore concurrency problem
            if (sticky && stickyInvoker != null && (selected == null || !selected.contains(stickyInvoker))) {
                if (availablecheck && stickyInvoker.isAvailable()) {
                    return stickyInvoker;
                }
            }
        }
        Invoker<T> invoker = doSelect(loadbalance, invocation, invokers, selected);

        if (sticky) {
            stickyInvoker = invoker;
        }
        return invoker;
    }
複製代碼

AbstractClusterInvoker#doSelect(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected)ui

private Invoker<T> doSelect(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
        if (invokers == null || invokers.isEmpty())
            return null;
        /*可有一個直接返回*/
        if (invokers.size() == 1)
            return invokers.get(0);
        if (loadbalance == null) {
            loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);
        }
        /*以前實例化的LoadBalance*/
        Invoker<T> invoker = loadbalance.select(invokers, getUrl(), invocation);

        //若是`invoker`在`selected`中或者invoker不可用&& availablecheck爲true,則從新選擇。
        if ((selected != null && selected.contains(invoker))
                || (!invoker.isAvailable() && getUrl() != null && availablecheck)) {
            try {
                Invoker<T> rinvoker = reselect(loadbalance, invocation, invokers, selected, availablecheck);
                if (rinvoker != null) {
                    invoker = rinvoker;
                } else {
                    //檢查當前所選調用者的索引,若是不是最後一個,選擇index+1的這個。
                    int index = invokers.indexOf(invoker);
                    try {
                        //Avoid collision
                        invoker = index < invokers.size() - 1 ? invokers.get(index + 1) : invokers.get(0);
                    } catch (Exception e) {
                        /*省略代碼,log.warn*/
                    }
                }
            } catch (Throwable t) {
                /*省略代碼,log.error*/
            }
        }
        return invoker;
    }
複製代碼

AbstractLoadBalance#select(List<Invoker<T>> invokers, URL url, Invocation invocation)this

public <T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) {
        if (invokers == null || invokers.isEmpty())
            return null;
        if (invokers.size() == 1)
            return invokers.get(0);
        return doSelect(invokers, url, invocation);
    }
複製代碼

RandomLoadBalance#doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation),計算隨機值,判斷在哪一個權重範圍內,則返回這個範圍中的Invoker

protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
        int length = invokers.size(); //invokers總數
        int totalWeight = 0; //總權重
        boolean sameWeight = true; // 權重是都否同樣
        for (int i = 0; i < length; i++) {
            int weight = getWeight(invokers.get(i), invocation);
            totalWeight += weight; // 累計總權重
            if (sameWeight && i > 0
                    && weight != getWeight(invokers.get(i - 1), invocation)) {
                sameWeight = false; //計算全部權重是否同樣
            }
        }
        if (totalWeight > 0 && !sameWeight) {
            // 若是(並不是每一個調用者具備相同的權重而且至少一個調用者的權重>0),則根據totalWeight隨機選擇。
            int offset = random.nextInt(totalWeight);
            // 根據隨機值返回一個調用者。
            for (int i = 0; i < length; i++) {
                offset -= getWeight(invokers.get(i), invocation);
                if (offset < 0) {
                    return invokers.get(i);
                }
            }
        }
        //若是全部調用者具備相同的權重值或totalWeight = 0,則均勻返回。
        return invokers.get(random.nextInt(length));
    }
複製代碼

Invoker獲取到了就能夠執行了,回到FailoverClusterInvoker#doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance),接下來invoker.invoke(invocation),該方法通過了幾層裝飾,調用責任鏈(之後再寫怎麼生成噠),最後調用了AbstractInvoker(實現類DubboInvoker)#invoke(Invocation inv)

在這裏插入圖片描述

public Result invoke(Invocation inv) throws RpcException {
        if (destroyed.get()) {
            /*省略代碼,throw new RpcException*/
        }
        RpcInvocation invocation = (RpcInvocation) inv;
        invocation.setInvoker(this);
        if (attachment != null && attachment.size() > 0) {
            invocation.addAttachmentsIfAbsent(attachment);
        }
        Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();
        if (contextAttachments != null && contextAttachments.size() != 0) {
            /** * 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);
        }
        if (getUrl().getMethodParameter(invocation.getMethodName(), Constants.ASYNC_KEY, false)) {
            invocation.setAttachment(Constants.ASYNC_KEY, Boolean.TRUE.toString());
        }
        RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);

        try {
            return doInvoke(invocation);
        } catch (InvocationTargetException e) { // biz exception
            /*省略,異常處理*/
        } catch (RpcException e) {
            /*省略,異常處理*/
        } catch (Throwable e) {
            return new RpcResult(e);
        }
    }
複製代碼

DubboInvoker.doInvoke(final Invocation invocation),發送接收消息

protected Result doInvoke(final Invocation invocation) throws Throwable {
        RpcInvocation inv = (RpcInvocation) invocation;
        final String methodName = RpcUtils.getMethodName(invocation);
        inv.setAttachment(Constants.PATH_KEY, getUrl().getPath());
        inv.setAttachment(Constants.VERSION_KEY, version);

        ExchangeClient currentClient;
        if (clients.length == 1) {
            currentClient = clients[0];
        } else {
            currentClient = clients[index.getAndIncrement() % clients.length];
        }
        try {
            boolean isAsync = RpcUtils.isAsync(getUrl(), invocation);
            boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);
            int timeout = getUrl().getMethodParameter(methodName, Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
            if (isOneway) {
                boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
                currentClient.send(inv, isSent);
                RpcContext.getContext().setFuture(null);
                return new RpcResult();
            } else if (isAsync) {
                ResponseFuture future = currentClient.request(inv, timeout);
                RpcContext.getContext().setFuture(new FutureAdapter<Object>(future));
                return new RpcResult();
            } else {
                RpcContext.getContext().setFuture(null);
                return (Result) currentClient.request(inv, timeout).get();
            }
        } catch (TimeoutException e) {
            /*省略代碼,throw new RpcException*/
        } catch (RemotingException e) {
            /*省略代碼,throw new RpcException*/
        }
    }
複製代碼

總結:

  1. 首先經過Directory得到全部Invokers;
  2. 接着Router得到全部非mock的Invokers;
  3. 而後經過LoadBalance得到某個Invoker;
  4. 最後通過一些裝飾和責任鏈後,調用發送接收消息,獲得最後結果。
相關文章
相關標籤/搜索