详解租约机制以及在hbase中的应用 - 行业资讯 - 肥雀云

<强>详解租约机制以及在hbase中的应用

为什么需要租赁
　　

分布式系统中为什么需要租约机制,这是因为在分布式系统,为了保证服务的高可用,需要在服务发生故障的时候及时启动另外一个服务实例以替换故障服务。这样就需要在服务端和客户端或者服务端和控制中心维持一个心跳信息,用于服务进程向控制中心汇报当前自己的健康情况,如果控制中心在一段时间收不到服务进程上报的心跳,则会启动新的进程继续对外提供服务。
　　

但是,由于实际网络情况的复杂性,控制中心无法收到心跳时不能准确地判断究竟是服务故障了还是服务进程和控制中心之间的网络发生了故障。这种情况下控制中心冒然地启用新进程有可能会造成“双主”这种情况出现。
　　

为避免上述情况的发生引入了租约机制,此时服务节点持续向控制中心申请短时间租约,控制中心在已派发的租约过期之前,不会启用新服务节点,而服务节点租约过期时若还无法从控制中心申请到新租约,自己中断客户链接。
　　

此外,租约机制还可用于客户端和服务端之间的解藕,避免客户端进程失去响应时,其占用的服务端资源长期得不到释放进而影响到服务端的稳定。
　　

<强>租赁的实现
　　

在实际系统中,如果依赖一个中心结点向外发布租赁存在很大的风险,那就是如果该中心结点发生宕机或者网络故障,那么服务节点由于接收不到新的租约那么会导致整个服务集群进入不可用状态,因此,在实际使用中,对外提供租赁服务的往往是由多个进程实例组成的另外一套集群,该集群具有高可用性,可以对外提供租赁服务,比如饲养员集群。
　　

HRegionServer的租约租赁管理
　　

租约线程的初始化
　　

在HRegionServer的运行主循环里会调用preRegistrationInitialization预先初始化一些线程,包括初始化集群连接信息setupClusterConnection (), healthCheckChore, pauseMonitor, initializeZookeeper以及initializeThreads ()。
　　

其中在initializeThreads()中会初始化各类线程,这些线程包括了这台regionServer的租赁线程:
　　

　　　　　　this.compactionChecker=new CompactionChecker(这个,这个。threadWakeFrequency,);//检查合并请求　　这一点。periodicFlusher=new PeriodicMemstoreFlusher(这一点。threadWakeFrequency,);//周期性地检查memstore的冲水请求　　这一点。出租=new租赁(this.threadWakeFrequency);之前　　　　

,租赁类的定义如下,它继承了HasThread这个抽象类,并定义了如下几个主要的成员变量:
　　

　　　　　　公共静态最终int MIN_WAIT_TIME=100; 　　私人最终Map<字符串,Lease>出租=new ConcurrentHashMap<字符串,Lease> (); 　　　　保护最终int leaseCheckFrequency; 　　保护不稳定的布尔stopRequested=false; 　　之前　　　　

,其中映射型成员变量租赁负责管理该regionserver进程中的租赁实例,我们看看租赁类都定义了哪些变量:
　　

　　　　　　私人leaseName最终字符串; 　　私人最终LeaseListener侦听器; 　　私人int leaseTimeoutPeriod; 　　私人长expirationTime; 　　之前　　　　

, leaseTimeoutPeriod是租约时间,expirationTime会在租赁被创建时被置位为系统时间与leaseTimeoutPeriod之和,用于周期性地计算该租约已经被使用多长时间,如果租约已经超过了leaseTimeoutPeriod定义的到期时间,则会触发一个过期的事件,LeaseListener会监听该事件并调用leaseExpired方法,不同类型的租赁都会继承LeaseListener接口并实现自己的leaseExpired方法,如下所示是扫描租赁对该方法的实现:

　　　　　　@Override 　　公共空间leaseExpired(){//处理租约过期　　RegionScannerHolder rsh=scanners.remove (this.scannerName); 　　如果(rsh !=null) { 　　RegionScanner s=rsh.s; 　　LOG.info(“扫描仪”+。租约到期scannerName + "> 　　公共空间run () { 　　长等=leaseCheckFrequency; 　　租赁nextLease=零; 　　长nextLeaseDelay=Long.MAX_VALUE; 　　　　而(!stopRequested | | (stopRequested,,! leases.isEmpty ())) {//睡眠一段时间　　　　nextLease=零; 　　nextLeaseDelay=Long.MAX_VALUE; 　　(Iterator比;它=leases.entrySet () .iterator ();it.hasNext ();) { 　　Lease> Map.Entry<字符串;输入=it.next (); 　　租赁租赁=entry.getValue (); 　　长thisLeaseDelay=lease.getDelay (TimeUnit.MILLISECONDS); 　　如果(thisLeaseDelay比;0){ 　　如果(nextLease==null | | thisLeaseDelay & lt;nextLeaseDelay) { 　　nextLease=租赁; 　　nextLeaseDelay=thisLeaseDelay; 　　} 　　其他}{//租约过期了。运行代码之前到期删除从地图//自其在地图是用来看看租赁仍然存在。　　如果(lease.getListener ()==null) { 　　日志。错误(“租赁租赁侦听器是null”+ lease.getLeaseName ()); 　　其他}{ 　　.leaseExpired lease.getListener () (); 　　} 　　拔除(); 　　} 　　} 　　} 　　close (); 　　} 　　