詳解租約機制以及在hbase中的應用

詳解租約機制以及在hbase中的應用

為什麼需要Lease

分布式系統中為什麼需要租約機制，這是因為在分布式系統，為了保證服務的高可用，需要在服務發生故障的時候及時啟動另外一個服務實例以替換故障服務。這樣就需要在服務端和客戶端或者服務端和控制中心維持一個心跳信息，用於服務進程向控制中心匯報當前自己的健康情況，如果控制中心在一段時間收不到服務進程上報的心跳，則會啟動新的進程繼續對外提供服務。

但是，由於實際網絡情況的復雜性，控制中心無法收到心跳時不能准確地判斷究竟是服務故障了還是服務進程和控制中心之間的網絡發生了故障。這種情況下控制中心冒然地啟用新進程有可能會造成“雙主”這種情況出現。

為避免上述情況的發生引入了租約機制，此時服務節點持續向控制中心申請短時間租約，控制中心在已派發的租約過期之前，不會啟用新服務節點，而服務節點租約過期時若還無法從控制中心申請到新租約，自己中斷客戶鏈接。

此外，租約機制還可用於客戶端和服務端之間的解藕，避免客戶端進程失去響應時，其占用的服務端資源長期得不到釋放進而影響到服務端的穩定。

Lease的實現

在實際系統中，如果依賴一個中心結點向外發布lease存在很大的風險，那就是如果該中心結點發生宕機或者網絡故障，那麼服務節點由於接收不到新的租約那麼會導致整個服務集群進入不可用狀態。因此，在實際使用中，對外提供lease服務的往往是由多個進程實例組成的另外一套集群，該集群具有高可用性，可以對外提供lease服務，比如zookeeper集群。

HRegionServer的租約Lease管理

租約線程的初始化

在HRegionServer的run主循環裡會調用preRegistrationInitialization預先初始化一些線程，包括初始化集群連接信息setupClusterConnection()、healthCheckChore、pauseMonitor、initializeZookeeper以及initializeThreads()。

其中在initializeThreads()中會初始化各類線程，這些線程包括了這台regionServer的lease線程：

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

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

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負責管理該regionserver進程中的lease實例，我們看看lease類都定義了哪些變量：

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

 leaseTimeoutPeriod是租約時間，expirationTime會在lease被創建時被置位為系統時間與leaseTimeoutPeriod之和，用於周期性地計算該租約已經被使用多長時間，如果租約已經超過了leaseTimeoutPeriod定義的到期時間，則會觸發一個expired事件，LeaseListener會監聽該事件並調用leaseExpired方法，不同類型的lease都會繼承LeaseListener接口並實現自己的leaseExpired方法，如下所示是scan lease對該方法的實現：

@Override 
public void leaseExpired() {    //處理租約過期 
  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請求是分解成多次RPC請求發到服務端的，分解的次數是scan的總數據量與客戶端setCache兩者的比值。每個scan請求發到服務端後會租用一個scanner，用於當前的scan結束後，後續的scan可以直接復用已有的資源，但是為防止scanner長期占用服務端資源，通過租約管理，關閉不再使用的scanner。

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

public void run() { 
  long toWait = leaseCheckFrequency; 
  Lease nextLease = null; 
  long nextLeaseDelay = Long.MAX_VALUE; 
 
  while (!stopRequested || (stopRequested && !leases.isEmpty()) ) { 
 
   //睡眠一段時間 
 
   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的循環周期中會逐一便利map中管理的lease，計算每個lease的thisLeaseDelay以檢查改lease是否已經過期。判斷lease是否過期的方法很簡單，就是取出當前時間與lease中定義的expirationTime做差，如果差值小於0，則說明該租約已經到期，則調用lease中定義的leaseExpired方法，這與上面我們講過的關聯上了。其中thisLeaseDelay決定了下一次的lease檢查在多久之後發生，thisLeaseDelay的計算依據是選擇選取所有未過期lease中leaseDelay最短的，通過thisLeaseDelay計算toWait時間，用於決定前面的睡眠時間。

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