`
dawuafang
  • 浏览: 1104589 次
文章分类
社区版块
存档分类
最新评论

HBase RegionServer详解(未完)

 
阅读更多

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

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();
  }
<script type="text/javascript"> $(function () { $('pre.prettyprint code').each(function () { var lines = $(this).text().split('\n').length; var $numbering = $('<ul/>').addClass('pre-numbering').hide(); $(this).addClass('has-numbering').parent().append($numbering); for (i = 1; i <= lines; i++) { $numbering.append($('<li/>').text(i)); }; $numbering.fadeIn(1700); }); }); </script>
分享到:
评论

相关推荐

Global site tag (gtag.js) - Google Analytics