dubbo的超時重試

dubbo的超時分為服務端超時 SERVER_TIMEOUT 和客戶端超時 CLIENT_TIMEOUT。本文討論服務端超時的情形：

超時：consumer發送調用請求后，等待服務端的響應，若超過timeout時間仍未收到響應，則拋異常。

dubbo consumer 超時重試的邏輯在 FailoverClusterInvoker.doInvoke 中：

public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, 
                            LoadBalance loadbalance) throws RpcException {
    List<Invoker<T>> copyinvokers = invokers;
    checkInvokers(copyinvokers, invocation);
    //取retries參數值，默認值為2，所以len默認為3
    int len = getUrl().getMethodParameter(invocation.getMethodName(), 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++) {
        //重試時，進行重新選擇，避免重試時invoker列表已發生變化.
        //注意：如果列表發生了變化，那么invoked判斷會失效，因為invoker示例已經改變
        if (i > 0) {
            checkWheatherDestoried();
            copyinvokers = list(invocation);
            //重新檢查一下
            checkInvokers(copyinvokers, invocation);
        }
        Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
        invoked.add(invoker);
        RpcContext.getContext().setInvokers((List)invoked);
        try {
　　　　　　　//繼續invoker鏈的調用
            Result result = invoker.invoke(invocation);
            if (le != null && logger.isWarnEnabled()) {
                //打印日志：上次調用產生的異常
                logger.warn("Although retry the method XXX");
            }
            //調用成功，即返回。如果產生RpcException異常，進入catch塊，設置le。
            return result;
        } catch (RpcException e) {
            //在DubboInvoker.doInvoke中會把TimeoutException封裝成RpcException
            //所以超時異常會進入這個catch分支，開始for循環的下一次調用
            if (e.isBiz()) { // biz exception.
                throw e;
            }
            le = e;
        } catch (Throwable e) {
            le = new RpcException(e.getMessage(), e);
        } finally {
            providers.add(invoker.getUrl().getAddress());
        }
    } // retry loop.
    //調用len次后，仍然沒有結果，則拋異常。
    throw new RpcException(le != null ? le.getCode() : 0, "Failed to invoke the method "
            + invocation.getMethodName() + " 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 != null ? le.getMessage() : ""), le != null && le.getCause() != null ? le.getCause() : le);
}

當invoker的調用鏈進行到DubboInvoker.doInvoke時：

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) {
            //oneway的意思是：consumer不需要調用結果。需要配置return="false"
            boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
            currentClient.send(inv, isSent);
            RpcContext.getContext().setFuture(null);
            return new RpcResult();
        } else if (isAsync) {
            //如果consumer需要調用結果，但又不想阻塞程序，則設置return="true", async="true" 
            ResponseFuture future = currentClient.request(inv, timeout);
            //在RpcContext中設置Future，返回空的RpcResult
            RpcContext.getContext().setFuture(new FutureAdapter<Object>(future));
            return new RpcResult();
        } else {
            //如果consumer想阻塞獲取provider的調用結果，不需要做配置，默認即可。
            RpcContext.getContext().setFuture(null);
            //currentClient.request會發送請求，返回Future。調用Future.get導致阻塞
            return (Result) currentClient.request(inv, timeout).get();
        }
    } catch (TimeoutException e) {
        //調用超時，將TimeoutException封裝成RpcException。
        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);
    }
}

currentClient.request(inv, timeout).get(); 會阻塞等待響應，超時則會拋出異常。

// HeaderExchangeChannel.request
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;
}

// DefaultFuture.get(int timeout)
public Object get(int timeout) throws RemotingException {
    if (timeout <= 0) {
        timeout = Constants.DEFAULT_TIMEOUT;
    }
    if (! isDone()) {
　　　　 //記錄開始時間
        long start = System.currentTimeMillis();
        lock.lock();
        try {
            while (! isDone()) {
 　　　　　　　　 //(1)await超時醒來，但是未收到響應：則isDone為false，但是System.currentTimeMillis() - start > timeout 為true
 //(2)provider及時響應。更具體的說法是等待DubboClientHandler線程接收響應后，喚醒該線程。isDone會設置為true
 //(3)RemotingInvocationTimeoutScan線程掃描到超時，然后創建一個超時響應，並喚醒這個等待。isDone被設置為true
                done.await(timeout, TimeUnit.MILLISECONDS);
                if (isDone() || System.currentTimeMillis() - start > timeout) {
                    break;
                }
            }
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            lock.unlock();
        }
        if (! isDone()) {
            //拋出超時異常
            throw new TimeoutException(sent > 0, channel, getTimeoutMessage(false));
        }
    }
    return returnFromResponse();
}

 在DefaultFuture類中有一個內部類RemotingInvocationTimeoutScan，負責掃描超時的調用，在客戶端構造超時響應。

private static class RemotingInvocationTimeoutScan implements Runnable {

    public void run() {
        while (true) {
            try {
                for (DefaultFuture future : FUTURES.values()) {
                    if (future == null || future.isDone()) {
                        continue;
                    }
                    if (System.currentTimeMillis() - future.getStartTimestamp() > future.getTimeout()) {
                        // create exception response.
                        Response timeoutResponse = new Response(future.getId());
                        // set timeout status.
                        timeoutResponse.setStatus(future.isSent() ? Response.SERVER_TIMEOUT : Response.CLIENT_TIMEOUT);
                        timeoutResponse.setErrorMessage(future.getTimeoutMessage(true));
                        // handle response.
                        DefaultFuture.received(future.getChannel(), timeoutResponse);
                    }
                }
                Thread.sleep(30);
            } catch (Throwable e) {
                logger.error("Exception when scan the timeout invocation of remoting.", e);
            }
        }
    }
}

static {
    Thread th = new Thread(new RemotingInvocationTimeoutScan(), "DubboResponseTimeoutScanTimer");
    th.setDaemon(true);
    th.start();
}