你好,我是杨四正,今天我和你分享的主题是集群容错:一个好汉三个帮(上篇)。
在前面的课时中,我们已经对 Directory、Router、LoadBalance 等概念进行了深入的剖析,本课时将重点分析 Cluster 接口的相关内容。
Cluster 接口提供了我们常说的集群容错功能。
集群中的单个节点有一定概率出现一些问题,例如,磁盘损坏、系统崩溃等,导致节点无法对外提供服务,因此在分布式 RPC 框架中,必须要重视这种情况。为了避免单点故障,我们的 Provider 通常至少会部署在两台服务器上,以集群的形式对外提供服务,对于一些负载比较高的服务,则需要部署更多 Provider 来抗住流量。
在 Dubbo 中,通过 Cluster 这个接口把一组可供调用的 Provider 信息组合成为一个统一的 Invoker 供调用方进行调用。经过 Router 过滤、LoadBalance 选址之后,选中一个具体 Provider 进行调用,如果调用失败,则会按照集群的容错策略进行容错处理。
Dubbo 默认内置了若干容错策略,并且每种容错策略都有自己独特的应用场景,我们可以通过配置选择不同的容错策略。如果这些内置容错策略不能满足需求,我们还可以通过自定义容错策略进行配置。
了解了上述背景知识之后,下面我们就正式开始介绍 Cluster 接口。
Cluster 的工作流程大致可以分为两步(如下图所示):①创建 Cluster Invoker 实例(在 Consumer 初始化时,Cluster 实现类会创建一个 Cluster Invoker 实例,即下图中的 merge 操作);②使用 Cluster Invoker 实例(在 Consumer 服务消费者发起远程调用请求的时候,Cluster Invoker 会依赖前面课时介绍的 Directory、Router、LoadBalance 等组件得到最终要调用的 Invoker 对象)。
Cluster 核心流程图
Cluster Invoker 获取 Invoker 的流程大致可描述为如下:
这个过程是一个正常流程,没有涉及容错处理。Dubbo 中常见的容错方式有如下几个。
下面我们再来看 Cluster 接口。Cluster 接口是一个扩展接口,通过 @SPI 注解的参数我们知道其使用的默认实现是 FailoverCluster,它只定义了一个 join() 方法,在其上添加了 @Adaptive 注解,会动态生成适配器类,其中会优先根据 Directory.getUrl() 方法返回的 URL 中的 cluster 参数值选择扩展实现,若无 cluster 参数则使用默认的 FailoverCluster 实现。Cluster 接口的具体定义如下所示:
@SPI(FailoverCluster.NAME)
public interface Cluster {
@Adaptive
<T> Invoker<T> join(Directory<T> directory) throws RpcException;
}
Cluster 接口的实现类如下图所示,分别对应前面提到的多种容错策略:
Cluster 接口继承关系
在每个 Cluster 接口实现中,都会创建对应的 Invoker 对象,这些都继承自 AbstractClusterInvoker 抽象类,如下图所示:
AbstractClusterInvoker 继承关系图
通过上面两张继承关系图我们可以看出,Cluster 接口和 Invoker 接口都会有相应的抽象实现类,这些抽象实现类都实现了一些公共能力。下面我们就来深入介绍 AbstractClusterInvoker 和 AbstractCluster 这两个抽象类。
了解了 Cluster Invoker 的继承关系之后,我们首先来看 AbstractClusterInvoker,它有两点核心功能:一个是实现的 Invoker 接口,对 Invoker.invoke() 方法进行通用的抽象实现;另一个是实现通用的负载均衡算法。
在 AbstractClusterInvoker.invoke() 方法中,会通过 Directory 获取 Invoker 列表,然后通过 SPI 初始化 LoadBalance,最后调用 doInvoke() 方法执行子类的逻辑。在 Directory.list() 方法返回 Invoker 集合之前,已经使用 Router 进行了一次筛选,你可以回顾前面[第 31 课时]对 RegistryDirectory 的分析。
public Result invoke(final Invocation invocation) throws RpcException {
// 检测当前Invoker是否已销毁
checkWhetherDestroyed();
// 将RpcContext中的attachment添加到Invocation中
Map<String, Object> contextAttachments = RpcContext.getContext().getObjectAttachments();
if (contextAttachments != null && contextAttachments.size() != 0) {
((RpcInvocation) invocation).addObjectAttachments(contextAttachments);
}
// 通过Directory获取Invoker对象列表,通过对RegistryDirectory的介绍我们知道,其中已经调用了Router进行过滤
List<Invoker<T>> invokers = list(invocation);
// 通过SPI加载LoadBalance
LoadBalance loadbalance = initLoadBalance(invokers, invocation);
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
// 调用doInvoke()方法,该方法是个抽象方法
return doInvoke(invocation, invokers, loadbalance);
}
protected List<Invoker<T>> list(Invocation invocation) throws RpcException {
return directory.list(invocation); // 调用Directory.list()方法
}
下面我们来看一下 AbstractClusterInvoker 是如何按照不同的 LoadBalance 算法从 Invoker 集合中选取最终 Invoker 对象的。
AbstractClusterInvoker 并没有简单粗暴地使用 LoadBalance.select() 方法完成负载均衡,而是做了进一步的封装,具体实现在 select() 方法中。在 select() 方法中会根据配置决定是否开启粘滞连接特性,如果开启了,则需要将上次使用的 Invoker 缓存起来,只要 Provider 节点可用就直接调用,不会再进行负载均衡。如果调用失败,才会重新进行负载均衡,并且排除已经重试过的 Provider 节点。
// 第一个参数是此次使用的LoadBalance实现,第二个参数Invocation是此次服务调用的上下文信息,
// 第三个参数是待选择的Invoker集合,第四个参数用来记录负载均衡已经选出来、尝试过的Invoker集合
protected Invoker<T> select(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
if (CollectionUtils.isEmpty(invokers)) {
return null;
}
// 获取调用方法名
String methodName = invocation == null ? StringUtils.EMPTY_STRING : invocation.getMethodName();
// 获取sticky配置,sticky表示粘滞连接,所谓粘滞连接是指Consumer会尽可能地
// 调用同一个Provider节点,除非这个Provider无法提供服务
boolean sticky = invokers.get(0).getUrl()
.getMethodParameter(methodName, CLUSTER_STICKY_KEY, DEFAULT_CLUSTER_STICKY);
// 检测invokers列表是否包含sticky Invoker,如果不包含,
// 说明stickyInvoker代表的服务提供者挂了,此时需要将其置空
if (stickyInvoker != null && !invokers.contains(stickyInvoker)) {
stickyInvoker = null;
}
// 如果开启了粘滞连接特性,需要先判断这个Provider节点是否已经重试过了
if (sticky && stickyInvoker != null // 表示粘滞连接
&& (selected == null || !selected.contains(stickyInvoker)) // 表示stickyInvoker未重试过
) {
// 检测当前stickyInvoker是否可用,如果可用,直接返回stickyInvoker
if (availablecheck && stickyInvoker.isAvailable()) {
return stickyInvoker;
}
}
// 执行到这里,说明前面的stickyInvoker为空,或者不可用
// 这里会继续调用doSelect选择新的Invoker对象
Invoker<T> invoker = doSelect(loadbalance, invocation, invokers, selected);
if (sticky) { // 是否开启粘滞,更新stickyInvoker字段
stickyInvoker = invoker;
}
return invoker;
}
doSelect() 方法主要做了两件事:
private Invoker<T> doSelect(LoadBalance loadbalance, Invocation invocation,
List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
// 判断是否需要进行负载均衡,Invoker集合为空,直接返回null
if (CollectionUtils.isEmpty(invokers)) {
return null;
}
if (invokers.size() == 1) { // 只有一个Invoker对象,直接返回即可
return invokers.get(0);
}
// 通过LoadBalance实现选择Invoker对象
Invoker<T> invoker = loadbalance.select(invokers, getUrl(), invocation);
// 如果LoadBalance选出的Invoker对象,已经尝试过请求了或不可用,则需要调用reselect()方法重选
if ((selected != null && selected.contains(invoker)) // Invoker已经尝试调用过了,但是失败了
|| (!invoker.isAvailable() && getUrl() != null && availablecheck) // Invoker不可用
) {
try {
// 调用reselect()方法重选
Invoker<T> rInvoker = reselect(loadbalance, invocation, invokers, selected, availablecheck);
// 如果重选的Invoker对象不为空,则直接返回这个 rInvoker
if (rInvoker != null) {
invoker = rInvoker;
} else {
int index = invokers.indexOf(invoker);
try {
// 如果重选的Invoker对象为空,则返回该Invoker的下一个Invoker对象
invoker = invokers.get((index + 1) % invokers.size());
} catch (Exception e) {
logger.warn("...");
}
}
} catch (Throwable t) {
logger.error("...");
}
}
return invoker;
}
reselect() 方法会重新进行一次负载均衡,首先对未尝试过的可用 Invokers 进行负载均衡,如果已经全部重试过了,则将尝试过的 Provider 节点过滤掉,然后在可用的 Provider 节点中重新进行负载均衡。
private Invoker<T> reselect(LoadBalance loadbalance, Invocation invocation,
List<Invoker<T>> invokers, List<Invoker<T>> selected, boolean availablecheck) throws RpcException {
// 用于记录要重新进行负载均衡的Invoker集合
List<Invoker<T>> reselectInvokers = new ArrayList<>(
invokers.size() > 1 ? (invokers.size() - 1) : invokers.size());
// 将不在selected集合中的Invoker过滤出来进行负载均衡
for (Invoker<T> invoker : invokers) {
if (availablecheck && !invoker.isAvailable()) {
continue;
}
if (selected == null || !selected.contains(invoker)) {
reselectInvokers.add(invoker);
}
}
// reselectInvokers不为空时,才需要通过负载均衡组件进行选择
if (!reselectInvokers.isEmpty()) {
return loadbalance.select(reselectInvokers, getUrl(), invocation);
}
// 只能对selected集合中可用的Invoker再次进行负载均衡
if (selected != null) {
for (Invoker<T> invoker : selected) {
if ((invoker.isAvailable()) // available first
&& !reselectInvokers.contains(invoker)) {
reselectInvokers.add(invoker);
}
}
}
if (!reselectInvokers.isEmpty()) {
return loadbalance.select(reselectInvokers, getUrl(), invocation);
}
return null;
}
常用的 ClusterInvoker 实现都继承了 AbstractClusterInvoker 类型,对应的 Cluster 扩展实现都继承了 AbstractCluster 抽象类。AbstractCluster 抽象类的核心逻辑是在 ClusterInvoker 外层包装一层 ClusterInterceptor,从而实现类似切面的效果。
下面是 ClusterInterceptor 接口的定义:
@SPI
public interface ClusterInterceptor {
// 前置拦截方法
void before(AbstractClusterInvoker<?> clusterInvoker, Invocation invocation);
// 后置拦截方法
void after(AbstractClusterInvoker<?> clusterInvoker, Invocation invocation);
// 调用ClusterInvoker的invoke()方法完成请求
default Result intercept(AbstractClusterInvoker<?> clusterInvoker, Invocation invocation) throws RpcException {
return clusterInvoker.invoke(invocation);
}
// 这个Listener用来监听请求的正常结果以及异常
interface Listener {
void onMessage(Result appResponse, AbstractClusterInvoker<?> clusterInvoker, Invocation invocation);
void onError(Throwable t, AbstractClusterInvoker<?> clusterInvoker, Invocation invocation);
}
}
在 AbstractCluster 抽象类的 join() 方法中,首先会调用 doJoin() 方法获取最终要调用的 Invoker 对象,doJoin() 是个抽象方法,由 AbstractCluster 子类根据具体的策略进行实现。之后,AbstractCluster.join() 方法会调用 buildClusterInterceptors() 方法加载 ClusterInterceptor 扩展实现类,对 Invoker 对象进行包装。具体实现如下:
private <T> Invoker<T> buildClusterInterceptors(AbstractClusterInvoker<T> clusterInvoker, String key) {
AbstractClusterInvoker<T> last = clusterInvoker;
// 通过SPI方式加载ClusterInterceptor扩展实现
List<ClusterInterceptor> interceptors = ExtensionLoader.getExtensionLoader(ClusterInterceptor.class).getActivateExtension(clusterInvoker.getUrl(), key);
if (!interceptors.isEmpty()) {
for (int i = interceptors.size() - 1; i >= 0; i--) {
// 将InterceptorInvokerNode收尾连接到一起,形成调用链
final ClusterInterceptor interceptor = interceptors.get(i);
final AbstractClusterInvoker<T> next = last;
last = new InterceptorInvokerNode<>(clusterInvoker, interceptor, next);
}
}
return last;
}
@Override
public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
// 扩展名称由reference.interceptor参数确定
return buildClusterInterceptors(doJoin(directory), directory.getUrl().getParameter(REFERENCE_INTERCEPTOR_KEY));
}
InterceptorInvokerNode 会将底层的 AbstractClusterInvoker 对象以及关联的 ClusterInterceptor 对象封装到一起,还会维护一个 next 引用,指向下一个 InterceptorInvokerNode 对象。
在 InterceptorInvokerNode.invoke() 方法中,会先调用 ClusterInterceptor 的前置逻辑,然后执行 intercept() 方法调用 AbstractClusterInvoker 的 invoke() 方法完成远程调用,最后执行 ClusterInterceptor 的后置逻辑。具体实现如下:
public Result invoke(Invocation invocation) throws RpcException {
Result asyncResult;
try {
interceptor.before(next, invocation); // 前置逻辑
// 执行invoke()方法完成远程调用
asyncResult = interceptor.intercept(next, invocation);
} catch (Exception e) {
if (interceptor instanceof ClusterInterceptor.Listener) {
// 出现异常时,会触发监听器的onError()方法
ClusterInterceptor.Listener listener = (ClusterInterceptor.Listener) interceptor;
listener.onError(e, clusterInvoker, invocation);
}
throw e;
} finally {
// 执行后置逻辑
interceptor.after(next, invocation);
}
return asyncResult.whenCompleteWithContext((r, t) -> {
if (interceptor instanceof ClusterInterceptor.Listener) {
ClusterInterceptor.Listener listener = (ClusterInterceptor.Listener) interceptor;
if (t == null) {
// 正常返回时,会调用onMessage()方法触发监听器
listener.onMessage(r, clusterInvoker, invocation);
} else {
listener.onError(t, clusterInvoker, invocation);
}
}
});
}
Dubbo 提供了两个 ClusterInterceptor 实现类,分别是 ConsumerContextClusterInterceptor 和 ZoneAwareClusterInterceptor,如下图所示:
ClusterInterceptor 继承关系图
在 ConsumerContextClusterInterceptor 的 before() 方法中,会在 RpcContext 中设置当前 Consumer 地址、此次调用的 Invoker 等信息,同时还会删除之前与当前线程绑定的 Server Context。在 after() 方法中,会删除本地 RpcContext 的信息。ConsumerContextClusterInterceptor 的具体实现如下:
public void before(AbstractClusterInvoker<?> invoker, Invocation invocation) {
// 获取当前线程绑定的RpcContext
RpcContext context = RpcContext.getContext();
// 设置Invoker、Consumer地址等信息 context.setInvocation(invocation).setLocalAddress(NetUtils.getLocalHost(), 0);
if (invocation instanceof RpcInvocation) {
((RpcInvocation) invocation).setInvoker(invoker);
}
RpcContext.removeServerContext();
}
public void after(AbstractClusterInvoker<?> clusterInvoker, Invocation invocation) {
RpcContext.removeContext(true); // 删除本地RpcContext的信息
}
ConsumerContextClusterInterceptor 同时继承了 ClusterInterceptor.Listener 接口,在其 onMessage() 方法中,会获取响应中的 attachments 并设置到 RpcContext 中的 SERVER_LOCAL 之中,具体实现如下:
public void onMessage(Result appResponse, AbstractClusterInvoker<?> invoker, Invocation invocation) {
// 从AppResponse中获取attachment,并设置到SERVER_LOCAL这个RpcContext中 RpcContext.getServerContext().setObjectAttachments(appResponse.getObjectAttachments());
}
介绍完 ConsumerContextClusterInterceptor,我们再来看 ZoneAwareClusterInterceptor。
在 ZoneAwareClusterInterceptor 的 before() 方法中,会从 RpcContext 中获取多注册中心相关的参数并设置到 Invocation 中(主要是 registry_zone 参数和 registry_zone_force 参数,这两个参数的具体含义,在后面分析 ZoneAwareClusterInvoker 时详细介绍),ZoneAwareClusterInterceptor 的 after() 方法为空实现。ZoneAwareClusterInterceptor 的具体实现如下:
public void before(AbstractClusterInvoker<?> clusterInvoker, Invocation invocation) {
RpcContext rpcContext = RpcContext.getContext();
// 从RpcContext中获取registry_zone参数和registry_zone_force参数
String zone = (String) rpcContext.getAttachment(REGISTRY_ZONE);
String force = (String) rpcContext.getAttachment(REGISTRY_ZONE_FORCE);
// 检测用户是否提供了ZoneDetector接口的扩展实现
ExtensionLoader<ZoneDetector> loader = ExtensionLoader.getExtensionLoader(ZoneDetector.class);
if (StringUtils.isEmpty(zone) && loader.hasExtension("default")) {
ZoneDetector detector = loader.getExtension("default");
zone = detector.getZoneOfCurrentRequest(invocation);
force = detector.isZoneForcingEnabled(invocation, zone);
}
// 将registry_zone参数和registry_zone_force参数设置到Invocation中
if (StringUtils.isNotEmpty(zone)) {
invocation.setAttachment(REGISTRY_ZONE, zone);
}
if (StringUtils.isNotEmpty(force)) {
invocation.setAttachment(REGISTRY_ZONE_FORCE, force);
}
}
需要注意的是,ZoneAwareClusterInterceptor 没有实现 ClusterInterceptor.Listener 接口,也就是不提供监听响应的功能。
本课时我们主要介绍的是 Dubbo Cluster 层中容错机制相关的内容。首先,我们了解了集群容错机制的作用。然后,我们介绍了 Cluster 接口的定义以及其各个实现类的核心功能。之后,我们深入讲解了 AbstractClusterInvoker 的实现,其核心是实现了一套通用的负载均衡算法。最后,我们还分析了 AbstractCluster 抽象实现类以及其中涉及的 ClusterInterceptor 接口的内容。