dubbo集羣容錯策略的代碼分析2
接上篇http://www.javashuo.com/article/p-dwqnopaq-hc.htmljava
dubbo版本2.5.3緩存
dubbo自己集羣容錯策略有7種。都實現了Cluster接口（spi擴展點）
從類結構上看，Cluster接口有9個實現類（其中MockClusterWrapper是服務降級處理用的，MergeableCluster是分組合並結果用的）安全
Cluster接口只有一個方法服務器
/**
* Merge the directory invokers to a virtual invoker.
*
* @param <T>
* @param directory
* @return cluster invoker
* @throws RpcException
*/
@Adaptive
<T> Invoker<T> join(Directory<T> directory) throws RpcException;
方法實現邏輯是，把directory目錄服務中多個提供者，通過容錯和負載均衡機制包裝，以一個虛擬的Invoker返給上層傳調用。
每一個虛擬的Invoker類型，就是一種集羣策略。併發
好比dubbo默認的集羣策略failover類的實現app
public class FailoverCluster implements Cluster {
public final static String NAME = "failover";
public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
return new FailoverClusterInvoker<T>(directory);//包裝成的虛擬Invoker類型是FailoverClusterInvoker，就是個集羣容錯策略
}
}
FailoverClusterInvoker 類擴展了抽象類AbstractClusterInvoker，實現了AbstractClusterInvoker的
抽象方法doInvoke()用於實現具體集羣策略，以下圖負載均衡
AbstractClusterInvoker實現了Invoker接口惟一方法invoke,對外層調用，以下ide
public Result invoke(final Invocation invocation) throws RpcException {
checkWhetherDestroyed();
LoadBalance loadbalance;
//從目錄中獲取全部的服務提供者
List<Invoker<T>> invokers = list(invocation);
//獲取負載均衡策略
if (invokers != null && invokers.size() > 0) {
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()
.getMethodParameter(invocation.getMethodName(), Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE));
} else {
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);
}
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
return doInvoke(invocation, invokers, loadbalance);//調用子類實現具體的容錯策略。
}
能夠看到其餘幾種集羣策略都是這種方式。其實就是模板方法模式。this
因此，經過看每種集羣容錯類的doInvoke方法的具體實現，就能夠理解每種的容錯策略。
前一篇，看了failover和available集羣策略,下面再看看其餘五種集羣策略。.net
broadcast策略：
public Result doInvoke(final Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
checkInvokers(invokers, invocation);
RpcContext.getContext().setInvokers((List) invokers);
RpcException exception = null;
Result result = null;
//遍歷調用全部的服務列表，並把結果覆蓋之前的。
for (Invoker<T> invoker : invokers) {
try {
result = invoker.invoke(invocation);
} catch (RpcException e) {
exception = e;
logger.warn(e.getMessage(), e);
} catch (Throwable e) {
exception = new RpcException(e.getMessage(), e);
logger.warn(e.getMessage(), e);
}
}
//其中有一個失敗，直接拋異常
if (exception != null) {
throw exception;
}
return result;
}
這個策略一般用於通知全部提供者更新緩存或日誌等本地資源信息
forking 策略：
public Result doInvoke(final Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
checkInvokers(invokers, invocation);
final List<Invoker<T>> selected;
//獲取並行調用個數
final int forks = getUrl().getParameter(Constants.FORKS_KEY, Constants.DEFAULT_FORKS);
//超時時間
final int timeout = getUrl().getParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
if (forks <= 0 || forks >= invokers.size()) {
selected = invokers;
} else {
selected = new ArrayList<Invoker<T>>();
//經過負載均衡策略，選出要並行調用的invokers，放入selected列表
for (int i = 0; i < forks; i++) {
//在invoker列表(排除selected)後,若是沒有選夠,則存在重複循環問題.見select實現.
Invoker<T> invoker = select(loadbalance, invocation, invokers, selected);
if (!selected.contains(invoker)) {//防止重複添加invoker
selected.add(invoker);
}
}
}
RpcContext.getContext().setInvokers((List) selected);
final AtomicInteger count = new AtomicInteger();
final BlockingQueue<Object> ref = new LinkedBlockingQueue<Object>();
//遍歷selected列表，經過線程池併發調用
for (final Invoker<T> invoker : selected) {
executor.execute(new Runnable() {
public void run() {
try {
Result result = invoker.invoke(invocation);
//把結果放入隊列
ref.offer(result);
} catch (Throwable e) {
int value = count.incrementAndGet();
//全部的都異常了，才把異常加入到對了尾部
//這就保證了，只要有一個成功，ref.poll()方法從隊列頭部就能取獲得結果返回。
if (value >= selected.size()) {
ref.offer(e);
}
}
}
});
}
try {
//從隊列頭部就能取獲得結果,返回，若是是異常，就拋出。
Object ret = ref.poll(timeout, TimeUnit.MILLISECONDS);
if (ret instanceof Throwable) {
Throwable e = (Throwable) ret;
throw new RpcException(e instanceof RpcException ? ((RpcException) e).getCode() : 0, "Failed to forking invoke provider " + selected + ", but no luck to perform the invocation. Last error is: " + e.getMessage(), e.getCause() != null ? e.getCause() : e);
}
return (Result) ret;
} catch (InterruptedException e) {
throw new RpcException("Failed to forking invoke provider " + selected + ", but no luck to perform the invocation. Last error is: " + e.getMessage(), e);
}
}
並行調用多個服務器，只要一個成功即返回。一般用於實時性要求較高的讀操做，但須要浪
費更多服務資源。可經過 forks="2" 來設置最大並行數。
failback策略：
protected Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
try {
checkInvokers(invokers, invocation);
//經過負載均衡策略選擇一個invoker
Invoker<T> invoker = select(loadbalance, invocation, invokers, null);
return invoker.invoke(invocation);
} catch (Throwable e) {
//失敗後，記錄異常日誌，不拋出異常，
logger.error("Failback to invoke method " + invocation.getMethodName() + ", wait for retry in background. Ignored exception: "
+ e.getMessage() + ", ", e);
//把異常調用記錄入異常hashmap,key是調用的方法信息，value是invoker自己
addFailed(invocation, this);
return new RpcResult(); // ignore
}
}
/***
* 建立調度器，放入重試對象
*/
private void addFailed(Invocation invocation, AbstractClusterInvoker<?> router) {
if (retryFuture == null) {
synchronized (this) {
if (retryFuture == null) {
//調度線程池，週期性（5秒一次）的調用retryFailed方法
retryFuture = scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {
public void run() {
// 收集統計信息
try {
//執行以前異常方法的調用
retryFailed();
} catch (Throwable t) { // 防護性容錯
logger.error("Unexpected error occur at collect statistic", t);
}
}
}, RETRY_FAILED_PERIOD, RETRY_FAILED_PERIOD, TimeUnit.MILLISECONDS);
}
}
}
//放入map
failed.put(invocation, router);
}
/***
* 遍歷失敗hashmap failed 取出調用環境棧，執行調用
*/
void retryFailed() {
if (failed.size() == 0) {
return;
}
for (Map.Entry<Invocation, AbstractClusterInvoker<?>> entry : new HashMap<Invocation, AbstractClusterInvoker<?>>(
failed).entrySet()) {
Invocation invocation = entry.getKey();
Invoker<?> invoker = entry.getValue();
try {
//執行調用
invoker.invoke(invocation);
failed.remove(invocation);
} catch (Throwable e) {
logger.error("Failed retry to invoke method " + invocation.getMethodName() + ", waiting again.", e);
}
}
}
此策略失敗自動恢復，後臺記錄失敗請求，定時重發。一般用於消息通知操做。
failsafe策略：
public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
try {
checkInvokers(invokers, invocation);
//利用負載均衡選擇一個調用者
Invoker<T> invoker = select(loadbalance, invocation, invokers, null);
return invoker.invoke(invocation);
} catch (Throwable e) {
//若是有異常，記錄異常信息，返回空值，不拋出異常
logger.error("Failsafe ignore exception: " + e.getMessage(), e);
return new RpcResult(); // ignore
}
}
失敗安全，出現異常時，直接忽略。一般用於寫入審計日誌等操做。
failfast策略
public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
checkInvokers(invokers, invocation);
Invoker<T> invoker = select(loadbalance, invocation, invokers, null);
try {
return invoker.invoke(invocation);
} catch (Throwable e) {
//若是有一次異常，當即拋出異常
if (e instanceof RpcException && ((RpcException) e).isBiz()) { // biz exception.
throw (RpcException) e;
}
throw new RpcException(e instanceof RpcException ? ((RpcException) e).getCode() : 0, "Failfast invoke providers " + invoker.getUrl() + " " + loadbalance.getClass().getSimpleName() + " select from all providers " + invokers + " for service " + getInterface().getName() + " method " + invocation.getMethodName() + " on consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + ", but no luck to perform the invocation. Last error is: " + e.getMessage(), e.getCause() != null ? e.getCause() : e);
}
}
快速失敗，只發起一次調用，失敗當即報錯。一般用於非冪等性的寫操做，好比新增記錄。最後有個終結篇http://www.javashuo.com/article/p-poudvoae-u.html
相關文章
- 1. dubbo集羣容錯策略的代碼分析3
- 2. dubbo 集羣容錯策略
- 3. 源碼分析Dubbo集羣容錯策略
- 4. Dubbo源碼解析之Cluster集羣容錯策略介紹
- 5. dubbo源碼分析-集羣容錯(二)
- 6. Dubbo容錯策略
- 7. Dubbo剖析-集羣容錯
- 8. Dubbo的容錯策略
- 9. dubbo集羣容錯機制代碼分析1
- 10. Dubbo 負載均衡策略和集羣容錯策略都有哪些？動態代理策略呢？
- 更多相關文章...
- • Markdown 代碼 - Markdown 教程
- • Swarm 集羣管理 - Docker教程
- • IntelliJ IDEA代碼格式化設置
- • Docker容器實戰(七) - 容器眼光下的文件系統
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
- 1. 外部其他進程嵌入到qt FindWindow獲得窗口句柄 報錯無法鏈接的外部符號 [email protected] 無法被([email protected]@[email protected]@@引用
- 2. UVa 11524 - InCircle
- 3. The Monocycle（bfs）
- 4. VEC-C滑窗
- 5. 堆排序的應用-TOPK問題
- 6. 實例演示ElasticSearch索引查詢term,match,match_phase,query_string之間的區別
- 7. 數學基礎知識 集合
- 8. amazeUI 復擇框問題解決
- 9. 揹包問題理解
- 10. 算數平均-幾何平均不等式的證明,從麥克勞林到柯西
歡迎關注本站公眾號,獲取更多信息
相關文章