dubbo服務引用之創(chuàng)建Invoker流程詳解

更新時間：2023年08月16日 11:09:33 作者：包月星

這篇文章主要為大家介紹了dubbo服務引用二之創(chuàng)建Invoker流程詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪

1、創(chuàng)建Invoker流程

1.1、收集引用參數(shù)

ReferenceConfig#init方法的結(jié)尾處，調(diào)用createProxy方法，采集的參數(shù)集合作為入?yún)鬟f到該方法中。

1.2、創(chuàng)建Invoker

@SuppressWarnings({"unchecked", "rawtypes", "deprecation"})
    private T createProxy(Map<String, String> map) {
        ...
        // 創(chuàng)建服務代理
        return (T) proxyFactory.getProxy(invoker);
    }

方法主要邏輯就是

1、默認情況下如果本地有服務暴露，則引用本地服務.
2、用戶寫死了引用的URL，指定的URL可能是對點對直連地址，也可能是注冊中心URL
3、通過注冊中心配置拼裝URL，List<URL> us = loadRegistries(false); 用戶配置了幾個注冊中心，就會產(chǎn)生幾個URL

不管走哪種引用類型，都會執(zhí)行下面的核心代碼

invoker = refprotocol.refer(interfaceClass, url);

refprotocol是一個Protocol接口，getAdaptiveExtension返回的是一個Protocol$Adpative，在該類的源碼及其生產(chǎn)方法。

private static final Protocol refprotocol = ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension();

又看到了Protocol這個接口

@SPI("dubbo")
public interface Protocol {
    int getDefaultPort();
    @Adaptive
    <T> Exporter<T> export(Invoker<T> var1) throws RpcException;
    @Adaptive
    <T> Invoker<T> refer(Class<T> var1, URL var2) throws RpcException;
    void destroy();
}

主要任務是，暴露遠程服務、引用遠程服務、釋放協(xié)議【釋放暴露于引用服務時占用的資源】，dubbo支持多種協(xié)議，http,thrift,RMI等，真是通過dubbo的SPI機制，才可以靈活的在這些協(xié)議來回切換。

我們第二種為例講一下核心邏輯，同服務暴露時的一樣，因為Dubbo的AOP機制，在獲RegistryProtocol時，會經(jīng)過兩個Wrapper類的包裝

這個地方也不例外，但是兩個Wrapper類的ProtocolFilterWrapper，ProtocolListenerWrapper并無實際的業(yè)務邏輯，我們直接跳過。

執(zhí)行refprotocol.refer(interfaceClass, url)即執(zhí)行RegistryProtocol#refer代碼。

@SuppressWarnings("unchecked")
    public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
        url = url.setProtocol(url.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_REGISTRY)).removeParameter(Constants.REGISTRY_KEY);
        Registry registry = registryFactory.getRegistry(url);
        if (RegistryService.class.equals(type)) {
            return proxyFactory.getInvoker((T) registry, type, url);
        }
        // group="a,b" or group="*"
        Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(Constants.REFER_KEY));
        String group = qs.get(Constants.GROUP_KEY);
        if (group != null && group.length() > 0) {
            if ((Constants.COMMA_SPLIT_PATTERN.split(group)).length > 1
                    || "*".equals(group)) {
                return doRefer(getMergeableCluster(), registry, type, url);
            }
        }
        return doRefer(cluster, registry, type, url);
    }

1.2.1、連接zookeeper

Registry registry = registryFactory.getRegistry(url);

在服務發(fā)布的時候，已經(jīng)講過了，其主要核心作用就是連接zookeeper服務器，并返回一個ZookeeperRegistry實例。

在RegistryProtocol#doRefer方法中，通過ZookeeperRegistry#register的執(zhí)行，創(chuàng)建引用服務的consumers節(jié)點。創(chuàng)建如下節(jié)點：

/dubbo/com.alibaba.dubbo.demo.DemoService/consumers/consumer%3A%2F%2F192.168.43.156%2Fcom.alibaba.dubbo.demo.DemoService%3Fapplication%3Ddemo-consumer%26category%3Dconsumers%26check%3Dfalse%26dubbo%3D2.0.0%26interface%3Dcom.alibaba.dubbo.demo.DemoService%26methods%3DsayHello%26pid%3D15775%26side%3Dconsumer%26timestamp%3D1525073802234

1.2.2、監(jiān)聽節(jié)點

1.2.3、加入集群

cluster.join(directory)

cluster也是一個帶有Adaptive注解的擴展類，默認實現(xiàn)時FailoverCluster

@SPI(FailoverCluster.NAME)
public interface Cluster {
    /**
     * Merge the directory invokers to a virtual invoker.
     *
     * @param <T>
     * @param directory
     * @return cluster invoker
     * @throws RpcException
     */
    @Adaptive
    <T> Invoker<T> join(Directory<T> directory) throws RpcException;
}

進入FailoverCluster#join,返回FailoverClusterInvoker，一個可以失敗轉(zhuǎn)移的Invoker，

public class FailoverCluster implements Cluster {
    public final static String NAME = "failover";
    public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
        return new FailoverClusterInvoker<T>(directory);
    }
}

FailoverClusterInvoker源碼中，它實現(xiàn)了父類中的一個模板子方法doInvoke。
父類AbstractClusterInvoker的invoke方法，

public Result invoke(final Invocation invocation) throws RpcException {
        checkWhetherDestroyed();
        LoadBalance loadbalance;
        List<Invoker<T>> invokers = list(invocation);
        if (invokers != null && invokers.size() > 0) {
            loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()
                    .getMethodParameter(invocation.getMethodName(), Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE));
        } else {
            loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);
        }
        RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
        return doInvoke(invocation, invokers, loadbalance);
    }

方法list(invocation)返回了List<Invoker<T>> invokers，這些Invoker就是實際與服務交互的對象，

protected List<Invoker<T>> list(Invocation invocation) throws RpcException {
        List<Invoker<T>> invokers = directory.list(invocation);
        return invokers;
    }

我們在構(gòu)造FailoverClusterInvoker時，傳入的Directory實現(xiàn)類是RegistryDirectory，即AbstractDirectory#list方法。

1.2.4、核心類DubboInvoker

DubboInvoker 最終Invoker執(zhí)行的方法是：

@Override
    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(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);
        }
    }

調(diào)用invoker，白話描述就是：
將通過遠程通信將Invocation信息傳遞給服務器端，服務器端接收到該Invocation信息后，找到對應的本地Invoker，然后通過反射執(zhí)行相應的方法，將方法的返回值再通過遠程通信將結(jié)果傳遞給客戶端。

這里分3種情況：

執(zhí)行方法不需要返回值：直接調(diào)用ExchangeClient.send()方法
執(zhí)行方法的結(jié)果需要異步返回：使用ExchangeClient.request()方法返回一個ResponseFuture對象，通過RpcContext中的ThreadLocal使ResponseFuture和當前線程綁定，未等服務端響應結(jié)果就直接返回，然后服務端通過ProtocolFilterWrapper.buildInvokerChain()方法會調(diào)用Filter.invoke()方法，即FutureFilter.invoker()->asyncCallback()，會獲取RpcContext的ResponseFuture對象，異步返回結(jié)果
執(zhí)行方法的結(jié)果需要同步返回：使用ExchangeClient.request()方法，返回一個ResponseFuture，一直阻塞到服務端返回響應結(jié)果

返回FailoverClusterInvoker