9.1 客户端发起请求源码

来看一下客户端请求代码：

        DemoService demoService = (DemoService) context.getBean("demoService"); // 获取远程服务代理
        String hello = demoService.sayHello("world"); // 执行远程方法

在8.2 构建客户端源码解析中我们看到最终得到的demoService是一个proxy0代理对象。现在来分析第二行代码。

一 客户端请求总体流程

//代理发出请求
proxy0.sayHello(String paramString)
-->InvokerInvocationHandler.invoke(Object proxy, Method method, Object[] args)
  -->new RpcInvocation(method, args)
  -->MockClusterInvoker.invoke(Invocation invocation)//服务降级的地方
    //ClusterInvoker将多个Invoker伪装成一个集群版的Invoker
    -->AbstractClusterInvoker.invoke(final Invocation invocation)
      //获取Invokers
      -->list(Invocation invocation)
        -->AbstractDirectory.list(Invocation invocation)
          -->RegistryDirectory.doList(Invocation invocation)//从Map<String, List<Invoker<T>>> methodInvokerMap中获取key为sayHello的List<Invoker<T>>
          -->MockInvokersSelector.getNormalInvokers(final List<Invoker<T>> invokers)//对上述的List<Invoker<T>>再进行一次过滤（这里比如说过滤出所有协议为mock的Invoker，如果一个也没有就全部返回），这就是router的作用
      //获取负载均衡器
      -->loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()
                    .getMethodParameter(invocation.getMethodName(), Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE))//默认为random
      -->RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation)//异步操作添加invocationID
      -->FailoverClusterInvoker.doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance)
        //使用负载均衡器选择一个Invoker出来：RegistryDirectory$InvokerDelegete实例
        -->AbstractClusterInvoker.select(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected)
          -->doselect(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected)
            -->AbstractLoadBalance.select(List<Invoker<T>> invokers, URL url, Invocation invocation)
              -->RandomLoadBalance.doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation)
        //执行listener和filter链
        -->ListenerInvokerWrapper.invoke
          -->ConsumerContextFilter.invoke(Invoker<?> invoker, Invocation invocation)//设置一些RpcContext属性，并且设置invocation中的invoker属性
            -->FutureFilter.invoke(Invocation invocation)
              -->MonitorFilter.invoke(Invocation invocation)//monitor在这里收集数据
                -->AbstractInvoker.invoke(Invocation inv)//重新设置了invocation中的invoker属性和attachment属性
                  -->DubboInvoker.doInvoke(final Invocation invocation)
                    //获取ExchangeClient进行消息的发送
                    -->ReferenceCountExchangeClient.request(Object request, int timeout)
                      -->HeaderExchangeClient.request(Object request, int timeout)
                        -->HeaderExchangeChannel.request(Object request, int timeout)
                          -->AbstractClient.send(Object message, boolean sent)//NettyClient的父类
                            -->getChannel()//NettyChannel实例，其内部channel实例=NioClientSocketChannel实例
                            -->NettyChannel.send(Object message, boolean sent)
                              -->NioClientSocketChannel.write(Object message)//已经是netty的东西了，这里的message=Request实例：最重要的是RpcInvocation [methodName=sayHello, parameterTypes=[class java.lang.String], arguments=[world], attachments={path=com.alibaba.dubbo.demo.DemoService, interface=com.alibaba.dubbo.demo.DemoService, version=0.0.0}]

总体流程：

• 将请求参数（方法名，方法参数类型，方法参数值，服务名，附加参数）封装成一个Invocation
  • 附加参数中的path：即接口名，将会用于服务端接收请求信息后从exportMap中选取Exporter实例
  • 方法名，方法参数类型，方法参数值：将用于JavassistProxyFactory$AbstractProxyInvoker执行对应的方法
• 使用Directory从Map<String, List<Invoker<T>>> methodInvokerMap中获取key为sayHello（指定方法名）的List<Invoker<T>>
• 使用Router对上述的List<Invoker<T>>再进行一次过滤，得到subList
• 使用LoadBalancer从subList中再获取一个Invoker，实际上是InvokerDelegete实例
• 使用InvokerDelegete实例执行真正的DubboInvoker的listener和filter链，然后执行到真正的DubboInvoker
• DubboInvoker使用NettyClient向服务端发出了请求

二 源码分析

首先来看proxy0.sayHello

    public String sayHello(String paramString) {
        Object[] arrayOfObject = new Object[1];
        arrayOfObject[0] = paramString;
        Object localObject = null;
        try {
            localObject = this.handler.invoke(this, DemoService.class.getMethod("sayHello"), arrayOfObject);
        } catch (Throwable e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        return (String) localObject;
    }

这里的handler就是InvokerInvocationHandler

public class InvokerInvocationHandler implements InvocationHandler {
    private final Invoker<?> invoker;//MockClusterInvoker实例

    public InvokerInvocationHandler(Invoker<?> handler) {
        this.invoker = handler;
    }

    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);
        }
        if ("toString".equals(methodName) && parameterTypes.length == 0) {
            return invoker.toString();
        }
        if ("hashCode".equals(methodName) && parameterTypes.length == 0) {
            return invoker.hashCode();
        }
        if ("equals".equals(methodName) && parameterTypes.length == 1) {
            return invoker.equals(args[0]);
        }
        return invoker.invoke(new RpcInvocation(method, args)).recreate();
    }
}

首先将请求参数封装成一个RpcInvocation实例，如下：

-->String methodName=sayHello
-->Class<?>[] parameterTypes=[class java.lang.String]
-->Object[] arguments=[world]
-->Map<String, String> attachments={}

之后使用MockClusterInvoker.invoke(Invocation invocation)进行远程调用：

    private final Directory<T> directory;//RegistryDirectory
    private final Invoker<T> invoker;//FailoverClusterInvoker

    /**
     * 这里实际上会根据配置的mock参数来做服务降级：
     * 1 如果没有配置mock参数或者mock=false，则进行远程调用；
     * 2 如果配置了mock=force:return null，则直接返回null，不进行远程调用；
     * 3 如果配置了mock=fail:return null，先进行远程调用，失败了在进行mock调用。
     */
    public Result invoke(Invocation invocation) throws RpcException {
        Result result = null;
        //sayHello.mock->mock->default.mock
        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")) {
            if (logger.isWarnEnabled()) {
                logger.info("force-mock: " + invocation.getMethodName() + " force-mock enabled , url : " + directory.getUrl());
            }
            //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 {
                    if (logger.isWarnEnabled()) {
                        logger.info("fail-mock: " + invocation.getMethodName() + " fail-mock enabled , url : " + directory.getUrl(), e);
                    }
                    result = doMockInvoke(invocation, e);
                }
            }
        }
        return result;
    }

注意：这里可以做服务降级，后续会说。

之后调用FailoverClusterInvoker.invoke方法，该方法在其父类AbstractClusterInvoker中，

    protected final Directory<T> directory;//RegistryDirectory    
    
    public Result invoke(final Invocation invocation) throws RpcException {
        ...
        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);//异步调用加调用ID
        return doInvoke(invocation, invokers, loadbalance);
    }

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

首先是获取一个List<Invoker<T>>，之后获取一个LoadBalance，最后调用doInvoke进行调用。

首先来看通过RegistryDirectory.list(Invocation invocation)，该方法在RegistryDirectory的父类AbstractDirectory中：

    private volatile List<Router> routers;
    public List<Invoker<T>> list(Invocation invocation) throws RpcException {
        ...
        List<Invoker<T>> invokers = doList(invocation);
        List<Router> localRouters = this.routers; // local reference
        if (localRouters != null && localRouters.size() > 0) {
            for (Router router : localRouters) {
                try {
                    if (router.getUrl() == null || router.getUrl().getParameter(Constants.RUNTIME_KEY, true)) {
                        invokers = router.route(invokers, getConsumerUrl(), invocation);
                    }
                } catch (Throwable t) {
                    logger.error("Failed to execute router: " + getUrl() + ", cause: " + t.getMessage(), t);
                }
            }
        }
        return invokers;
    }

首先执行doList(invocation)方法获取出List<Invoker<T>>，之后使用router循环过滤，最后返回过滤后的List<Invoker<T>>。

RegistryDirectory.doList(invocation)

    public List<Invoker<T>> doList(Invocation invocation) {
        ...
        List<Invoker<T>> invokers = null;
        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);
            if (args != null && args.length > 0 && args[0] != null
                    && (args[0] instanceof String || args[0].getClass().isEnum())) {
                invokers = localMethodInvokerMap.get(methodName + "." + args[0]); // 可根据第一个参数枚举路由 sayHello.world
            }
            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;
    }

其中Map<String, List<Invoker<T>>> methodInvokerMap在8.2 构建客户端源码解析已经初始化好了：

Map<String, List<Invoker<T>>> methodInvokerMap={
sayHello=[provider1的RegistryDirectory$InvokerDelegete实例, provider2的RegistryDirectory$InvokerDelegete实例], *=[provider1的RegistryDirectory$InvokerDelegete实例, provider2的RegistryDirectory$InvokerDelegete实例]}

这里根据方法名sayHello取出两个RegistryDirectory$InvokerDelegete实例。最后通过Router进行过滤，这里只有一个Router，就是MockInvokersSelector。

    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 {
            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) {
        if (!hasMockProviders(invokers)) {
            return invokers;
        } else {
            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;
        }
    }

这里直接返回了。到此就已经选出可以被调用的RegistryDirectory$InvokerDelegete实例子集了。记下来先获取负载均衡器，默认是RandomLoadBalance。最后执行FailoverClusterInvoker.

doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance)：

    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;//默认是2+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++) {
            //重试时，进行重新选择，避免重试时invoker列表已发生变化.
            //注意：如果列表发生了变化，那么invoked判断会失效，因为invoker示例已经改变
            if (i > 0) {
                checkWhetherDestroyed();
                copyinvokers = list(invocation);
                //重新检查一下
                checkInvokers(copyinvokers, invocation);
            }
            Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
            invoked.add(invoker);
            RpcContext.getContext().setInvokers((List) invoked);
            try {
                Result result = invoker.invoke(invocation);
                ...
                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 ...);
    }

首先使用负载均衡器获取一个RegistryDirectory$InvokerDelegete实例，然后使用选出的RegistryDirectory$InvokerDelegete.invoke进行请求发送。

    protected Invoker<T> select(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
        ...
        Invoker<T> invoker = doselect(loadbalance, invocation, invokers, selected);
        ..
        return invoker;
    }

    private Invoker<T> doselect(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
        if (invokers == null || invokers.size() == 0)
            return null;
        if (invokers.size() == 1)
            return invokers.get(0);
        // 如果只有两个invoker，并且其中一个已经有至少一个被选过了，退化成轮循
        if (invokers.size() == 2 && selected != null && selected.size() > 0) {
            return selected.get(0) == invokers.get(0) ? invokers.get(1) : invokers.get(0);
        }
        Invoker<T> invoker = loadbalance.select(invokers, getUrl(), invocation);

        //如果 selected中包含（优先判断） 或者 不可用&&availablecheck=true 则重试.
        if ((selected != null && selected.contains(invoker))
                || (!invoker.isAvailable() && getUrl() != null && availablecheck)) {
            try {
                Invoker<T> rinvoker = reselect(loadbalance, invocation, invokers, selected, availablecheck);
                ...
            } catch (Throwable t) {
               ...
            }
        }
        return invoker;
    }

RandomLoadBalance.doSelect

    private final Random random = new Random();

    protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
        int length = invokers.size(); // 总个数
        ...//权重计算
        // 如果权重相同或权重为0则均等随机
        return invokers.get(random.nextInt(length));
    }

最后来看RegistryDirectory$InvokerDelegete.invoke，该方法实际在其父类InvokerWrapper中：

    private final Invoker<T> invoker;//ListenerInvokerWrapper

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

ListenerInvokerWrapper.invoke

    private final Invoker<T> invoker;//ProtocolFilterWrapper$Invoker

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

之后就会执行一系列的filter，这些filter后续会讲，现在直接执行到DubboInvoker.invoke，实际上该方法在其父类AbstractInvoker中，AbstractInvoker又调用了DubboInvoker.doInvoke：

    private final ExchangeClient[] clients;

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

其中ExchangeClient[] clients在8.2 构建客户端源码解析已经被初始化好了：

ExchangeClient[] clients = [ReferenceCountExchangeClient实例]//如果设置了多条连接，此处有多个client

ReferenceCountExchangeClient.request

    private ExchangeClient client;//HeaderExchangeClient

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

HeaderExchangeClient.request

    private final ExchangeChannel channel;//HeaderExchangeChannel

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

HeaderExchangeChannel.request

    private final Channel channel;//NettyClient

    public ResponseFuture request(Object request, int timeout) throws RemotingException {
        if (closed) {
            throw new RemotingException(this.getLocalAddress(), null, "Failed to send request " + request + ", cause: The channel " + this + " is closed!");
        }
        // create request.
        Request req = new Request();
        req.setVersion("2.0.0");
        req.setTwoWay(true);
        req.setData(request);
        DefaultFuture future = new DefaultFuture(channel, req, timeout);
        try {
            channel.send(req);
        } catch (RemotingException e) {
            future.cancel();
            throw e;
        }
        return future;
    }

上边的channel是NettyClient实例，这里的send实际上是调用其父类AbstractClient的父类AbstractPeer，AbstractPeer调用AbstractClient.send：

    public void send(Object message, boolean sent) throws RemotingException {
        if (send_reconnect && !isConnected()) {
            connect();
        }
        Channel channel = getChannel();//NettyChannel
        //TODO getChannel返回的状态是否包含null需要改进
        if (channel == null || !channel.isConnected()) {
            throw new RemotingException(this, "message can not send, because channel is closed . url:" + getUrl());
        }
        channel.send(message, sent);
    }

NettyChannel.send

    private final org.jboss.netty.channel.Channel channel;//NioClientSocketChannel

    public void send(Object message, boolean sent) throws RemotingException {
        super.send(message, sent);

        boolean success = true;
        int timeout = 0;
        try {
            ChannelFuture future = channel.write(message);
            if (sent) {
                timeout = getUrl().getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
                success = future.await(timeout);
            }
            Throwable cause = future.getCause();
            if (cause != null) {
                throw cause;
            }
        } catch (Throwable e) {
            throw new RemotingException(this, "Failed to send message " + message + " to " + getRemoteAddress() + ", cause: " + e.getMessage(), e);
        }

        if (!success) {
            throw new RemotingException(this, "Failed to send message " + message + " to " + getRemoteAddress()
                    + "in timeout(" + timeout + "ms) limit");
        }
    }

这里就执行到了netty内部，通过netty自己的NioClientSocketChannel将消息发送给服务端。（这里发送之前有编码行为，后续会讲）