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