dubbo服务引用二之建立Invoker

一、建立Invoker流程

1.一、收集引用参数

ReferenceConfig#init方法的结尾处，调用createProxy方法，采集的参数集合做为入参传递到该方法中。

1.二、建立Invoker

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

方法主要逻辑就是
一、默认状况下若是本地有服务暴露，则引用本地服务.
二、用户写死了引用的URL，指定的URL多是对点对直连地址，也多是注册中心URL
三、经过注册中心配置拼装URL，List<URL> us = loadRegistries(false); 用户配置了几个注册中心，就会产生几个URL

无论走哪一种引用类型，都会执行下面的核心代码

invoker = refprotocol.refer(interfaceClass, url);

refprotocol是一个Protocol接口，getAdaptiveExtension返回的是一个Protocol$Adpative，在https://segmentfault.com/a/11... 该类的源码及其生产方法。

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

主要任务是，暴露远程服务、引用远程服务、释放协议【释放暴露于引用服务时占用的资源】，
dubbo支持多种协议，http,thrift,RMI等，真是经过dubbo的SPI机制，才能够灵活的在这些协议来回切换。
咱们第二种为例讲一下核心逻辑，

同服务暴露时的同样，由于Dubbo的AOP机制，在得到RegistryProtocol时，会通过两个Wrapper类的包装

这个地方也不例外，可是两个Wrapper类的ProtocolFilterWrapper，ProtocolListenerWrapper并没有实际的业务逻辑，咱们直接跳过。

执行refprotocol.refer(interfaceClass, url)即执行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.一、 链接zookeeper

Registry registry = registryFactory.getRegistry(url);

在服务发布的时候，已经讲过了，其主要核心做用就是链接zookeeper服务器，并返回一个ZookeeperRegistry实例。

在RegistryProtocol#doRefer方法中，经过ZookeeperRegistry#register的执行，建立引用服务的consumers节点。建立以下节点：
/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.二、 监听节点

1.2.三、 加入集群

cluster.join(directory)
cluster也是一个带有Adaptive注解的扩展类，默认实现时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，一个能够失败转移的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源码中，它实现了父类中的一个模板子方法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;
    }

咱们在构造FailoverClusterInvoker时，传入的Directory实现类是RegistryDirectory，即AbstractDirectory#list方法。

1.2.四、 核心类DubboInvoker

DubboInvoker 最终Invoker执行的方法是：

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

调用invoker，白话描述就是：
将经过远程通讯将Invocation信息传递给服务器端，服务器端接收到该Invocation信息后，找到对应的本地Invoker，而后经过反射执行相应的方法，将方法的返回值再经过远程通讯将结果传递给客户端。

这里分3种状况：

• 执行方法不须要返回值：直接调用ExchangeClient.send()方法
• 执行方法的结果须要异步返回：使用ExchangeClient.request()方法返回一个ResponseFuture对象，经过RpcContext中的ThreadLocal使ResponseFuture和当前线程绑定，未等服务端响应结果就直接返回，而后服务端经过ProtocolFilterWrapper.buildInvokerChain()方法会调用Filter.invoke()方法，即FutureFilter.invoker()->asyncCallback()，会获取RpcContext的ResponseFuture对象，异步返回结果
• 执行方法的结果须要同步返回：使用ExchangeClient.request()方法，返回一个ResponseFuture，一直阻塞到服务端返回响应结果

返回FailoverClusterInvoker