HBaseRegionServer详解

HBase RegionServer详解

RegionServer组件介绍

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

MemStoreFlusher

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

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

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

  MemStoreFlusher的主要成员变量:

  class MemStoreFlusher implements FlushRequester {
static final Log LOG &＃61; 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 flushQueue &＃61;
new DelayQueue();
private final Map regionsInQueue &＃61;
new HashMap();
private AtomicBoolean wakeupPending &＃61; new AtomicBoolean();
private final long threadWakeFrequency;
private final HRegionServer server;
private final ReentrantReadWriteLock lock &＃61; new ReentrantReadWriteLock();
private final Object blockSignal &＃61; new Object();
protected long globalMemStoreLimit;
protected float globalMemStoreLimitLowMarkPercent;
protected long globalMemStoreLimitLowMark;
private long blockingWaitTime;
private final Counter updatesBlockedMsHighWater &＃61; new Counter();
private final FlushHandler[] flushHandlers;
private List flushRequestListeners &＃61; new ArrayList(1);

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

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

  MemStoreFlusher相关配置项&＃xff1a;

HeapMemoryManager

CompactSplitThread

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

ZooKeeperWatcher

MasterAddressTracker

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

ClusterStatusTracker

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

SplitLogWorker

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

RegionServer启动过程分析

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

public static void main(String[] args) throws Exception {
VersionInfo.logVersion();
Configuration conf &＃61; HBaseConfiguration.create();
&＃64;SuppressWarnings("unchecked")
Class regionServerClass &＃61; (Class) 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 &＃61; this.conf.getInt(HConstants.HEALTH_CHORE_WAKE_FREQ,
HConstants.DEFAULT_THREAD_WAKE_FREQUENCY);
healthCheckChore &＃61; new HealthCheckChore(sleepTime, this, getConfiguration());
}
this.pauseMonitor &＃61; 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服务组件&＃xff0c;主要包括初始化compactSplitThread&＃xff0c;cacheFlusher&＃xff0c;compactionChecker&＃xff0c;以及Leases等。具体代码如下&＃xff1a;

private void initializeThreads() throws IOException {
// Cache flushing thread.
this.cacheFlusher &＃61; new MemStoreFlusher(conf, this);
// Compaction thread
this.compactSplitThread &＃61; 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 &＃61; new CompactionChecker(this, this.threadWakeFrequency, this);
this.periodicFlusher &＃61; new PeriodicMemstoreFlusher(this.threadWakeFrequency, this);
this.leases &＃61; new Leases(this.threadWakeFrequency);
// Create the thread to clean the moved regions list
movedRegionsCleaner &＃61; MovedRegionsCleaner.createAndStart(this);
if (this.nonceManager !&＃61; null) {
// Create the chore that cleans up nonces.
nonceManagerChore &＃61; this.nonceManager.createCleanupChore(this);
}
// Setup RPC client for master communication
rpcClient &＃61; RpcClientFactory.createClient(conf, clusterId, new InetSocketAddress(
rpcServices.isa.getAddress(), 0));
int storefileRefreshPeriod &＃61; conf.getInt(
StorefileRefresherChore.REGIONSERVER_STOREFILE_REFRESH_PERIOD
, StorefileRefresherChore.DEFAULT_REGIONSERVER_STOREFILE_REFRESH_PERIOD);
if (storefileRefreshPeriod > 0) {
this.storefileRefresher &＃61; new StorefileRefresherChore(storefileRefreshPeriod, this, this);
}
registerConfigurationObservers();
}