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

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

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; 

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

@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!"); 
  } 
 } 

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(); 
 }