Dubbo是Alibaba开源的分布式服务框架,咱们能够很是容易地经过Dubbo来构建分布式服务,并根据本身实际业务应用场景来选择合适的集群容错模式,这个对于不少应用都是迫切但愿的,只须要经过简单的配置就可以实现分布式服务调用,也就是说服务提供方(Provider)发布的服务能够自然就是集群服务,好比,在实时性要求很高的应用场景下,可能但愿来自消费方(Consumer)的调用响应时间最短,只须要选择Dubbo的Forking Cluster模式配置,就能够对一个调用请求并行发送到多台对等的提供方(Provider)服务所在的节点上,只选择最快一个返回响应的,而后将调用结果返回给服务消费方(Consumer),显然这种方式是以冗余服务为基础的,须要消耗更多的资源,可是可以知足高实时应用的需求。
有关Dubbo服务框架的简单使用,能够参考个人其余两篇文章(《基于Dubbo的Hessian协议实现远程调用》,《Dubbo实现RPC调用使用入门》,后面参考连接中已给出连接),这里主要围绕Dubbo分布式服务相关配置的使用来讲明与实践。java
Dubbo服务集群容错redis
假设咱们使用的是单机模式的Dubbo服务,若是在服务提供方(Provider)发布服务之后,服务消费方(Consumer)发出一次调用请求,刚好此次因为网络问题调用失败,那么咱们能够配置服务消费方重试策略,可能消费方第二次重试调用是成功的(重试策略只须要配置便可,重试过程是透明的);可是,若是服务提供方发布服务所在的节点发生故障,那么消费方再怎么重试调用都是失败的,因此咱们须要采用集群容错模式,这样若是单个服务节点因故障没法提供服务,还能够根据配置的集群容错模式,调用其余可用的服务节点,这就提升了服务的可用性。
首先,根据Dubbo文档,咱们引用文档提供的一个架构图以及各组件关系说明,以下所示:

上述各个组件之间的关系(引自Dubbo文档)说明以下:算法
- 这里的Invoker是Provider的一个可调用Service的抽象,Invoker封装了Provider地址及Service接口信息。
- Directory表明多个Invoker,能够把它当作List,但与List不一样的是,它的值多是动态变化的,好比注册中心推送变动。
- Cluster将Directory中的多个Invoker假装成一个Invoker,对上层透明,假装过程包含了容错逻辑,调用失败后,重试另外一个。
- Router负责从多个Invoker中按路由规则选出子集,好比读写分离,应用隔离等。
- LoadBalance负责从多个Invoker中选出具体的一个用于本次调用,选的过程包含了负载均衡算法,调用失败后,须要重选。
咱们也简单说明目前Dubbo支持的集群容错模式,每种模式适应特定的应用场景,能够根据实际须要进行选择。Dubbo内置支持以下6种集群模式:spring
配置值为failover。这种模式是Dubbo集群容错默认的模式选择,调用失败时,会自动切换,从新尝试调用其余节点上可用的服务。对于一些幂等性操做可使用该模式,如读操做,由于每次调用的反作用是相同的,因此能够选择自动切换并重试调用,对调用者彻底透明。能够看到,若是重试调用必然会带来响应端的延迟,若是出现大量的重试调用,可能说明咱们的服务提供方发布的服务有问题,如网络延迟严重、硬件设备须要升级、程序算法很是耗时,等等,这就须要仔细检测排查了。
例如,能够这样显式指定Failover模式,或者不配置则默认开启Failover模式,配置示例以下:apache
1 |
< dubbo:service interface = "org.shirdrn.dubbo.api.ChatRoomOnlineUserCounterService" version = "1.0.0" |
2 |
cluster = "failover" retries = "2" timeout = "100" ref = "chatRoomOnlineUserCounterService" protocol = "dubbo" > |
3 |
< dubbo:method name = "queryRoomUserCount" timeout = "80" retries = "2" /> |
上述配置使用Failover Cluster模式,若是调用失败一次,能够再次重试2次调用,服务级别调用超时时间为100ms,调用方法queryRoomUserCount的超时时间为80ms,容许重试2次,最坏状况调用花费时间160ms。若是该服务接口org.shirdrn.dubbo.api.ChatRoomOnlineUserCounterService还有其余的方法可供调用,则其余方法没有显式配置则会继承使用dubbo:service配置的属性值。api
配置值为failfast。这种模式称为快速失败模式,调用只执行一次,失败则当即报错。这种模式适用于非幂等性操做,每次调用的反作用是不一样的,如写操做,好比交易系统咱们要下订单,若是一次失败就应该让它失败,一般由服务消费方控制是否从新发起下订单操做请求(另外一个新的订单)。缓存
配置值为failsafe。失败安全模式,若是调用失败, 则直接忽略失败的调用,而是要记录下失败的调用到日志文件,以便后续审计。安全
配置值为failback。失败自动恢复,后台记录失败请求,定时重发。一般用于消息通知操做。服务器
配置值为forking。并行调用多个服务器,只要一个成功即返回。一般用于实时性要求较高的读操做,但须要浪费更多服务资源。网络
配置值为broadcast。广播调用全部提供者,逐个调用,任意一台报错则报错(2.1.0开始支持)。一般用于通知全部提供者更新缓存或日志等本地资源信息。
上面的6种模式均可以应用于生产环境,咱们能够根据实际应用场景选择合适的集群容错模式。若是咱们以为Dubbo内置提供的几种集群容错模式都不能知足应用须要,也能够定制实现本身的集群容错模式,由于Dubbo框架给我提供的扩展的接口,只须要实现接口com.alibaba.dubbo.rpc.cluster.Cluster便可,接口定义以下所示:
01 |
@SPI (FailoverCluster.NAME) |
02 |
public interface Cluster { |
05 |
* Merge the directory invokers to a virtual invoker. |
08 |
* @return cluster invoker |
09 |
* @throws RpcException |
12 |
<T> Invoker<T> join(Directory<T> directory) throws RpcException; |
关于如何实现一个自定义的集群容错模式,能够参考Dubbo源码中内置支持的汲取你容错模式的实现,6种模式对应的实现类以下所示:
1 |
com.alibaba.dubbo.rpc.cluster.support.FailoverCluster |
2 |
com.alibaba.dubbo.rpc.cluster.support.FailfastCluster |
3 |
com.alibaba.dubbo.rpc.cluster.support.FailsafeCluster |
4 |
com.alibaba.dubbo.rpc.cluster.support.FailbackCluster |
5 |
com.alibaba.dubbo.rpc.cluster.support.ForkingCluster |
6 |
com.alibaba.dubbo.rpc.cluster.support.AvailableCluster |
可能咱们初次接触Dubbo时,不知道如何在实际开发过程当中使用Dubbo的集群模式,后面咱们会以Failover Cluster模式为例开发咱们的分布式应用,再进行详细的介绍。
Dubbo服务负载均衡
Dubbo框架内置提供负载均衡的功能以及扩展接口,咱们能够透明地扩展一个服务或服务集群,根据须要很是容易地增长/移除节点,提升服务的可伸缩性。Dubbo框架内置提供了4种负载均衡策略,以下所示:
- Random LoadBalance:随机策略,配置值为random。能够设置权重,有利于充分利用服务器的资源,高配的能够设置权重大一些,低配的能够稍微小一些
- RoundRobin LoadBalance:轮询策略,配置值为roundrobin。
- LeastActive LoadBalance:配置值为leastactive。根据请求调用的次数计数,处理请求更慢的节点会受到更少的请求
- ConsistentHash LoadBalance:一致性Hash策略,具体配置方法能够参考Dubbo文档。相同调用参数的请求会发送到同一个服务提供方节点上,若是某个节点发生故障没法提供服务,则会基于一致性Hash算法映射到虚拟节点上(其余服务提供方)
在实际使用中,只须要选择合适的负载均衡策略值,配置便可,下面是上述四种负载均衡策略配置的示例:
1 |
< dubbo:service interface = "org.shirdrn.dubbo.api.ChatRoomOnlineUserCounterService" version = "1.0.0" |
2 |
cluster = "failover" retries = "2" timeout = "100" loadbalance = "random" |
3 |
ref = "chatRoomOnlineUserCounterService" protocol = "dubbo" > |
4 |
< dubbo:method name = "queryRoomUserCount" timeout = "80" retries = "2" loadbalance = "leastactive" /> |
上述配置,也体现了Dubbo配置的继承性特色,也就是dubbo:service元素配置了loadbalance=”random”,则该元素的子元素dubbo:method若是没有指定负载均衡策略,则默认为loadbalance=”random”,不然若是dubbo:method指定了loadbalance=”leastactive”,则使用子元素配置的负载均衡策略覆盖了父元素指定的策略(这里调用queryRoomUserCount方法使用leastactive负载均衡策略)。
固然,Dubbo框架也提供了实现自定义负载均衡策略的接口,能够实现com.alibaba.dubbo.rpc.cluster.LoadBalance接口,接口定义以下所示:
02 |
* LoadBalance. (SPI, Singleton, ThreadSafe) |
06 |
* @see com.alibaba.dubbo.rpc.cluster.Cluster#join(Directory) |
08 |
* @author william.liangf |
10 |
@SPI (RandomLoadBalance.NAME) |
11 |
public interface LoadBalance { |
14 |
* select one invoker in list. |
15 |
* @param invokers invokers. |
16 |
* @param url refer url |
17 |
* @param invocation invocation. |
18 |
* @return selected invoker. |
20 |
@Adaptive ( "loadbalance" ) |
21 |
<T> Invoker<T> select(List<Invoker<T>> invokers, URL url, Invocation invocation) throws RpcException; |
如何实现一个自定义负载均衡策略,能够参考Dubbo框架内置的实现,以下所示的3个实现类:
1 |
com.alibaba.dubbo.rpc.cluster.loadbalance.RandomLoadBalance |
2 |
com.alibaba.dubbo.rpc.cluster.loadbalance.RoundRobinLoadBalance |
3 |
com.alibaba.dubbo.rpc.cluster.loadbalance.LeastActiveLoadBalance |
Dubbo服务集群容错实践
手机应用是以聊天室为基础的,咱们须要收集用户的操做行为,而后计算聊天室中在线人数,并实时在手机应用端显示人数,整个系统的架构如图所示:

上图中,主要包括了两大主要流程:日志收集并实时处理流程、调用读取实时计算结果流程,咱们使用基于Dubbo框架开发的服务来提供实时计算结果读取聊天人数的功能。上图中,实际上业务接口服务器集群也能够基于Dubbo框架构建服务,就看咱们想要构建什么样的系统来知足咱们的须要。
若是不使用注册中心,服务消费方也可以直接调用服务提供方发布的服务,这样须要服务提供方将服务地址暴露给服务消费方,并且也没法使用监控中心的功能,这种方式成为直连。
若是咱们使用注册中心,服务提供方将服务发布到注册中心,而服务消费方能够经过注册中心订阅服务,接收服务提供方服务变动通知,这种方式能够隐藏服务提供方的细节,包括服务器地址等敏感信息,而服务消费方只能经过注册中心来获取到已注册的提供方服务,而不能直接跨过注册中心与服务提供方直接链接。这种方式的好处是还可使用监控中心服务,可以对服务的调用状况进行监控分析,还能使用Dubbo服务管理中心,方便管理服务,咱们在这里使用的是这种方式,也推荐使用这种方式。使用注册中心的Dubbo分布式服务相关组件结构,以下图所示:

下面,开发部署咱们的应用,经过以下4个步骤来完成:
服务接口将服务提供方(Provider)和服务消费方(Consumer)链接起来,服务提供方实现接口中定义的服务,即给出服务的实现,而服务消费方负责调用服务。咱们接口中给出了2个方法,一个是实时查询获取当前聊天室内人数,另外一个是查询一天中某个/某些聊天室中在线人数峰值,接口定义以下所示:
01 |
package org.shirdrn.dubbo.api; |
03 |
import java.util.List; |
05 |
public interface ChatRoomOnlineUserCounterService { |
07 |
String queryRoomUserCount(String rooms); |
09 |
List<String> getMaxOnlineUserCount(List<String> rooms, String date, String dateFormat); |
接口是服务提供方和服务消费方公共遵照的协议,通常状况下是服务提供方将接口定义好后提供给服务消费方。
服务提供方实现接口中定义的服务,其实现和普通的服务没什么区别,咱们的实现类为ChatRoomOnlineUserCounterServiceImpl,代码以下所示:
01 |
package org.shirdrn.dubbo.provider.service; |
03 |
import java.util.List; |
05 |
import org.apache.commons.logging.Log; |
06 |
import org.apache.commons.logging.LogFactory; |
07 |
import org.shirdrn.dubbo.api.ChatRoomOnlineUserCounterService; |
08 |
import org.shirdrn.dubbo.common.utils.DateTimeUtils; |
10 |
import redis.clients.jedis.Jedis; |
11 |
import redis.clients.jedis.JedisPool; |
13 |
import com.alibaba.dubbo.common.utils.StringUtils; |
14 |
import com.google.common.base.Strings; |
15 |
import com.google.common.collect.Lists; |
17 |
public class ChatRoomOnlineUserCounterServiceImpl implements ChatRoomOnlineUserCounterService { |
19 |
private static final Log LOG = LogFactory.getLog(ChatRoomOnlineUserCounterServiceImpl. class ); |
20 |
private JedisPool jedisPool; |
21 |
private static final String KEY_USER_COUNT = "chat::room::play::user::cnt" ; |
22 |
private static final String KEY_MAX_USER_COUNT_PREFIX = "chat::room::max::user::cnt::" ; |
23 |
private static final String DF_YYYYMMDD = "yyyyMMdd" ; |
25 |
public String queryRoomUserCount(String rooms) { |
26 |
LOG.info( "Params[Server|Recv|REQ] rooms=" + rooms); |
27 |
StringBuffer builder = new StringBuffer(); |
28 |
if (!Strings.isNullOrEmpty(rooms)) { |
31 |
jedis = jedisPool.getResource(); |
32 |
String[] fields = rooms.split( "," ); |
33 |
List<String> results = jedis.hmget(KEY_USER_COUNT, fields); |
34 |
builder.append(StringUtils.join(results, "," )); |
35 |
} catch (Exception e) { |
43 |
LOG.info( "Result[Server|Recv|RES] " + builder.toString()); |
44 |
return builder.toString(); |
48 |
public List<String> getMaxOnlineUserCount(List<String> rooms, String date, String dateFormat) { |
50 |
LOG.info( "Params[Server|Recv|REQ] rooms=" + rooms + ",date=" + date + ",dateFormat=" + dateFormat); |
51 |
String whichDate = DateTimeUtils.format(date, dateFormat, DF_YYYYMMDD); |
52 |
String key = KEY_MAX_USER_COUNT_PREFIX + whichDate; |
53 |
StringBuffer builder = new StringBuffer(); |
54 |
if (rooms != null && !rooms.isEmpty()) { |
57 |
jedis = jedisPool.getResource(); |
58 |
return jedis.hmget(key, rooms.toArray( new String[rooms.size()])); |
59 |
} catch (Exception e) { |
67 |
LOG.info( "Result[Server|Recv|RES] " + builder.toString()); |
68 |
return Lists.newArrayList(); |
71 |
public void setJedisPool(JedisPool jedisPool) { |
72 |
this .jedisPool = jedisPool; |
代码中经过读取Redis中数据来完成调用,逻辑比较简单。对应的Maven POM依赖配置,以下所示:
03 |
< groupId >org.shirdrn.dubbo</ groupId > |
04 |
< artifactId >dubbo-api</ artifactId > |
05 |
< version >0.0.1-SNAPSHOT</ version > |
08 |
< groupId >org.shirdrn.dubbo</ groupId > |
09 |
< artifactId >dubbo-commons</ artifactId > |
10 |
< version >0.0.1-SNAPSHOT</ version > |
13 |
< groupId >redis.clients</ groupId > |
14 |
< artifactId >jedis</ artifactId > |
15 |
< version >2.5.2</ version > |
18 |
< groupId >org.apache.commons</ groupId > |
19 |
< artifactId >commons-pool2</ artifactId > |
20 |
< version >2.2</ version > |
23 |
< groupId >org.jboss.netty</ groupId > |
24 |
< artifactId >netty</ artifactId > |
25 |
< version >3.2.7.Final</ version > |
有关对Dubbo框架的一些依赖,咱们单独放到一个通用的Maven Module中(详见后面“附录:Dubbo使用Maven构建依赖配置”),这里再也不多说。服务提供方实现,最关键的就是服务的配置,由于Dubbo基于Spring来管理配置和实例,因此经过配置能够指定服务是不是分布式服务,以及经过配置增长不少其它特性。咱们的配置文件为provider-cluster.xml,内容以下所示:
01 |
<? xml version = "1.0" encoding = "UTF-8" ?> |