HBase RegionServer详解

RegionServer组件介绍

RegionServer是HBase集群运行在每个工作节点上的服务。它是整个HBase系统的关键所在，一方面它维护了Region的状态，提供了对于Region的管理和服务；另一方面，它与Master交互，参与Master的分布式协调管理。

MemStoreFlusher

MemStoreFlusher主要功能是将MemStore刷新到文件中，当满足一下条件时会出发MemStore执行flush操作，最小的flush单元是region：

• 当一个MemStore的大小等于hbase.hregion.memstore.flush.size指定大小时，所有属于当前region的memstores都将写入到文件；
• 当MemStore使用内存总量达到hbase.regionserver.global.memstore.upperLimit指定值时，将会有多个MemStores flush到文件中，MemStore flush 顺序是按照大小降序执行的，直到刷新到MemStore使用内存略小于hbase.regionserver.global.memstore.lowerLimit。

• 当每一个region server WAL数量达到hbase.regionserver.max.logs指定的值时，不同的region将按照时间先后顺序flush memstores ,较早的将先被刷新，直到WAL数量低于hbase.regionserver.max.logs为止。

  MemStoreFlusher的主要成员变量:

  class MemStoreFlusher implements FlushRequester { 
  static final Log LOG = LogFactory.getLog(MemStoreFlusher.class); 
   
  // These two data members go together.  Any entry in the one must have 
  // a corresponding entry in the other. 
  private final BlockingQueue<FlushQueueEntry> flushQueue = 
  new DelayQueue<FlushQueueEntry>(); 
  private final Map<HRegion, FlushRegionEntry> regionsInQueue = 
  new HashMap<HRegion, FlushRegionEntry>(); 
  private AtomicBoolean wakeupPending = new AtomicBoolean(); 
   
  private final long threadWakeFrequency; 
  private final HRegionServer server; 
  private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(); 
  private final Object blockSignal = new Object(); 
   
  protected long globalMemStoreLimit; 
  protected float globalMemStoreLimitLowMarkPercent; 
  protected long globalMemStoreLimitLowMark; 
   
  private long blockingWaitTime; 
  private final Counter updatesBlockedMsHighWater = new Counter(); 
   
  private final FlushHandler[] flushHandlers; 
  private List<FlushRequestListener> flushRequestListeners = new ArrayList<FlushRequestListener>(1);

• flushQueue :代表某一个region 的Flush请求，Flusher线程不断地从该队列中获取 请求信息，完成Region的Flush操作；
• regionsInQueue :维护HRegion实例与请求FlushRegionEntry之间的对应关系;某一个FlushQueueEntry实例存在regionsInQueue 中也必然存在于flushQueue中 。

• threadWakeFrequency:用于flushQueue执行poll操作时，最大等待时间，配置项为hbase.server.thread.wakefrequency，默认值10000ms。

  MemStoreFlusher相关配置项：

HeapMemoryManager

CompactSplitThread

合并文件清理不需要的数据，控制Region的规模。在Store内的文件个数超过阈值时，触发Compact合并文件操作，一是清理被删除的数据，二是多余版本的清理。在Region内的Store文件大小超过阈值，会触发Region的Split操作，一个Region被切分成两个Region。这两个操作都是在CompactSplitThread的各自的线程池中被触发。

ZooKeeperWatcher

MasterAddressTracker

捕获Master服务节点的变化。HBase使用多Master来解决Master单点故障的问题，主Master服务故障时，它与ZooKeeper的心跳延迟超过阈值，ZooKeeeper路径下的数据被清理，备Master上的ActiveMaserManager服务会竞争该Master路径，成为主Master。MasterAddresTracker是RS内部监听Master节点变化的追踪器。

ClusterStatusTracker

HBase集群状态追踪器。该选项可以标识当前集群的状态，及它的启动时间。该设置选项有利于集群中的各个工作节点(RS)统一执行启动和退出操作。

SplitLogWorker

基于Region的HLog文件切分器。在RS宕机之后，RS上的保存的HLog文件，需要按照Region进行切分。HMaster会把这些文件作为任务放置到Zookeeper的splitlog路径下，RS上SplitLogWorker会尝试获取任务，对获取到的HLog文件按照Region进行分组，处理的结果保存到相应Region的recovered.edits目录下。

RegionServer启动过程分析

1.HRegionServer main方法
RegionServer是一个独立的服务，有一个mian方法在启动时被调用。mian方法内部调用HRegionServerCommandLine实现RegionServer的启动。
具体代码如下：

  public static void main(String[] args) throws Exception { 
    VersionInfo.logVersion(); 
    Configuration conf = HBaseConfiguration.create(); 
    @SuppressWarnings("unchecked") 
    Class<? extends HRegionServer> regionServerClass = (Class<? extends HRegionServer>) conf 
        .getClass(HConstants.REGION_SERVER_IMPL, HRegionServer.class); 
 
    new HRegionServerCommandLine(regionServerClass).doMain(args); 
  }

1. HRegionServer run方法

2. HRegionServer preRegistrationInitialization方法
   此方法内部主要包括初始化zookeeper相关的服务以及RegionServer服务组件的初始化。

 private void preRegistrationInitialization(){ 
    try { 
      setupClusterConnection(); 
 
      // Health checker thread. 
      if (isHealthCheckerConfigured()) { 
        int sleepTime = this.conf.getInt(HConstants.HEALTH_CHORE_WAKE_FREQ, 
          HConstants.DEFAULT_THREAD_WAKE_FREQUENCY); 
        healthCheckChore = new HealthCheckChore(sleepTime, this, getConfiguration()); 
      } 
      this.pauseMonitor = new JvmPauseMonitor(conf); 
      pauseMonitor.start(); 
 
      initializeZooKeeper(); 
      if (!isStopped() && !isAborted()) { 
        initializeThreads(); 
      } 
    } catch (Throwable t) { 
      // Call stop if error or process will stick around for ever since server 
      // puts up non-daemon threads. 
      this.rpcServices.stop(); 
      abort("Initialization of RS failed.  Hence aborting RS.", t); 
    } 
  }

initializeThreads主要初始化RegionServer服务组件，主要包括初始化compactSplitThread，cacheFlusher，compactionChecker，以及Leases等。具体代码如下：

  private void initializeThreads() throws IOException { 
    // Cache flushing thread. 
    this.cacheFlusher = new MemStoreFlusher(conf, this); 
 
    // Compaction thread 
    this.compactSplitThread = new CompactSplitThread(this); 
 
    // Background thread to check for compactions; needed if region has not gotten updates 
    // in a while. It will take care of not checking too frequently on store-by-store basis. 
    this.compactionChecker = new CompactionChecker(this, this.threadWakeFrequency, this); 
    this.periodicFlusher = new PeriodicMemstoreFlusher(this.threadWakeFrequency, this); 
    this.leases = new Leases(this.threadWakeFrequency); 
 
    // Create the thread to clean the moved regions list 
    movedRegionsCleaner = MovedRegionsCleaner.createAndStart(this); 
 
    if (this.nonceManager != null) { 
      // Create the chore that cleans up nonces. 
      nonceManagerChore = this.nonceManager.createCleanupChore(this); 
    } 
 
    // Setup RPC client for master communication 
    rpcClient = RpcClientFactory.createClient(conf, clusterId, new InetSocketAddress( 
        rpcServices.isa.getAddress(), 0)); 
 
    int storefileRefreshPeriod = conf.getInt( 
        StorefileRefresherChore.REGIONSERVER_STOREFILE_REFRESH_PERIOD 
      , StorefileRefresherChore.DEFAULT_REGIONSERVER_STOREFILE_REFRESH_PERIOD); 
    if (storefileRefreshPeriod > 0) { 
      this.storefileRefresher = new StorefileRefresherChore(storefileRefreshPeriod, this, this); 
    } 
    registerConfigurationObservers(); 
  }