詳解租約機(jī)制以及在hbase中的應(yīng)用
為什么需要Lease
分布式系統(tǒng)中為什么需要租約機(jī)制，這是因?yàn)樵诜植际较到y(tǒng)，為了保證服務(wù)的高可用，需要在服務(wù)發(fā)生故障的時候及時啟動另外一個服務(wù)實(shí)例以替換故障服務(wù)。這樣就需要在服務(wù)端和客戶端或者服務(wù)端和控制中心維持一個心跳信息，用于服務(wù)進(jìn)程向控制中心匯報(bào)當(dāng)前自己的健康情況，如果控制中心在一段時間收不到服務(wù)進(jìn)程上報(bào)的心跳，則會啟動新的進(jìn)程繼續(xù)對外提供服務(wù)。
但是，由于實(shí)際網(wǎng)絡(luò)情況的復(fù)雜性，控制中心無法收到心跳時不能準(zhǔn)確地判斷究竟是服務(wù)故障了還是服務(wù)進(jìn)程和控制中心之間的網(wǎng)絡(luò)發(fā)生了故障。這種情況下控制中心冒然地啟用新進(jìn)程有可能會造成“雙主”這種情況出現(xiàn)。
為避免上述情況的發(fā)生引入了租約機(jī)制，此時服務(wù)節(jié)點(diǎn)持續(xù)向控制中心申請短時間租約，控制中心在已派發(fā)的租約過期之前，不會啟用新服務(wù)節(jié)點(diǎn)，而服務(wù)節(jié)點(diǎn)租約過期時若還無法從控制中心申請到新租約，自己中斷客戶鏈接。
此外，租約機(jī)制還可用于客戶端和服務(wù)端之間的解藕，避免客戶端進(jìn)程失去響應(yīng)時，其占用的服務(wù)端資源長期得不到釋放進(jìn)而影響到服務(wù)端的穩(wěn)定。
Lease的實(shí)現(xiàn)
在實(shí)際系統(tǒng)中，如果依賴一個中心結(jié)點(diǎn)向外發(fā)布lease存在很大的風(fēng)險(xiǎn)，那就是如果該中心結(jié)點(diǎn)發(fā)生宕機(jī)或者網(wǎng)絡(luò)故障，那么服務(wù)節(jié)點(diǎn)由于接收不到新的租約那么會導(dǎo)致整個服務(wù)集群進(jìn)入不可用狀態(tài)。因此，在實(shí)際使用中，對外提供lease服務(wù)的往往是由多個進(jìn)程實(shí)例組成的另外一套集群，該集群具有高可用性，可以對外提供lease服務(wù)，比如zookeeper集群。
HRegionServer的租約Lease管理
租約線程的初始化
在HRegionServer的run主循環(huán)里會調(diào)用preRegistrationInitialization預(yù)先初始化一些線程，包括初始化集群連接信息setupClusterConnection()、healthCheckChore、pauseMonitor、initializeZookeeper以及initializeThreads()。
其中在initializeThreads()中會初始化各類線程，這些線程包括了這臺regionServer的lease線程：
|
1
2
3
|
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這個抽象類，并定義了如下幾個主要的成員變量：
|
1
2
3
4
5
|
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類都定義了哪些變量：
|
1
2
3
4
|
private final String leaseName; private final LeaseListener listener; private int leaseTimeoutPeriod; private long expirationTime; |
leaseTimeoutPeriod是租約時間，expirationTime會在lease被創(chuàng)建時被置位為系統(tǒng)時間與leaseTimeoutPeriod之和，用于周期性地計(jì)算該租約已經(jīng)被使用多長時間，如果租約已經(jīng)超過了leaseTimeoutPeriod定義的到期時間，則會觸發(fā)一個expired事件，LeaseListener會監(jiān)聽該事件并調(diào)用leaseExpired方法，不同類型的lease都會繼承LeaseListener接口并實(shí)現(xiàn)自己的leaseExpired方法，如下所示是scan lease對該方法的實(shí)現(xiàn)：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
|
@Overridepublic 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請求發(fā)到服務(wù)端的，分解的次數(shù)是scan的總數(shù)據(jù)量與客戶端setCache兩者的比值。每個scan請求發(fā)到服務(wù)端后會租用一個scanner，用于當(dāng)前的scan結(jié)束后，后續(xù)的scan可以直接復(fù)用已有的資源，但是為防止scanner長期占用服務(wù)端資源，通過租約管理，關(guān)閉不再使用的scanner。
OK，回到前面的Leases類，看看它是如何管理regionServer進(jìn)程中的各個lease的，這部分邏輯在它覆寫的run方法中：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
|
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的循環(huán)周期中會逐一便利map中管理的lease，計(jì)算每個lease的thisLeaseDelay以檢查改lease是否已經(jīng)過期。判斷l(xiāng)ease是否過期的方法很簡單，就是取出當(dāng)前時間與lease中定義的expirationTime做差，如果差值小于0，則說明該租約已經(jīng)到期，則調(diào)用lease中定義的leaseExpired方法，這與上面我們講過的關(guān)聯(lián)上了。其中thisLeaseDelay決定了下一次的lease檢查在多久之后發(fā)生，thisLeaseDelay的計(jì)算依據(jù)是選擇選取所有未過期lease中l(wèi)easeDelay最短的，通過thisLeaseDelay計(jì)算toWait時間，用于決定前面的睡眠時間。
感謝閱讀，希望能幫助到大家，謝謝大家對本站的支持！
原文鏈接：http://blog.csdn.net/bryce123phy/article/details/55668199
