dubbo服务调用过程

介绍

服务调用过程
推荐

调用的时序图

暴露过程

  首先provider启动，通过Proxy组件根据具体的协议，将需要暴露的接口封装成invoker，invoker是dubbo一个很核心的组件，代表一个可执行体。
  然后再通过Exporter包装一下，这是为了在注册中心暴露自己套的一层。然后将Exporter通过Registry注册到注册中心。 这就是整体服务暴露过程。
  暴露会生成Exporter和服务提供invoker，会将Exporter保存在ExporterMap中，在调用请求到来时，会在这个map中找到对应的Exporter然后逐步还原出原来的invoker。

引入过程

  首先消费者启动会向注册中心拉取服务提供者的元信息，然后调用流程也是从 Proxy 开始，毕竟都需要代理才能无感知。
  Proxy持有一个invoker对象，调用invoke之后需要通过Cluster先从Directory获取所有可调用的远程服务的Invoker列表，如果配置了某些路由规则，比如某个接口只能调用某个节点的那就再过滤一遍Invoker列表。
  剩下的Invoker再通过LoadBalance做负载均衡选取一个。然后再经过Filter做一些统计什么的，再通过Client做数据传输，比如用Netty来传输。
  传输需要通过Codec做协议构造，再序化，最终发往对应的服务提供者。
  服务提供者接收到之后也会进行Codec协议处理，然后反序列化后将请求扔到线程池处理。某个线程会根据请求找到对应的Exporter，找到Exporter就是找到了invoker，经过一层层过滤链之后最终实现调用，最后原路返回。

调用流程

远程调用细节

  服务暴露时，provider启动，Proxy组件会将ref对象（需要暴露的接口）封装成invoker对象（服务提供invoker，AbstractProxyInvoker），就是实际提供服务的对象的代理。
  通过Protocol将invoker对象包装成为Exporter对象保存在ExporterMap中，再在注册中心创建节点，订阅通知，将自己注册到注册中心。
  在调用前，服务引入时，会在注册中心注册消费者节点，订阅通知，获取所有的服务提供者消息（服务提供方export对象信息，内含invoker对象信息）。
  会使用到消费方的Proxy对象（这个对象在消费者初始化完成的时候生成，会注入到容器中），它持有一个消费invoker对象（dubboinvoker等），通过Cluster从Directory获取所有可调用的远程服务Invoker列表。
  在消费方获取到提供方invoker列表信息后，会根据负载均衡等策略确定需要调用的具体提供invoker对象。
  我们在项目发起调用后，消费代理对象，即Proxy组件会执行持有invoker对象（这里是消费invoker）的invoke方法，doinvoke方法。
  doinvoke方法底层，会经过协议构造，序列化后，经过netty客户端，发往服务提供方。
  提供方收到netty请求信息后，也会进行协议构造，反序列化后，将调用请求放入dubbo线程池（貌似默认是200）。
  线程池的请求开始执行，会根据请求的信息，在ExporterMap中找到对应的exporter对象，也就是拿到了对应的服务提供invoker对象，就是实际服务提供的代理，然后执行我们提供的方法，完成调用。

---

调用过程

  服务暴露过程，与服务引入过程在单独的笔记，这里只有实际调用过程。

消费端发起请求

  在消费者初始化的时候，会生成一个消费者代理注册到容器中，消费调用接口时会把服务接口的method对象和参数放到RpcInvocation对象中，传入MockClusterInvoker.invoke，这个主要是看是否配置mock，一般情况下不走mock，会调用FailOverClusterInvoker.invoke。在服务接口消费者初始化时，接口方法和提供者Invoker对应关系保存在RegistryDirectory的methodInvokerMap中，通过调用的方法名称（或方法名称+第一个参数）获得对应的提供者invoker列表，如注册中心设置了路由规则，对这些invoker根据路由规则进行过滤。

  这里先从Directory中拿到了invoker列表。

// AbstractDirectory
@Override
public List<Invoker<T>> list(Invocation invocation) throws RpcException {
    if (destroyed) {
        throw new RpcException("Directory already destroyed .url: " + getUrl());
    }

    return doList(invocation);
}

// RegistryDirectory
@Override
public List<Invoker<T>> doList(Invocation invocation) {
    if (forbidden) {
        // 1. No service provider 2. Service providers are disabled
        throw new RpcException(RpcException.FORBIDDEN_EXCEPTION, "No provider available from registry " +
                getUrl().getAddress() + " for service " + getConsumerUrl().getServiceKey() + " on consumer " +
                NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() +
                ", please check status of providers(disabled, not registered or in blacklist).");
    }

    if (multiGroup) {
        return this.invokers == null ? Collections.emptyList() : this.invokers;
    }

    List<Invoker<T>> invokers = null;
    try {
        // Get invokers from cache, only runtime routers will be executed.
        invokers = routerChain.route(getConsumerUrl(), invocation);
    } catch (Throwable t) {
        logger.error("Failed to execute router: " + getUrl() + ", cause: " + t.getMessage(), t);
    }


    // FIXME Is there any need of failing back to Constants.ANY_VALUE or the first available method invokers when invokers is null?
    /*Map<String, List<Invoker<T>>> localMethodInvokerMap = this.methodInvokerMap; // local reference
    if (localMethodInvokerMap != null && localMethodInvokerMap.size() > 0) {
        String methodName = RpcUtils.getMethodName(invocation);
        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 ? Collections.emptyList() : invokers;
}

  在读取到所有的invoke列表后，会根据负载均衡算法选择一个进行调用。
  这里也是用了SPI机制，支持多种负载均衡算法：随机，RR循环，最不活跃，一致性hash，默认随机。

random=org.apache.dubbo.rpc.cluster.loadbalance.RandomLoadBalance
roundrobin=org.apache.dubbo.rpc.cluster.loadbalance.RoundRobinLoadBalance
leastactive=org.apache.dubbo.rpc.cluster.loadbalance.LeastActiveLoadBalance
consistenthash=org.apache.dubbo.rpc.cluster.loadbalance.ConsistentHashLoadBalance

  确定调用对象后，就会调用对应的doinvoke方法。

@Override
@SuppressWarnings({"unchecked", "rawtypes"})
public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
    List<Invoker<T>> copyInvokers = invokers;
    checkInvokers(copyInvokers, invocation);
    String methodName = RpcUtils.getMethodName(invocation);
    int len = getUrl().getMethodParameter(methodName, 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++) {
        //Reselect before retry to avoid a change of candidate `invokers`.
        //NOTE: if `invokers` changed, then `invoked` also lose accuracy.
        if (i > 0) {
            checkWhetherDestroyed();

            // 获取所有的服务提供invoker列表
            copyInvokers = list(invocation);

            // check again
            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);
            if (le != null && logger.isWarnEnabled()) {
                logger.warn("Although retry the method " + methodName
                        + " in the service " + getInterface().getName()
                        + " was successful by the provider " + invoker.getUrl().getAddress()
                        + ", but there have been failed providers " + providers
                        + " (" + providers.size() + "/" + copyInvokers.size()
                        + ") from the registry " + directory.getUrl().getAddress()
                        + " on the consumer " + NetUtils.getLocalHost()
                        + " using the dubbo version " + Version.getVersion() + ". Last error is: "
                        + le.getMessage(), le);
            }
            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.getCode(), "Failed to invoke the method "
            + methodName + " in the service " + getInterface().getName()
            + ". Tried " + len + " times of the providers " + providers
            + " (" + providers.size() + "/" + copyInvokers.size()
            + ") from the registry " + directory.getUrl().getAddress()
            + " on the consumer " + NetUtils.getLocalHost() + " using the dubbo version "
            + Version.getVersion() + ". Last error is: "
            + le.getMessage(), le.getCause() != null ? le.getCause() : le);
}

  methodInvokerMap保存的是持有DubboInvoker（dubbo协议）实例的InvokerDelegete对象，是Invoker-Filter链的头部，先激活Filter连然后最终调到DubboInvoker.invoke(RpcInvocation)。
  从FailoverClusterInvoker的doinvoke方法会进入AbstractInvoker的invoke方法，最终根据协议头进入具体的invoker中，执行doinvoke方法。

// FailoverClusterInvoker
@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 isAsyncFuture = RpcUtils.isReturnTypeFuture(inv);
        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);
            // For compatibility
            FutureAdapter<Object> futureAdapter = new FutureAdapter<>(future);
            RpcContext.getContext().setFuture(futureAdapter);

            Result result;
            if (isAsyncFuture) {
                // register resultCallback, sometimes we need the async result being processed by the filter chain.
                result = new AsyncRpcResult(futureAdapter, futureAdapter.getResultFuture(), false);
            } else {
                result = new SimpleAsyncRpcResult(futureAdapter, futureAdapter.getResultFuture(), false);
            }
            return result;
        } 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);
    }
}

// AbstractInvoker
@Override
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);
    }
    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
        Throwable te = e.getTargetException();
        if (te == null) {
            return new RpcResult(e);
        } else {
            if (te instanceof RpcException) {
                ((RpcException) te).setCode(RpcException.BIZ_EXCEPTION);
            }
            return new RpcResult(te);
        }
    } catch (RpcException e) {
        if (e.isBiz()) {
            return new RpcResult(e);
        } else {
            throw e;
        }
    } catch (Throwable e) {
        return new RpcResult(e);
    }
}

// DubboInvoker
@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 isAsyncFuture = RpcUtils.isReturnTypeFuture(inv);
        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);
            // For compatibility
            FutureAdapter<Object> futureAdapter = new FutureAdapter<>(future);
            RpcContext.getContext().setFuture(futureAdapter);

            Result result;
            if (isAsyncFuture) {
                // register resultCallback, sometimes we need the async result being processed by the filter chain.
                result = new AsyncRpcResult(futureAdapter, futureAdapter.getResultFuture(), false);
            } else {
                result = new SimpleAsyncRpcResult(futureAdapter, futureAdapter.getResultFuture(), false);
            }
            return result;
        } 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);
    }
}

  在这里面会调用到HeaderExchangeClient的request方法，这里底层就是和netty打交道的地方了。从request方法中就会继续调用HeaderExchangeChannel的request方法，

// HeaderExchangeClient
@Override
public ResponseFuture request(Object request) throws RemotingException {
    return channel.request(request);
}

// HeaderExchangeChannel
@Override
public ResponseFuture request(Object request) throws RemotingException {
    return request(request, channel.getUrl().getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT));
}

@Override
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(Version.getProtocolVersion());
    req.setTwoWay(true);
    req.setData(request);
    DefaultFuture future = DefaultFuture.newFuture(channel, req, timeout);
    try {
        channel.send(req);
    } catch (RemotingException e) {
        future.cancel();
        throw e;
    }
    return future;
}

  这里最终会经过AbstractChannel调用NettyChannel发送请求。

// NettyChanne
@Override
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");
    }
}