詳解租約機(jī)制以及在hbase中的應(yīng)用

為什么需要Lease

分布式系統(tǒng)中為什么需要租約機(jī)制，這是因?yàn)樵诜植际较到y(tǒng)，為了保證服務(wù)的高可用，需要在服務(wù)發(fā)生故障的時(shí)候及時(shí)啟動(dòng)另外一個(gè)服務(wù)實(shí)例以替換故障服務(wù)。這樣就需要在服務(wù)端和客戶端或者服務(wù)端和控制中心維持一個(gè)心跳信息，用于服務(wù)進(jìn)程向控制中心匯報(bào)當(dāng)前自己的健康情況，如果控制中心在一段時(shí)間收不到服務(wù)進(jìn)程上報(bào)的心跳，則會(huì)啟動(dòng)新的進(jìn)程繼續(xù)對(duì)外提供服務(wù)。

但是，由于實(shí)際網(wǎng)絡(luò)情況的復(fù)雜性，控制中心無(wú)法收到心跳時(shí)不能準(zhǔn)確地判斷究竟是服務(wù)故障了還是服務(wù)進(jìn)程和控制中心之間的網(wǎng)絡(luò)發(fā)生了故障。這種情況下控制中心冒然地啟用新進(jìn)程有可能會(huì)造成“雙主”這種情況出現(xiàn)。

為避免上述情況的發(fā)生引入了租約機(jī)制，此時(shí)服務(wù)節(jié)點(diǎn)持續(xù)向控制中心申請(qǐng)短時(shí)間租約，控制中心在已派發(fā)的租約過(guò)期之前，不會(huì)啟用新服務(wù)節(jié)點(diǎn)，而服務(wù)節(jié)點(diǎn)租約過(guò)期時(shí)若還無(wú)法從控制中心申請(qǐng)到新租約，自己中斷客戶鏈接。

此外，租約機(jī)制還可用于客戶端和服務(wù)端之間的解藕，避免客戶端進(jìn)程失去響應(yīng)時(shí)，其占用的服務(wù)端資源長(zhǎng)期得不到釋放進(jìn)而影響到服務(wù)端的穩(wěn)定。

Lease的實(shí)現(xiàn)

在實(shí)際系統(tǒng)中，如果依賴一個(gè)中心結(jié)點(diǎn)向外發(fā)布lease存在很大的風(fēng)險(xiǎn)，那就是如果該中心結(jié)點(diǎn)發(fā)生宕機(jī)或者網(wǎng)絡(luò)故障，那么服務(wù)節(jié)點(diǎn)由于接收不到新的租約那么會(huì)導(dǎo)致整個(gè)服務(wù)集群進(jìn)入不可用狀態(tài)。因此，在實(shí)際使用中，對(duì)外提供lease服務(wù)的往往是由多個(gè)進(jìn)程實(shí)例組成的另外一套集群，該集群具有高可用性，可以對(duì)外提供lease服務(wù)，比如zookeeper集群。

HRegionServer的租約Lease管理

租約線程的初始化

在HRegionServer的run主循環(huán)里會(huì)調(diào)用preRegistrationInitialization預(yù)先初始化一些線程，包括初始化集群連接信息setupClusterConnection()、healthCheckChore、pauseMonitor、initializeZookeeper以及initializeThreads()。

其中在initializeThreads()中會(huì)初始化各類線程，這些線程包括了這臺(tái)regionServer的lease線程：

this.compactionChecker = new CompactionChecker(this, this.threadWakeFrequency, this); //檢查合并請(qǐng)求 
this.periodicFlusher = new PeriodicMemstoreFlusher(this.threadWakeFrequency, this);  //周期性地檢查memstore的flush請(qǐng)求 
this.leases = new Leases(this.threadWakeFrequency); 

 Leases類的定義如下，它繼承了HasThread這個(gè)抽象類，并定義了如下幾個(gè)主要的成員變量：

public static final int MIN_WAIT_TIME = 100; 
private final Map<String, Lease> leases = new ConcurrentHashMap<String, Lease>(); 
 
protected final int leaseCheckFrequency; 
protected volatile boolean stopRequested = false; 

 其中Map型成員變量leases負(fù)責(zé)管理該regionserver進(jìn)程中的lease實(shí)例，我們看看lease類都定義了哪些變量：

private final String leaseName; 
private final LeaseListener listener; 
private int leaseTimeoutPeriod; 
private long expirationTime; 

 leaseTimeoutPeriod是租約時(shí)間，expirationTime會(huì)在lease被創(chuàng)建時(shí)被置位為系統(tǒng)時(shí)間與leaseTimeoutPeriod之和，用于周期性地計(jì)算該租約已經(jīng)被使用多長(zhǎng)時(shí)間，如果租約已經(jīng)超過(guò)了leaseTimeoutPeriod定義的到期時(shí)間，則會(huì)觸發(fā)一個(gè)expired事件，LeaseListener會(huì)監(jiān)聽(tīng)該事件并調(diào)用leaseExpired方法，不同類型的lease都會(huì)繼承LeaseListener接口并實(shí)現(xiàn)自己的leaseExpired方法，如下所示是scan lease對(duì)該方法的實(shí)現(xiàn)：

@Override 
public void leaseExpired() {    //處理租約過(guò)期 
  RegionScannerHolder rsh = scanners.remove(this.scannerName); 
  if (rsh != null) { 
   RegionScanner s = rsh.s; 
   LOG.info("Scanner " + this.scannerName + " lease expired on region " 
     + s.getRegionInfo().getRegionNameAsString()); 
   try { 
    Region region = regionServer.getRegion(s.getRegionInfo().getRegionName()); 
    if (region != null && region.getCoprocessorHost() != null) { 
     region.getCoprocessorHost().preScannerClose(s); 
    } 
 
    s.close(); 
    if (region != null && region.getCoprocessorHost() != null) { 
     region.getCoprocessorHost().postScannerClose(s); 
    } 
   } catch (IOException e) { 
    LOG.error("Closing scanner for " 
     + s.getRegionInfo().getRegionNameAsString(), e); 
   } 
  } else { 
   LOG.warn("Scanner " + this.scannerName + " lease expired, but no related" + 
    " scanner found, hence no chance to close that related scanner!"); 
  } 
 } 

 客戶端的scan請(qǐng)求是分解成多次RPC請(qǐng)求發(fā)到服務(wù)端的，分解的次數(shù)是scan的總數(shù)據(jù)量與客戶端setCache兩者的比值。每個(gè)scan請(qǐng)求發(fā)到服務(wù)端后會(huì)租用一個(gè)scanner，用于當(dāng)前的scan結(jié)束后，后續(xù)的scan可以直接復(fù)用已有的資源，但是為防止scanner長(zhǎng)期占用服務(wù)端資源，通過(guò)租約管理，關(guān)閉不再使用的scanner。

OK，回到前面的Leases類，看看它是如何管理regionServer進(jìn)程中的各個(gè)lease的，這部分邏輯在它覆寫的run方法中：

public void run() { 
  long toWait = leaseCheckFrequency; 
  Lease nextLease = null; 
  long nextLeaseDelay = Long.MAX_VALUE; 
 
  while (!stopRequested || (stopRequested && !leases.isEmpty()) ) { 
 
   //睡眠一段時(shí)間 
 
   nextLease = null; 
   nextLeaseDelay = Long.MAX_VALUE; 
   for (Iterator<Map.Entry<String, Lease>> it = leases.entrySet().iterator(); it.hasNext();) { 
    Map.Entry<String, Lease> entry = it.next(); 
    Lease lease = entry.getValue(); 
    long thisLeaseDelay = lease.getDelay(TimeUnit.MILLISECONDS); 
    if ( thisLeaseDelay > 0) { 
     if (nextLease == null || thisLeaseDelay < nextLeaseDelay) { 
      nextLease = lease; 
      nextLeaseDelay = thisLeaseDelay; 
     } 
    } else { 
     // A lease expired. Run the expired code before removing from map 
     // since its presence in map is used to see if lease exists still. 
     if (lease.getListener() == null) { 
      LOG.error("lease listener is null for lease " + lease.getLeaseName()); 
     } else { 
      lease.getListener().leaseExpired(); 
     } 
     it.remove(); 
    } 
   } 
  } 
  close(); 
 } 

 我們省略掉一些異常處理，在while的循環(huán)周期中會(huì)逐一便利map中管理的lease，計(jì)算每個(gè)lease的thisLeaseDelay以檢查改lease是否已經(jīng)過(guò)期。判斷l(xiāng)ease是否過(guò)期的方法很簡(jiǎn)單，就是取出當(dāng)前時(shí)間與lease中定義的expirationTime做差，如果差值小于0，則說(shuō)明該租約已經(jīng)到期，則調(diào)用lease中定義的leaseExpired方法，這與上面我們講過(guò)的關(guān)聯(lián)上了。其中thisLeaseDelay決定了下一次的lease檢查在多久之后發(fā)生，thisLeaseDelay的計(jì)算依據(jù)是選擇選取所有未過(guò)期lease中l(wèi)easeDelay最短的，通過(guò)thisLeaseDelay計(jì)算toWait時(shí)間，用于決定前面的睡眠時(shí)間。

感謝閱讀，希望能幫助到大家，謝謝大家對(duì)本站的支持！

最新評(píng)論