public Result invoke(Invocation invocation) throws RpcException {
    Result result = null;
    //獲取mock參數(shù)配置
    String value = this.directory.getUrl().getMethodParameter(invocation.getMethodName(), "mock", Boolean.FALSE.toString()).trim();
    //如果配置了并且配置值為true
    if (value.length() != 0 && !value.equalsIgnoreCase("false")) {
        //強(qiáng)制走mock流程
        if (value.startsWith("force")) {
            result = this.doMockInvoke(invocation, (RpcException)null);
        } else {
            //不走mock流程
            try {
                result = this.invoker.invoke(invocation);
            } catch (RpcException var5) {
                ....
            }
            ....
                result = this.doMockInvoke(invocation, var5);
        }
    }
} else {
    result = this.invoker.invoke(invocation);
}
	return result;
}

public Result invoke(Invocation invocation) throws RpcException {
    //檢查是否被銷(xiāo)毀
    this.checkWhetherDestroyed();
    LoadBalance loadbalance = null;
    //從上下文中獲取attachments，如果獲取得到的話綁定到invocation中
    Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();
    if (contextAttachments != null && contextAttachments.size() != 0) {
        ((RpcInvocation)invocation).addAttachments(contextAttachments);
    }
	//調(diào)用的是directory.list，其中會(huì)做路由過(guò)濾
    List<Invoker<T>> invokers = this.list(invocation);
    //如果過(guò)濾完之后還有Invoker，就通過(guò)SPI獲取對(duì)應(yīng)的LoadBalance實(shí)現(xiàn)類(lèi)
    if (invokers != null && !invokers.isEmpty()) {
        loadbalance = (LoadBalance)ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(((Invoker)invokers.get(0)).getUrl().getMethodParameter(RpcUtils.getMethodName(invocation), "loadbalance", "random"));
    }
    RpcUtils.attachInvocationIdIfAsync(this.getUrl(), invocation);
    return this.doInvoke(invocation, invokers, loadbalance);	//調(diào)用子類(lèi)方法
}
protected List<Invoker<T>> list(Invocation invocation) throws RpcException {
    //獲取invokers目錄，實(shí)際調(diào)用的是AbstractDirectory.list()
    List<Invoker<T>> invokers = this.directory.list(invocation);
    return invokers;
}

public List<Invoker<T>> list(Invocation invocation) throws RpcException {
    if (this.destroyed) {
        throw new RpcException("Directory already destroyed .url: " + this.getUrl());
    } else {
        //抽象方法doList,同樣由子類(lèi)實(shí)現(xiàn)
        List<Invoker<T>> invokers = this.doList(invocation);
        List<Router> localRouters = this.routers;
        if (localRouters != null && !localRouters.isEmpty()) {
            Iterator i$ = localRouters.iterator();
            while(i$.hasNext()) {
                Router router = (Router)i$.next();
                try {	//遍歷router，并判斷是否進(jìn)行路由過(guò)濾
                    if (router.getUrl() == null || router.getUrl().getParameter("runtime", false)) {
                        invokers = router.route(invokers, this.getConsumerUrl(), invocation);
                    }
                } catch (Throwable var7) {
                    logger.error("Failed to execute router: " + this.getUrl() + ", cause: " + var7.getMessage(), var7);
                }
            }
        }
        return invokers;
    }
}

public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
    //重試次數(shù)
    int len = this.getUrl().getMethodParameter(invocation.getMethodName(), "retries", 2) + 1;
    if (len <= 0) {
        len = 1;
    }
    ....
	//重試
    for(int i = 0; i < len; ++i) {
		//負(fù)載均衡篩選出一個(gè)Invoker
        Invoker<T> invoker = this.select(loadbalance, invocation, copyinvokers, invoked);
        invoked.add(invoker);
        //在上下文中保存調(diào)用過(guò)的invoker
        RpcContext.getContext().setInvokers(invoked);
        try {
            Result result = invoker.invoke(invocation);
            ....
            return result;
        } catch (RpcException e) {
		   ....
        } finally {
            providers.add(invoker.getUrl().getAddress());
        }
    }
    throw new RpcException();
}

protected Result doInvoke(Invocation invocation) throws Throwable {
    RpcInvocation inv = (RpcInvocation)invocation;
    String methodName = RpcUtils.getMethodName(invocation);
    inv.setAttachment("path", this.getUrl().getPath());	//設(shè)置path到attachment中
    inv.setAttachment("version", this.version);	//設(shè)置版本號(hào)
    ExchangeClient currentClient;
    if (this.clients.length == 1) {	//選擇client
        currentClient = this.clients[0];
    } else {
        currentClient = this.clients[this.index.getAndIncrement() % this.clients.length];
    }
    try {
        boolean isAsync = RpcUtils.isAsync(this.getUrl(), invocation);	//是否異步調(diào)用
        boolean isOneway = RpcUtils.isOneway(this.getUrl(), invocation); //是否oneway調(diào)用
        int timeout = this.getUrl().getMethodParameter(methodName, "timeout", 1000); //獲取超時(shí)限制
        if (isOneway) {	//oneway 
            boolean isSent = this.getUrl().getMethodParameter(methodName, "sent", false);
            currentClient.send(inv, isSent);	//發(fā)送
            RpcContext.getContext().setFuture((Future)null); //返回空的future
            return new RpcResult(); //返回空結(jié)果
        } else if (isAsync) { //異步調(diào)用
            ResponseFuture future = currentClient.request(inv, timeout);	
            RpcContext.getContext().setFuture(new FutureAdapter(future)); //上下文中設(shè)置future
            return new RpcResult(); //返回空結(jié)果
        } else { //同步調(diào)用
            RpcContext.getContext().setFuture((Future)null); 
            return (Result)currentClient.request(inv, timeout).get(); //直接調(diào)用future.get() 進(jìn)行等待，完成get操作之后再返回結(jié)果
        }
    } catch (TimeoutException var9) {
        throw new RpcException();
    }
}

public ResponseFuture request(Object request, int timeout) throws RemotingException {
    if (this.closed) {
        throw new RemotingException(this.getLocalAddress(), (InetSocketAddress)null, "Failed to send request " + request + ", cause: The channel " + this + " is closed!");
    } else {
        Request req = new Request();	//構(gòu)建request
        req.setVersion(Version.getProtocolVersion());
        req.setTwoWay(true);
        req.setData(request);
        DefaultFuture future = new DefaultFuture(this.channel, req, timeout);
        try {
            this.channel.send(req);		//調(diào)用NettyServer.sent()進(jìn)行發(fā)送請(qǐng)求
            return future;
        } catch (RemotingException var6) {
            future.cancel();
            throw var6;
        }
    }
}

public class Request {
    private final long mId;
    public Request() {
        this.mId = newId();
    }
    //靜態(tài)變量遞增，依次構(gòu)造唯一ID
    private static long newId() {
        return INVOKE_ID.getAndIncrement();
    }
}
public class DefaultFuture implements ResponseFuture {
 	private static final Map<Long, DefaultFuture> FUTURES = new ConcurrentHashMap();
        public DefaultFuture(Channel channel, Request request, int timeout) {
        this.done = this.lock.newCondition();
        this.start = System.currentTimeMillis();
        this.channel = channel;
        this.request = request;
        this.id = request.getId();		//唯一ID
        this.timeout = timeout > 0 ? timeout : channel.getUrl().getPositiveParameter("timeout", 1000);
        FUTURES.put(this.id, this);		//將唯一ID和future的關(guān)系保存到這個(gè)ConcurrentHashMap中
        CHANNELS.put(this.id, channel);
    }
}

public static void received(Channel channel, Response response) {
    try {
        //獲取響應(yīng)的ID去FUTURES中獲取對(duì)應(yīng)的future，獲取之后將future移除
        DefaultFuture future = (DefaultFuture)FUTURES.remove(response.getId());
        if (future != null) {
            //確認(rèn)接收響應(yīng)
            future.doReceived(response);
        } else {
            logger.warn("....");
        }
    } finally {
        CHANNELS.remove(response.getId());
    }
}
private void doReceived(Response res) {
    this.lock.lock();
    try {
        this.response = res;	//響應(yīng)賦值
        if (this.done != null) {
            this.done.signal();	//通知響應(yīng)返回
        }
    } finally {
        this.lock.unlock();
    }
    if (this.callback != null) {
        this.invokeCallback(this.callback);
    }
}

public void received(Channel channel, Object message) throws RemotingException {
    ExecutorService cexecutor = this.getExecutorService();
    try {
        cexecutor.execute(new ChannelEventRunnable(channel, this.handler, ChannelState.RECEIVED, message));
    } catch (Throwable var8) {
        if (message instanceof Request && var8 instanceof RejectedExecutionException) {
            Request request = (Request)message;
            if (request.isTwoWay()) {		//如果需要返回響應(yīng)，將錯(cuò)誤封裝起來(lái)之后返回
                String msg = "Server side(" + this.url.getIp() + "," + this.url.getPort() + ") threadpool is exhausted ,detail msg:" + var8.getMessage();
                Response response = new Response(request.getId(), request.getVersion());
                response.setStatus((byte)100);
                response.setErrorMessage(msg);
                channel.send(response);
                return;
            }
        }
        throw new ExecutionException(message, channel, this.getClass() + " error when process received event .", var8);
    }
}

Response handleRequest(ExchangeChannel channel, Request req) throws RemotingException {
    Response res = new Response(req.getId(), req.getVersion());		//通過(guò)requestId構(gòu)造響應(yīng)
    Object data;
    if (req.isBroken()) {
        data = req.getData();
        String msg;
        if (data == null) {
            msg = null;
        } else if (data instanceof Throwable) {
            msg = StringUtils.toString((Throwable)data);
        } else {
            msg = data.toString();
        }
        res.setErrorMessage("Fail to decode request due to: " + msg);
        res.setStatus((byte)40);
        return res;
    } else {
        data = req.getData();

        try {
            Object result = this.handler.reply(channel, data);	//最終調(diào)用DubboProtocol.reply()
            res.setStatus((byte)20);
            res.setResult(result);
        } catch (Throwable var6) {
            res.setStatus((byte)70);
            res.setErrorMessage(StringUtils.toString(var6));
        }
        return res;
    }
}

public Object reply(ExchangeChannel channel, Object message) throws RemotingException {
    if (!(message instanceof Invocation)) {
        throw new RemotingException();
    } else {
        Invocation inv = (Invocation)message;
        Invoker<?> invoker = DubboProtocol.this.getInvoker(channel, inv);	 //根據(jù)inv得到對(duì)應(yīng)的Invoker
        if (Boolean.TRUE.toString().equals(inv.getAttachments().get("_isCallBackServiceInvoke"))) {
				//一些回調(diào)邏輯
            } else {
                hasMethod = inv.getMethodName().equals(methodsStr);
            }
        }
        RpcContext.getContext().setRemoteAddress(channel.getRemoteAddress());	
        return invoker.invoke(inv);		//調(diào)用選擇的Invoker.invoke()
    }
}

Invoker<?> getInvoker(Channel channel, Invocation inv) throws RemotingException {
    //....
    int port = channel.getLocalAddress().getPort();
    String path = (String)inv.getAttachments().get("path");	
    //根據(jù)port、path以及其他信息獲取serviceKey
    String serviceKey = serviceKey(port, path, (String)inv.getAttachments().get("version"), (String)inv.getAttachments().get("group"));
    //根據(jù)serviceKey在之前提到的exportMap中獲取exporter
    DubboExporter<?> exporter = (DubboExporter)this.exporterMap.get(serviceKey);
    if (exporter == null) {
        throw new RemotingException(....);
    } else {
        return exporter.getInvoker();	//返回Invoker
    }
}