Dubbo负载均衡与集群容错 java开发Dubbo负载均衡与集群容错示例详解
又蠢又笨的懒羊羊程序猿 人气:0负载均衡与集群容错
Invoker
在Dubbo中Invoker就是一个具有调用功能的对象,在服务提供端就是实际的服务实现,只是将服务实现封装起来变成一个Invoker。
在服务消费端,从注册中心得到服务提供者的信息之后,将一条条信息封装为Invoker,这个Invoker就具备了远程调用的能力。
综上,Dubbo就是创建了一个统一的模型,将可调用(可执行体)的服务对象都统一封装为Invoker。
而ClusterInvoker
就是将多个服务引入的Invoker封装起来,对外统一暴露一个Invoker,并且赋予这些Invoker集群容错的功能。
服务目录
服务目录,即Directory
,实际上它就是多个Invoker的集合,服务提供端一般都会集群分布,同样的服务会有多个提供者,因此需要一个服务目录来统一存放它们,需要调用服务的时候便从这个服务目录中进行挑选。
同时服务目录还是实现了NotifyListener
接口,当集群中新增了一台服务提供者或者下线了一台服务提供者,目录都会对服务提供者进行更新,新增或者删除对应的Invoker。
从上图中,可以看到用了一个抽象类AbstractDirectory
来实现 Directory
接口,抽象类中运用到了模板方法模式,将一些公共方法和逻辑写好,作为一个骨架,然后具体实现由了两个子类来完成,两个子类分别为StaticDirectory
和RegistryDirectory
。
RegistryDirectory
RegistryDirectory
实现了NotifyListener
接口,可以监听注册中心的变化,当注册中心配置发生变化时,服务目录也可以收到变更通知,然后根据更新之后的配置刷新Invoker列表。
由此可知RegistryDirectory
共有三个作用:
获取Invoker列表监听注册中心刷新Invoker列表
获取Invoker列表
RegistryDirectory
实现了父类AbstractDirectory
的抽象方法doList()
,该方法可以得到Invoker列表
public List<Invoker<T>> doList(Invocation invocation) { if (this.forbidden) { throw new RpcException(....); } else { List<Invoker<T>> invokers = null; Map<String, List<Invoker<T>>> 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<List<Invoker<T>>> 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<URL> urls) { List<URL> invokerUrls = new ArrayList(); List<URL> routerUrls = new ArrayList(); List<URL> 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<URL> 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<String, Invoker<T>> 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<String, Invoker<T>> newUrlInvokerMap = this.toInvokers(invokerUrls); //根据新的InvokerMap生成方法名和Invoker列表对应的Map Map<String, List<Invoker<T>>> 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,并且得到<url,Invoker>的映射关系。
再进一步进行转化,得到<methodName,List>的映射关系,再将同一组的Invoker进行合并,将合并结果赋值给methodInvokerMap
,这个methodInvokerMap
就是在doList
中使用到的Map。
最后刷新InvokerMap
,销毁无效的Invoker。
StaticDirectory
StaticDirectory是静态目录,所有Invoker是固定的不会删减的,并且所有Invoker由构造器来传入。
内部逻辑也相当简单,只定义了一个列表用于存储Invokers。实现父类的方法也只是将这些Invokers原封不动地返回。
private final List<Invoker<T>> invokers; protected List<Invoker<T>> 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 <T> Invoker<T> join(Directory<T> directory) throws RpcException { return new FailoverClusterInvoker(directory); } }
FailoverClusterInvoker
FailoverClusterInvoker
实现的功能是失败调用(有重试次数)自动切换。
public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException { List<Invoker<T>> 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<Invoker<T>> invoked = new ArrayList(invokers.size()); Set<String> 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<T> 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.isWarnEnabled()) { 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<Invoker<T>> 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<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException { try { this.checkInvokers(invokers, invocation); //负载均衡选出Invoker后直接进行调用 Invoker<T> 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<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException { try { this.checkInvokers(invokers, invocation); //负载均衡选出Invoker Invoker<T> 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<Invocation, AbstractClusterInvoker<?>> 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<invocation,router>
中,并且会创建一个定时任务定时地去调用failed
中记录的失败调用,如果调用成功了就从failed
中移除这个调用。
ForkingClusterInvoker
ForkingClusterInvoker
运行时,会将所有Invoker都放入线程池中并发调用,只要有一个Invoker调用成功了就返回结果,doInvoker方法立即停止运行。
适用于对实时性比较高的读写操作。
public Result doInvoke(final Invocation invocation, List<Invoker<T>> 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<T> 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<Object> ref = new LinkedBlockingQueue(); Iterator i$ = ((List)selected).iterator(); while(i$.hasNext()) { final Invoker<T> 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<Invoker<T>> 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 <T> Invoker<T> join(Directory<T> directory) throws RpcException { return new AbstractClusterInvoker<T>(directory) { public Result doInvoke(Invocation invocation, List<Invoker<T>> 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 <T> Invoker<T> select(List<Invoker<T>> 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 <T> Invoker<T> doSelect(List<Invoker<T>> 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 <T> Invoker<T> doSelect(List<Invoker<T>> 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负载均衡与集群容错的资料请关注其它相关文章!
加载全部内容