dubbo超時重試
1.此截圖來自官網http://dubbo.io，超時的配置關係以下，方法超時參數爲findXxx.timeout，接口超時參數爲timeout，全局配置超時參數爲default.timeout，當提供端url有變更時這幾個參數都會設置到url的參數中。html
2.超時等參數的設置。緩存
當消費端引用某個接口服務時會訂閱提供端服務的相關變更，開源代碼的註冊中心實現是Zookeeper，當有變更時動觸發RegistryDirectory監聽器，進而調用public synchronized void notify(List<URL> urls)方法，refreshInvoker(invokerUrls);根據提供端的url來更新消費端的url調用關係，這裏有可能傳過來的是提供者下線的url，也就是協議爲empty，進入第一個分支，設置禁止訪問的標誌，消費端這時再出現調用該方法時就會報錯。錯誤信息後續會列出。服務器
/**
* 根據invokerURL列表轉換爲invoker列表。轉換規則以下：
* 1.若是url已經被轉換爲invoker，則不在從新引用，直接從緩存中獲取，注意若是url中任何一個參數變動也會從新引用
* 2.若是傳入的invoker列表不爲空，則表示最新的invoker列表
* 3.若是傳入的invokerUrl列表是空，則表示只是下發的override規則或route規則，須要從新交叉對比，決定是否須要從新引用。
* @param invokerUrls 傳入的參數不能爲null
*/
private void refreshInvoker(List<URL> invokerUrls){
if (invokerUrls != null && invokerUrls.size() == 1 && invokerUrls.get(0) != null
&& Constants.EMPTY_PROTOCOL.equals(invokerUrls.get(0).getProtocol())) {
this.forbidden = true; // 禁止訪問
this.methodInvokerMap = null; // 置空列表
destroyAllInvokers(); // 關閉全部Invoker
} else {
this.forbidden = false; // 容許訪問
Map<String, Invoker<T>> oldUrlInvokerMap = this.urlInvokerMap; // local reference
if (invokerUrls.size() == 0 && this.cachedInvokerUrls != null){
invokerUrls.addAll(this.cachedInvokerUrls);
} else {
this.cachedInvokerUrls = new HashSet<URL>();
this.cachedInvokerUrls.addAll(invokerUrls);//緩存invokerUrls列表，便於交叉對比
}
if (invokerUrls.size() ==0 ){
return;
}
Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls) ;// 將URL列表轉成Invoker列表
Map<String, List<Invoker<T>>> newMethodInvokerMap = toMethodInvokers(newUrlInvokerMap); // 換方法名映射Invoker列表
// state change
//若是計算錯誤，則不進行處理.
if (newUrlInvokerMap == null || newUrlInvokerMap.size() == 0 ){
logger.error(new IllegalStateException("urls to invokers error .invokerUrls.size :"+invokerUrls.size() + ", invoker.size :0. urls :"+invokerUrls.toString()));
return ;
}
this.methodInvokerMap = multiGroup ? toMergeMethodInvokerMap(newMethodInvokerMap) : newMethodInvokerMap;
this.urlInvokerMap = newUrlInvokerMap;
try{
destroyUnusedInvokers(oldUrlInvokerMap,newUrlInvokerMap); // 關閉未使用的Invoker
}catch (Exception e) {
logger.warn("destroyUnusedInvokers error. ", e);
}
}
}提供者正常更新url時，消費端會先
URL列表轉成Invoker列表，Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls) ;組裝新的消費者url，URL url = mergeUrl(providerUrl);在這個方法中
合併消費端參數providerUrl = ClusterUtils.mergeUrl(providerUrl, queryMap); 先獲取提供端參數remoteUrl.getParameters()，而後放入新的HashMap中map.putAll(remoteMap);，再把消費端參數放入map中覆蓋已有的參數 map.putAll(localMap);，最後設置providerUrl = providerUrl.addParameter(Constants.CHECK_KEY, String.valueOf(false)); // 不檢查鏈接是否成功，老是建立Invoker！超時等參數設置完成，而後就是引用服務的後續流程。
3.超時等參數的使用。
當消費端調用引用方法時的路徑爲，先取出全部可用的代理。AbstractClusterInvoker.invoke(final Invocation invocation)->AbstractDirectory.list(invocation)->RegistryDirectory.doList(Invocation invocation)，這個方法中有一個判斷，如今是否禁止訪問，這個參數的設置在前面已經說過。ide
if (forbidden) {
throw new RpcException(RpcException.FORBIDDEN_EXCEPTION, "Forbid consumer " + NetUtils.getLocalHost() + " access service " + getInterface().getName() + " from registry " + getUrl().getAddress() + " use dubbo version " + Version.getVersion() + ", Please check registry access list (whitelist/blacklist).");
}代理集合獲取完畢以後就會開始調用，最下面就是常見的屢次調用失敗的錯誤。
/**
* 失敗轉移，當出現失敗，重試其它服務器，一般用於讀操做，但重試會帶來更長延遲。
*
* <a href="http://en.wikipedia.org/wiki/Failover">Failover</a>
*
* @author william.liangf
* @author chao.liuc
*/
public class FailoverClusterInvoker<T> extends AbstractClusterInvoker<T> {
private static final Logger logger = LoggerFactory.getLogger(FailoverClusterInvoker.class);
public FailoverClusterInvoker(Directory<T> directory) {
super(directory);
}
@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);
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 {
Result result = invoker.invoke(invocation);
if (le != null && logger.isWarnEnabled()) {
logger.warn("Although retry the method " + invocation.getMethodName()
+ " 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 != 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);
}
}開始invoke調用，在DubboInvoker中出現了timeout的具體使用
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);
}
}在url中獲取該方法的timeout參數值，默認值爲Constants.DEFAULT_TIMEOUT=1000，先查詢有沒有設置方法級別的超時時間
public int getMethodParameter(String method, String key, int defaultValue) {
String methodKey = method + "." + key;
Number n = getNumbers().get(methodKey);
if (n != null) {
return n.intValue();
}
String value = getMethodParameter(method, key);
if (value == null || value.length() == 0) {
return defaultValue;
}
int i = Integer.parseInt(value);
getNumbers().put(methodKey, i);
return i;
}
public String getMethodParameter(String method, String key) {
String value = parameters.get(method + "." + key);
if (value == null || value.length() == 0) {
return getParameter(key);
}
return value;
}查找接口服務級別的超時時間，最後再查找有沒有全局的超時時間
public String getParameter(String key) {
String value = parameters.get(key);
if (value == null || value.length() == 0) {
value = parameters.get(Constants.DEFAULT_KEY_PREFIX + key);
}
return value;
}4.超時等待
return (Result) currentClient.request(inv, timeout).get();在DefaultFuture中oop
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()) {
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();
}最後拋出超時異常TimeoutException，DubboInvoker捕獲後包裝成
throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);繼續拋給FailoverClusterInvoker，根據重試次數retries決定是否換一個提供者再次重試
int len = getUrl().getMethodParameter(invocation.getMethodName(), Constants.RETRIES_KEY, Constants.DEFAULT_RETRIES) + 1;當重試次數達到後依然出現異常，則會拋出最終錯誤
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);