java开发Dubbo负载均衡与集群容错示例详解

java开发Dubbo负载均衡与集群容错示例详解

目录负载均衡与集群容错Invoker服务目录RegistryDirectory获取Invoker列表监听注册中心刷新Invoker列表StaticDirectory服务路由ClusterFailoverClusterInvokerFailfastClusterInvokerFailsafeClusterInvokerFailbackClusterInvokerForkingClusterInvokerBroadcastClusterInvokerAbstractClusterInvoker小结负载均衡AbstractLoadBalanceRandomLoadBalanceLeastActiveLoadBalanceConsistentHashLoadBalanceRoundRobinLoadBalance总结

负载均衡与集群容错

Invoker

在Dubbo中Invoker就是一个具有调用功能的对象，在服务提供端就是实际的服务实现，只是将服务实现封装起来变成一个Invoker。

在服务消费端，从注册中心得到服务提供者的信息之后，将一条条信息封装为Invoker，这个Invoker就具备了远程调用的能力。

综上，Dubbo就是创建了一个统一的模型，将可调用（可执行体）的服务对象都统一封装为Invoker。

而ClusterInvoker就是将多个服务引入的Invoker封装起来，对外统一暴露一个Invoker，并且赋予这些Invoker集群容错的功能。

服务目录

服务目录，即Directory，实际上它就是多个Invoker的集合，服务提供端一般都会集群分布，同样的服务会有多个提供者，因此需要一个服务目录来统一存放它们，需要调用服务的时候便从这个服务目录中进行挑选。

同时服务目录还是实现了NotifyListener接口，当集群中新增了一台服务提供者或者下线了一台服务提供者，目录都会对服务提供者进行更新，新增或者删除对应的Invoker。

从上图中http://，可以看到用了一个抽象类AbstractDirectory来实现 Directory接口，抽象类中运用到了模板方法模式，将一些公共方法和逻辑写好，作为一个骨架，然后具体实现由了两个子类来完成，两个子类分别为StaticDirectory和RegistryDirectory。

RegistryDirectory

RegistryDirectory实现了NotifyListener接口，可以监听注册中心的变化，当注册中心配置发生变化时，服务目录也可以收到变更通知，然后根据更新之后的配置刷新Invoker列表。

由此可知RegistryDirectory共有三个作用：

获取Invoker列表监听注册中心刷新Invoker列表

获取Invoker列表

RegistryDirectory实现了父类AbstractDirectory的抽象方法doList()，该方法可以得到Invoker列表

public List> doList(Invocation invocation) {

if (this.forbidden) {

throw new RpcException(....);

} else {

List> invokers = null;

Map>> localMethodInvokerMap = this.methodInvokerMap; //获取方法调用名和Invoker的映射表

if (localMethodInvokerMap != null && localMethodInvokerMap.size() > 0) {

String methodName = RpcUtils.getMethodName(invocation);

Object[] args = RpcUtils.getArguments(invocation);

//以下就是根据方法名和方法参数获取可调用的Invoker

if (args != null && args.length > 0 && args[0] != null && (args[0] instanceof String || args[0].getClass().isEnum())) {

invokers = (List)localMethodInvokerMap.get(methodName + "." + args[0]);

}

if (invokers == null) {

invokers = (List)localMethodInvokerMap.get(methodName);

}

if (invokers == null) {

invokers = (List)localMethodInvokerMap.get("*");

}

if (invokers == null) {

Iterator>> iterator = localMethodInvokerMap.values().iterator();

if (iterator.hasNext()) {

invokers = (List)iterator.next();

}

}

}

return (List)(invokers == null ? new ArrayList(0) : invokers);

}

}

监听注册中心

通过实现NotifyListener接口可以感知注册中心的数据变更。

RegistryDirectory定义了三个集合invokerUrls routerUrls configuratorUrls分别处理对应的配置然后转化成对象。

public synchronized void notify(List urls) {

List invokerUrls = new ArrayLihttp://st();

List routerUrls = new ArrayList();

List configuratorUrls = new ArrayList();

Iterator i$ = urls.iterator();

while(true) {

while(true) {

while(i$.hasNext()) {

//....根据urls填充上述三个列表

}

if (configuratorUrls != null && !configuratorUrls.isEmpty()) {

this.configurators = toConfigurators(configuratorUrls); //根据urls转化为configurators配置

}

List localConfigurators;

if (routerUrls != null && !routerUrls.isEmpty()) {

localConfigurators = this.toRouters(routerUrls);

if (localConfigurators != null) {

this.setRouters(localConfigurators); //根据urls转化为routers配置

}

}

localConfigurators = this.configurators;

this.overrideDirectoryUrl = this.directoryUrl;

Configurator configurator;

if (localConfigurators != null && !localConfigurators.isEmpty()) {

for(Iterator i$ = localConfigurators.iterator(); i$.hasNext(); this.overrideDirectoryUrl = configurator.configure(this.overrideDirectoryUrl)) {

configurator = (Configurator)i$.next();

}

}

this.refreshInvoker(invokerUrls); //根据invokerUrls刷新invoker列表

return;

}

}

}

刷新Invoker列表

private void refreshInvoker(List invokerUrls) {

//如果invokerUrls只有一个URL并且协议是empty，那么清除所有invoker

if (invokerUrls != null && invokerUrls.size() == 1 && invokerUrls.get(0) != null && "empty".equals(((URL)invokerUrls.get(0)).getProtocol())) {

this.forbidden = true;

this.methodInvokerMap = null;

this.destroyAllInvokers();

} else {

this.forbidden = false;

Map> oldUrlInvokerMap = this.urlInvokerMap; //获取旧的Invoker列表

if (invokerUrls.isEmpty() && this.cachedInvokerUrls != null) {

invokerUrls.addAll(this.cachedInvokerUrls);

} else {

this.cachedInvokerUrls = new HashSet();

this.cachedInvokerUrls.addAll(invokerUrls);

}

if (invokerUrls.isEmpty()) {

return;

}

//根据URL生成InvokerMap

Map> newUrlInvokerMap = this.toInvokers(invokerUrls);

//根据新的InvokerMap生成方法名和Invoker列表对应的Map

Map>> newMethodInvokerMap = this.toMethodInvokers(newUrlInvokerMap);

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 = this.multiGroup ? this.toMergeMethodInvokerMap(newMethodInvokerMap) : newMethodInvokerMap;

this.urlInvokerMap = newUrlInvokerMap;

try {

this.destroyUnusedInvokers(oldUrlInvokerMap, newUrlInvokerMap); //销毁无效的Invoker

} catch (Exception var6) {

logger.warn("destroyUnusedInvokers error. ", var6);

}

}

}

上述操作就是根据invokerUrls数量以及协议头是否为empty来判断是否禁用所有invokers，如果不禁用的话将invokerUrls转化为Invoker，并且得到的映射关系。

再进一步进行转化，得到的映射关系，再将同一组的Invoker进行合并，将合并结果赋值给methodInvokerMap，这个methodInvokerMap就是在doList中使用到的Map。

最后刷新InvokerMap，销毁无效的Invoker。

StaticDirectory

StaticDirectory是静态目录，所有Invoker是固定的不会删减的，并且所有Invoker由构造器来传入。

内部逻辑也相当简单，只定义了一个列表用于存储Invokers。实现父类的方法也只是将这些Invokers原封不动地返回。

private final List> invokers;

protected List> doList(Invocation invocation) throws RpcException {

return this.invokers;

}

服务路由

服务路由规定了服务消费者可以调用哪些服务提供者，Dubbo常用的是条件路由ConditionRouter。

条件路由由两个条件组成，格式为[服务消费者匹配条件] => [服务提供者匹配条件]，例如172.26.29.15 => 172.27.19.89规定了只有IP为172.26.29.15的服务消费者才可以访问IP为172.27.19.89的服务提供者，不可以调用其他的服务。

路由一样是通过RegistryDirectory中的notify()更新的，在调用toMethodInvokers()的时候会进行服务器级别的路由和方法级别的路由。

Cluster

在前面的流程中我们已经通过Directory获取了服务目录，并且通过路由获取了一个或多个Invoker，但是对于服务消费者还是需要进行选择，筛选出一个Invoker进行调用。

Dubbo默认的Cluster实现有多种，如下：

FailoverClusterFailfastClusterFailsafeClusterFailbackClusterBroadcastClusterAvailableCluster

每个Cluster内部返回的都是xxxClusterInvoker，例如FailoverCluster：

public class FailoverCluster implements Cluster {

public static final String NAME = "failover";

public FailoverCluster() {

}

public Invoker join(Directory directory) throws RpcException {

return new FailoverClusterInvoker(directory);

}

}

FailoverClusterInvoker

FailoverClusterInvoker实现的功能是失败调用(有重试次数)自动切换。

public Result doInvoke(Invocation invocation, List> invokers, LoadBalance loadbalance) throws RpcException {

List> copyinvokers = invokers;

this.checkInvokers(invokers, invocation);

//重试次数

int len = this.getUrl().getMethodParameter(invocation.getMethodName(), "retries", 2) + 1;

if (len <= 0) {

len = 1;

}

RpcException le = null;

List> invoked = new ArrayList(invokers.size());

Set providers = new HashSet(len);

//根据重试次数循环调用

for(int i = 0; i < len; ++i) {

if (i > 0) {

this.checkWhetherDestroyed();

copyinvokers = this.list(invocation);

this.checkInvokers(copyinvokers, invocation);

}

//负载均衡筛选出一个Invoker作本次调用

Invoker invoker = this.select(loadbalance, invocation, copyinvokers, invoked);

//将使用过的Invoker保存起来，下次重试时做过滤用

invoked.add(invoker);

//记录到上下文中

RpcContext.getContext().setInvokers(invoked);

try {

//发起调用

Result result = invoker.invoke(invocation);

if (le != null && logger.iJopieVaHGZsWarnEnabled()) {

logger.warn("....");

}

Result var12 = result;

return var12;

} catch (RpcException var17) { //catch异常 继续下次循环重试

if (var17.isBiz()) {

throw var17;

}

le = var17;

} catch (Throwable var18) {

le = new RpcException(var18.getMessage(), var18);

} finally {

providers.add(invoker.getUrl().getAddress());

}

}

throw new RpcException(....);

}

上述方法中，首先获取重试次数len，根据重试次数进行循环调用，调用发生异常会被catch住，然后重新调用。

每次循环会通过负载均衡选出一个Invoker，然后利用这个Invoker进行远程调用，每次选出的Invoker会记录下来，在下次调用的select()中会将使用上次调用的Invoker进行重试，如果上一次没有调用或者上次调用的Invoker下线了，那么会重新进行负载均衡进行选择。

FailfastClusterInvoker

FailfastClusterInvoker只会进行一次远程调用，如果失败后立马抛出异常。

public Result doInvoke(Invocation invocation, List> invokers, LoadBalance loadbalance) throws RpcException {

this.checkInvokers(invokers, invocation);

Invoker invoker = this.select(loadbalance, invocation, invokers, (List)null); //负载均衡选择Invoker

try {

return invoker.invoke(invocation); //发起远程调用

} catch (Throwable var6) { //失败调用直接将错误抛出

if (var6 instanceof RpcException && ((RpcException)var6).isBiz()) {

throw (RpcException)var6;

} else {

throw new RpcException(....);

}

}

}

FailsafeClusterInvoker

FailsafeClusterInvoker是一种安全失败的cluster，调用发生错误仅仅是记录一下日志，然后就返回了空结果。

public Result doInvoke(Invocation invocation, List> invokers, LoadBalance loadbalance) throws RpcException {

try {

this.checkInvokers(invokers, invocation);

//负载均衡选出Invoker后直接进行调用

Invoker invoker = this.select(loadbalance, invocation, invokers, (List)null);

return invoker.invoke(invocation);

} catch (Throwable var5) { //调用错误只是打印日志

logger.error("Failsafe ignore exception: " + var5.getMessage(), var5);

return new RpcResult();

}

}

FailbackClusterInvoker

FailbackClusterInvoker调用失败后，会记录下本次调用，然后返回一个空结果给服务消费者，并且会通过一个定时任务对失败的调用进行重试。适用于执行消息通知等最大努力场景。

protected Result doInvoke(Invocation invocation, List> invokers, LoadBalance loadbalance) throws RpcException {

try {

this.checkInvokers(invokers, invocation);

//负载均衡选出Invoker

Invoker invoker = this.select(loadbalance, invocation, invokers, (List)null);

//执行调用，执行成功返回调用结果

return invoker.invoke(invocation);

} catch (Throwable var5) {

//调用失败

logger.error("....");

//记录下本次失败调用

this.addFailed(invocation, this);

//返回空结果

return new RpcResult();

}

}

private void addFailed(Invocation invocation, AbstractClusterInvoker> router) {

if (this.retryFuture == null) {

synchronized(this) {

//如果未创建重试本次调用的定时任务

if (this.retryFuture == null) {

//创建定时任务

this.retryFuture = this.scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {

public void run() {

try {

//定时进行重试

FailbackClusterInvoker.this.retryFailed();

} catch (Throwable var2) {

FailbackClusterInvoker.logger.error("....", var2);

}

}

}, 5000L, 5000L, TimeUnit.MILLISECONDS);

}

}

}

//将invocation和router存入map

this.failed.put(invocation, router);

}

void retryFailed() {

if (this.failed.size() != 0) {

Iterator i$ = (new HashMap(this.failed)).entrySet().iterator();

while(i$.hasNext()) {

Entry> entry = (Entry)i$.next();

Invocation invocation = (Invocation)entry.getKey();

Invoker invoker = (Invoker)entry.getValue();

try {

//进行重试调用

invoker.invoke(invocation);

//调用成功未产生异常则移除本次失败调用的记录，销毁定时任务

this.failed.remove(invocation);

} catch (Throwable var6) {

logger.error("....", var6);

}

}

}

}

逻辑比较简单，大致就是当调用错误时返回空结果，并记录下本次失败调用到failed中，并且会创建一个定时任务定时地去调用failed中记录的失败调用，如果调用成功了就从failed中移除这个调用。

ForkingClusterInvoker

ForkingClusterInvoker运行时，会将所有Invoker都放入线程池中并发调用，只要有一个Invoker调用成功了就返回结果，doInvoker方法立即停止运行。

适用于对实时性比较高的读写操作。

public Result doInvoke(final Invocation invocation, List> invokers, LoadBalance loadbalance) throws RpcException {

Result var19;

try {

this.checkInvokers(invokers, invocation);

int forks = this.getUrl().getParameter("forks", 2);

int timeout = this.getUrl().getParameter("timeout", 1000);

final Object selected;

if (forks > 0 && forks < invokers.size()) {

selected = new ArrayList();

for(int i = 0; i < forks; ++i) {

Invoker invoker = this.select(loadbalance, invocation, invokers, (List)selected);

if (!((List)selected).contains(invoker)) {

//选择好的Invoker放入这个selected列表

((List)selected).add(invoker);

}

}

} else {

selected = invokers;

}

RpcContext.getContext().setInvokers((List)selected);

final AtomicInteger count = new AtomicInteger();

//阻塞队列

final BlockingQueue

Iterator i$ = ((List)selected).iterator();

while(i$.hasNext()) {

final Invoker invoker = (Invoker)i$.next();

this.executor.execute(new Runnable() {

public void run() {

try {

Result result = invoker.invoke(invocation);

ref.offer(result);

} catch (Throwable var3) {

int value = count.incrementAndGet();

if (value >= ((List)selected).size()) { //等待所有调用都产生异常才入队

ref.offer(var3);

}

}

}

});

}

try {

//阻塞获取结果

Object ret = ref.poll((long)timeout, TimeUnit.MILLISECONDS);

if (ret instanceof Throwable) {

Throwable e = (Throwable)ret;

throw new RpcException(....);

}

var19 = (Result)ret;

} catch (InterruptedException var14) {

throw new RpcException(....);

}

} finally {

RpcContext.getContext().clearAttachments();

}

return var19;

}

BroadcastClusterInvoker

BroadcastClusterInvoker运行时会将所有Invoker逐个调用，在最后判断中如果有一个调用产生错误，则抛出异常。

适用于通知所有提供者更新缓存或日志等本地资源的场景。

public Result doInvoke(Invocation invocation, List> invokers, LoadBalance loadbalance) throws RpcException {

this.checkInvokers(invokers, invocation);

RpcContext.getContext().setInvokers(invokers);

RpcException exception = null;

Result result = null;

Iterator i$ = invokers.iterator();

while(i$.hasNext()) {

Invoker invoker = (Invoker)i$.next();

try {

result = invoker.invoke(invocation);

} catch (RpcException var9) {

exception = var9;

logger.warn(var9.getMessage(), var9);

} catch (Throwable var10) {

exception = new RpcException(var10.getMessage(), var10);

logger.warn(var10.getMessage(), var10);

}

}

//如果调用过程中发生过错误 抛出异常

if (exception != null) {

throw exception;

} else {

//返回调用结果

return result;

}

}

AbstractClusterInvoker

AbstractClusterInvoker是上述所有类的父类，内部结构较为简单。AvailableCluster内部返回结果就是AvailableClusterInvoker。

public class AvailableCluster implements Cluster {

public static final String NAME = "available";

public AvailableCluster() {

}

public Invoker join(Directory directory) throws RpcException {

return new AbstractClusterInvoker(directory) {

public Result doInvoke(Invocation invocation, List> invokers, LoadBalance loadbalance) throws RpcException {

Iterator i$ = invokers.iterator();

Invoker invoker;

do { //循环判断：哪个invoker能用就调用哪个

if (!i$.hasNext()) {

throw new RpcException("No provider available in " + invokers);

}

invoker = (Invoker)i$.next();

} while(!invoker.isAvailable());

return invoker.invoke(invocation);

}

};

}

}

小结

上述中有很多种集群的实现，各适用于不同的场景，加了Cluster这个中间层，向服务消费者屏蔽了集群调用的细节，并且支持不同场景使用不同的模式。

负载均衡

Dubbo中的负载均衡，即LoadBalance，服务提供者一般都是集群分布，所以需要Dubbo选择出合适的服务提供者来给服务消费者调用。

Dubbo中提供了多种负载均衡算法：

RandomLoadBalanceLeastActiveLoadBalanceConsistentHashLoadBalanceRoundRobinLoadBalance

AbstractLoadBalance

实现类都继承了于这个类，该类实现了LoadBalance，使用模板方法模式，将一些公用的逻辑封装好，而具体的实现由子类自定义。

public Invoker select(List> invokers, URL url, Invocation invocation) {

if (invokers != null && !invokers.isEmpty()) {

//子类实现

return invokers.size() == 1 ? (Invoker)invokers.get(0) : this.doSelect(invokers, url, invocation);

} else {

return null;

}

}

protected abstract Invoker doSelect(List> var1, URL var2, Invocation var3);

服务刚启动需要预热，不能突然让服务负载过高，需要进行服务的降权。

protected int getWeight(Invoker> invoker, Invocation invocation) {

int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), "weight", 100); //获得权重

if (weight > 0) {

long timestamp = invoker.getUrl().getParameter("remote.timestamp", 0L); //启动时间

if (timestamp > 0L) {

int uptime = (int)(System.currentTimeMillis() - timestamp); //计算已启动时长

int warmup = invoker.getUrl().getParameter("warmup", 600000);

if (uptime > 0 && uptime < warmup) {

weight = calculateWarmupWeight(uptime, warmup, weight); //降权

}

}

}

return weight;

}

RandomLoadBalance

使用了加权随机算法，假设现在有三个节点A，B，C，然后赋予这几个节点一定权重，分别为1，2，3，那么可计算得到总权重为6，那么这几个节点被访问的可能性分别为1/6，2/6，3/6。

protected Invoker doSelect(List> invokers, URL url, Invocation invocation) {

int length = invokers.size(); //Invoker个数

int totalWeight = 0; //总权重

boolean sameWeight = true; //权重是否相同

int offset;

int i;

for(offset = 0; offset < length; ++offset) {

i = this.getWeight((Invoker)invokers.get(offset), invocation); //得到权重

totalWeight += i; //计算总权重

//是否权重都相同

if (sameWeight && offset > 0 && i != this.getWeight((Invoker)invokers.get(offset - 1), invocation)) {

sameWeight = false;

}

}

if (totalWeight > 0 && !sameWeight) {

offset = this.random.nextInt(totalWeight); //获得随机偏移量

//判断偏移量落在哪个片段上

for(i = 0; i < length; ++i) {

offset -= this.getWeight((Invoker)invokers.get(i), invocation);

if (offset < 0) {

return (Invoker)invokers.get(i);

}

}

}

return (Invoker)invokers.get(this.random.nextInt(length));

}

LeastActiveLoadBalance

最少活跃数负载均衡，接收一个请求后，请求活跃数+1，处理完一个请求后，请求活跃数-1，请求活跃数少既说明现在服务器压力小也说明该服务器处理请求快，没有堆积什么请求。

总的流程是先遍历Invokers列表，寻找当前请求活跃数最少的Invoker，如果有多个Invoker具有相同的最小请求活跃数，则根据他们的权重来进行筛选。

ConsistentHashLoadBalance

将服务器的IP等信息生成一个Hash值，将这个值映射到Hash圆环上作为某个节点，当查找节点时，通过一个Key来顺时针查找。

Dubbo还引入了160个虚拟节点，使得数据更加分散，避免请求积压在某个节点上。

并且Hash值是方法级别的，一个服务的每个方法都有一个ConsistentHashSelector，根据参数值来计算得出Hash值，

RoundRobinLoadBalance

加权轮询负载均衡，这种轮询是平滑的，假设A和B的权重为10:30，那么轮询的结果可能是A、B、B、A、A、B、B、B…，40次调用下来A调用了10次，B调用了30次。

总结

服务引入时，会将多个远程调用塞入Directory，然后通过Cluster来封装，同时根据需要提供各种容错功能，最终统一暴露一个Invoker给服务消费者，服务消费者调用的时候会从目录得到Invoker列表，经过路由的过滤以及负载均衡最终得到一个Invoker发起调用。

以上就是java开发Dubbo负载均衡与集群容错示例详解的详细内容，如有不足或错误欢迎指正。

更多关于Dubbo负载均衡与集群容错的资料请关注我们其它相关文章！

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。