Dubbo集群簡介
阿新 • • 發佈：2019-04-07
director amp syn tac 打印日誌 基類 edt 切換 兩個
目錄
- 1 簡介
- 1.1 概述
- 1.2 Cluster和ClusterInvoker
- 2 ClusterInvoker分析
- 2.1 抽象基類
- 2.1.1 invoke
- 2.1.2 list
- 2.1.3 select
- 2.2 FailoverClusterInvoker
- 2.3 FailbackClusterInvoker
- 2.4 FailfastClusterInvoker
- 2.5 FailsafeClusterInvoker
- 2.1 抽象基類
1 簡介
1.1 概述
為避免單點故障，通常會部署多臺服務器。作為服務消費者，在調用時需要考慮：
- ① 選擇哪個服務提供者進行調用
- ② 若服務提供者調用失敗，將如何操作
ClusterInvoker選擇服務提供者的大概過程如下：
- 首先，調用
Directory#list()獲取Invoker列表（該列表經過路由過濾）： - 最後，通過
LoadBalance#select()獲取最終的Invoker
1.2 Cluster和ClusterInvoker
Dubbo定義了兩個集群對象：
- Cluster接口：用於創建ClusterInvoker
- ClusterInvoker抽象類：實現Invoker接口
示例代碼如下：
public class FailoverCluster implements Cluster { public final static String NAME = "failover"; @Override public <T> Invoker<T> join(Directory<T> directory) throws RpcException { return new FailoverClusterInvoker<T>(directory); } }
2 ClusterInvoker分析
2.1 抽象基類
2.1.1 invoke
該方法會執行如下邏輯：
- 綁定attachments到invocation中.
- 從服務目錄中獲取Invoker列表
- 執行Invoker
@Override
public Result invoke(final Invocation invocation) throws RpcException {
checkWhetherDestroyed();
LoadBalance loadbalance = null;
// 綁定attachments到invocation中.
Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();
if (contextAttachments != null && contextAttachments.size() != 0) {
((RpcInvocation) invocation).addAttachments(contextAttachments);
}
// 從服務目錄中獲取Invoker列表
List<Invoker<T>> invokers = list(invocation);
if (invokers != null && !invokers.isEmpty()) {
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()
.getMethodParameter(RpcUtils.getMethodName(invocation), Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE));
}
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
// 執行Invoker
return doInvoke(invocation, invokers, loadbalance);
}2.1.2 list
該方法用於從服務目錄中獲取Invoker列表，代碼十分簡單，如下：
protected List<Invoker<T>> list(Invocation invocation) throws RpcException {
List<Invoker<T>> invokers = directory.list(invocation);
return invokers;
}2.1.3 select
select()方法主要邏輯：
- 對粘帶屬性的支持：讓服務消費者盡可能的調用同一個服務提供者，除非該提供者掛了再進行切換。
- doSelect()方法：根據負載均衡算法獲取一個服務提供者Invoker
// invokers：服務目錄中獲取的所有Invoker列表
// invoked：在本次調用中，已經執行過的invokers列表
protected Invoker<T> select(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
if (invokers == null || invokers.isEmpty())
return null;
// 獲取調用方法
String methodName = invocation == null ? "" : invocation.getMethodName();
// 獲取sticky屬性（粘帶屬性）
boolean sticky = invokers.get(0).getUrl().getMethodParameter(methodName, Constants.CLUSTER_STICKY_KEY, Constants.DEFAULT_CLUSTER_STICKY);
{
// 如果stickyInvoker!=null且invokers列表也不包含stickyInvoker，說明stickyInvoker代表的服務提供者掛了
if (stickyInvoker != null && !invokers.contains(stickyInvoker)) {
stickyInvoker = null;
}
//如果sticky==true且stickyInvoker!=null，此時如果selected包含stickyInvoker
//說明stickyInvoker對應服務提供者可能因為網絡問題未能成功提供服務，但該提供者並沒有掛（invokers仍包含stickyInvoker）
if (sticky && stickyInvoker != null && (selected == null || !selected.contains(stickyInvoker))) {
// availablecheck是否開啟可用性檢查，若開啟，檢測stickyInvoker是否可用
if (availablecheck && stickyInvoker.isAvailable()) {
return stickyInvoker;
}
}
}
// 若代碼走到此處，說明stickyInvoker==null，或不可用
Invoker<T> invoker = doSelect(loadbalance, invocation, invokers, selected);
if (sticky) {
stickyInvoker = invoker;
}
return invoker;
}doSelect方法用於真正去篩選Invoker：
- 利用負載均衡算法，選出一個Invoker
- 判斷Invoker是否可用，若不可用則進行重選（只做一次）
private Invoker<T> doSelect(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
if (invokers == null || invokers.isEmpty())
return null;
if (invokers.size() == 1)
return invokers.get(0);
if (loadbalance == null) {
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);
}
Invoker<T> invoker = loadbalance.select(invokers, getUrl(), invocation);
//如果selected包含負載均衡選擇出的Invoker，或者該Invoker無法經過可用性檢查，此時進行重選
if ((selected != null && selected.contains(invoker))
|| (!invoker.isAvailable() && getUrl() != null && availablecheck)) {
try {
Invoker<T> rinvoker = reselect(loadbalance, invocation, invokers, selected, availablecheck);
if (rinvoker != null) {
invoker = rinvoker;
} else {
//Check the index of current selected invoker, if it's not the last one, choose the one at index+1.
int index = invokers.indexOf(invoker);
try {
//Avoid collision
invoker = index < invokers.size() - 1 ? invokers.get(index + 1) : invokers.get(0);
} catch (Exception e) { /* 日誌 */ }
}
} catch (Throwable t) { /* 日誌 */ }
}
return invoker;
}2.2 FailoverClusterInvoker
十分簡單：失敗則循環調用。循環次數可配置，默認3次。
public class FailoverClusterInvoker<T> extends AbstractClusterInvoker<T> {
private static final Logger logger = LoggerFactory.getLogger(FailoverClusterInvoker.class);
public FailoverClusterInvoker(Directory<T> directory) {
super(directory);
}
@Override
@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++) {
//Reselect before retry to avoid a change of candidate `invokers`.
//NOTE: if `invokers` changed, then `invoked` also lose accuracy.
if (i > 0) {
checkWhetherDestroyed();
copyinvokers = list(invocation);
// check again
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()) {
//
}
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(...);
}
}2.3 FailbackClusterInvoker
調用失敗，返回空結果給服務提供者，定時任務對失敗的調用進行補償，適合執行消息通知等操作，如下：
public class FailbackClusterInvoker<T> extends AbstractClusterInvoker<T> {
private static final Logger logger = LoggerFactory.getLogger(FailbackClusterInvoker.class);
private static final long RETRY_FAILED_PERIOD = 5 * 1000;
private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2,
new NamedInternalThreadFactory("failback-cluster-timer", true));
private final ConcurrentMap<Invocation, AbstractClusterInvoker<?>> failed = new ConcurrentHashMap<Invocation, AbstractClusterInvoker<?>>();
private volatile ScheduledFuture<?> retryFuture;
public FailbackClusterInvoker(Directory<T> directory) {
super(directory);
}
// 放入失敗隊列Map結構中
private void addFailed(Invocation invocation, AbstractClusterInvoker<?> router) {
if (retryFuture == null) {
synchronized (this) {
if (retryFuture == null) {
retryFuture = scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
// collect retry statistics
try {
retryFailed();
} catch (Throwable t) { // Defensive fault tolerance
logger.error("Unexpected error occur at collect statistic", t);
}
}
}, RETRY_FAILED_PERIOD, RETRY_FAILED_PERIOD, TimeUnit.MILLISECONDS);
}
}
}
failed.put(invocation, router);
}
// 失敗重試
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);
}
}
}
@Override
protected 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) {
/* 日誌 */
addFailed(invocation, this);
return new RpcResult(); // ignore
}
}
}2.4 FailfastClusterInvoker
調用一次，快速失敗，失敗後立即拋出異常：
public class FailfastClusterInvoker<T> extends AbstractClusterInvoker<T> {
public FailfastClusterInvoker(Directory<T> directory) {
super(directory);
}
@Override
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);
}
}
}
2.5 FailsafeClusterInvoker
調用一次，快速失敗，失敗僅打印日誌，不拋出異常
public class FailsafeClusterInvoker<T> extends AbstractClusterInvoker<T> {
private static final Logger logger = LoggerFactory.getLogger(FailsafeClusterInvoker.class);
public FailsafeClusterInvoker(Directory<T> directory) {
super(directory);
}
@Override
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
}
}
}Dubbo集群簡介