HBASE REGIONSERVER启动过程

HBASEHRegionServer启动分析

regionserver的启动入口是HRegionServer.main方法

生成HRegionServercommandLine实例,并执行doMain方法,

此方法中通过ToolRunner.run去调用HRegionServercommandLine.run方法

a.判断传入参数是start还是stop，如果是start,调用HRegionServercommandLine.start()方法

b.调用HRegionServer(Configuration)生成regionserver实例,请点这里

c.通过Thread的start去调用regionserver.run方法,请点这里

publicstaticvoidmain(String[]args)throws Exception {

VersionInfo.logVersion();

Configurationconf= HBaseConfiguration.create();

@SuppressWarnings("unchecked")

Class<?extendsHRegionServer>regionServerClass= (Class<?extendsHRegionServer>)conf

.getClass(HConstants.REGION_SERVER_IMPL,HRegionServer.class);

 

newHRegionServerCommandLine(regionServerClass).doMain(args);

}

 

 

生成HRegionServer实例

1.通过hbase.regionserver.codecs配置regionserver的压缩

2.0检查通过hbase对hdfs进行本地读取时，是否需要检验，

dfs.client.read.shortcircuit.skip.checksum,默认为false

2.1通过hbase.regionserver.checksum.verify来配置regionserver读取到数据后是否检验

publicHRegionServer(Configurationconf)

throwsIOException, InterruptedException {

this.fsOk=true;

this.conf=conf;

this.isOnline=false;//默认情况下,rs是非在线状态

.........此处省去一些代码

//读取client的最大重试次数hbase.client.retries.number,default=31

this.numRetries=this.conf.getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,

Hconstants.DEFAULT_HBASE_CLIENT_RETRIES_NUMBER);

得到hbase.server.thread.wakefrequency的值,默认为10*1000ms,

检查memstore是否超过hbase.hregion.memstore.flush.size设置的flush大小的时间间隔

this.threadWakeFrequency=conf.getInt(HConstants.THREAD_WAKE_FREQUENCY,10 * 1000);

定时向master发送此rs的报告的间隔时间,默认为3s=3000ms

this.msgInterval=conf.getInt("hbase.regionserver.msginterval",3 * 1000);

 

this.sleeper=newSleeper(this.msgInterval,this);

通过hbase.client.scanner.max.result.size配置client的最大响应字节数,默认为long.maxvalue,也就是不设置

this.maxScannerResultSize=conf.getLong(

HConstants.HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE_KEY,

HConstants.DEFAULT_HBASE_CLIENT_SCANNER_MAX_RESULT_SIZE);

配置向master进行region的region个数,

this.numRegionsToReport=conf.getInt(

"hbase.regionserver.numregionstoreport",10);

通过hbase.rpc.shortoperation.timeout配置rpc的超时时间,默认为10000ms=10s

this.rpcTimeout=conf.getInt(

HConstants.HBASE_RPC_SHORTOPERATION_TIMEOUT_KEY,

HConstants.DEFAULT_HBASE_RPC_SHORTOPERATION_TIMEOUT);

 

this.abortRequested=false;

this.stopped=false;

通过hbase.client.scanner.timeout.period配置client的scan租约到期时间,默认为60000ms=60s

老版本通过hbase.regionserver.lease.period配置

this.scannerLeaseTimeoutPeriod= HBaseConfiguration.getInt(conf,

HConstants.HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD,

HConstants.HBASE_REGIONSERVER_LEASE_PERIOD_KEY,

HConstants.DEFAULT_HBASE_CLIENT_SCANNER_TIMEOUT_PERIOD);

 

//Server to handle client requests.

Stringhostname=conf.get("hbase.regionserver.ipc.address",

.........此处省去DNS解析代码

得到通过hbase.regionserver.port配置的rs的rpc端口,默认为60020

intport =conf.getInt(HConstants.REGIONSERVER_PORT,

HConstants.DEFAULT_REGIONSERVER_PORT);

//Creation of a HSA will force a resolve.

InetSocketAddressinitialIsa=newInetSocketAddress(hostname,port);

.........此处省去一些判断的代码

this.rand=newRandom(initialIsa.hashCode());

Stringname="regionserver/"+initialIsa.toString();

//Set how many times to retry talking to another server overHconnection.

设置Hconnection的执行重试次数,

hbase.client.retries.number*hbase.client.serverside.retries.multiplier,default=31*10

HConnectionManager.setServerSideHConnectionRetries(this.conf,name,LOG);

生成rpcserver，通过hostname与port，读取rpchandler的线程数

this.rpcServer=newRpcServer(this,name,getServices(),

/*HBaseRPCErrorHandler.class,OnlineRegions.class},*/

initialIsa,// BindAddress is IP we got for thisserver.

conf.getInt("hbase.regionserver.handler.count",10),

conf.getInt("hbase.regionserver.metahandler.count",10),

conf,HConstants.QOS_THRESHOLD);

 

//Set our address.

this.isa=this.rpcServer.getListenerAddress();

 

this.rpcServer.setErrorHandler(this);

this.rpcServer.setQosFunction((qosFunction=newQosFunction(this)));

this.startcode= System.currentTimeMillis();

.........此处省去一些zklogin and hbase rs login的代码

 

生成用来记录此rs中所有的memstore所占大小的实例

regionServerAccounting=newRegionServerAccounting();

针对hfile操作的cache配置实例

cacheConfig=newCacheConfig(conf);

uncaughtExceptionHandler=newUncaughtExceptionHandler(){

@Override

publicvoiduncaughtException(Threadt,Throwablee){

abort("Uncaughtexception in service thread " +t.getName(),e);

}

};

是否启用分布式日志重播,通过hbase.master.distributed.log.replay进行配置,默认为false

this.distributedLogReplay=this.conf.getBoolean(HConstants.DISTRIBUTED_LOG_REPLAY_KEY,

HConstants.DEFAULT_DISTRIBUTED_LOG_REPLAY_CONFIG);

}

 

 

执行HRegionServer.run方法

Run方法作用在regionserver的启动

publicvoidrun(){

try{

//Dopre-registrationinitializations;zookeeper,lease threads, etc.

初始化rs与zk的连接,初始化rs中的各种线程,与hmaster创建连接

a.调用initializeZooKeeper,请点这里

b.调用initializeThreads,请点这里

preRegistrationInitialization();

}catch(Throwablee){

abort("Fatalexception during initialization",e);

}

 

try{

//Try and register with the Master; tell it we are here. Break if

//server is stopped or theclusterupflag is down orhdfswentwacky.

如果当前rs的stoping状态为false，同时cluster(zk中的running)已经注册成功

while(keepLooping()){

向master进行注册,并得到master的响应,如果注册失败一直等待,直到注册成功

通过masterAddressManager在zk中拿到master的注册地址,

创建rs与master通信的rpc连接,

(master实现接口)RegionServerStatusProtos.RegionServerStatusService.BlockingInterface

(rs接口实现类)RegionServerStatusProtos.RegionServerStatusService.BlockingStub

生成向master请求的request(当前rs的端口,启动时间(startcode),当前时间)

通过rpc的client端实现BlockingStub调用regionServerStartup向master注册rs节点

master与rs通信请查看master与rs注册与心跳分析

RegionServerStartupResponsew=reportForDuty();

if(w == null){

LOG.warn("reportForDutyfailed; sleeping and then retrying.");

this.sleeper.sleep();

}else{

向master注册成功,向zk中的rs路径注册此rs节点,

注册的节点为zk的EPHEMERAL类型的节点,超时时间通过zookeeper.session.timeout配置,默认为180000

EPHEMERAL注册的zk节点,只要rs中的session挂掉,数据就自动删除,

zksession中会有一个线程定期与zk进行ping

master向rs响应的response有一个map实例，此实例中的响应conf会更新rs中原来的conf中具体key对应的值

更新mapred.task.id配置为hb_rs_servername

生成FileSystem/Hlog实例,设置rs为online

调用startServiceThreads启动相关线程,具体请查看HRegionServer中的相关代码

handleReportForDutyResponse(w);

break;

}

}

 

//Initialize the RegionServerCoprocessorHost now that our ephemeral

//node was created by reportForDuty, in case anycoprocessorswant

//to use ZooKeeper

加载通过hbase.coprocessor.regionserver.classes配置的全局cp

this.rsHost=newRegionServerCoprocessorHost(this,this.conf);

 

if(!this.stopped&&isHealthy()){

//start the snapshot handler, since the server is ready to run

启动快照管理器

this.snapshotManager.start();

}

 

//We registered with the Master. Go into run mode.

longlastMsg =0;

longoldRequestCount= -1;

//The main run loop.

迭代执行,只要当前rs节点还没有被停止,同时节点健康状态正常,向master发送心跳报告

while(!this.stopped&&isHealthy()){

if(!isClusterUp()){

........此处省去一些如果cluster不在线,关闭regions的代码

}

longnow =System.currentTimeMillis();

if((now -lastMsg) >=msgInterval){

向master发送心跳,请查看master与rs注册与心跳分析

tryRegionServerReport(lastMsg,now);

lastMsg= System.currentTimeMillis();

}

if(!this.stopped)this.sleeper.sleep();

}//for

}catch(Throwablet){

if(!checkOOME(t)){

........此处省去一些错误处理的代码

}

}

//Run shutdown.

.........此处省去一些rs停止的代码,包括停止rpc/webserver,启动的线程,删除zk上的节点

}

 

initializeZooKeeper方法分析

privatevoidinitializeZooKeeper()throwsIOException, InterruptedException {

//Open connection tozookeeperand set primary watcher

创建当前rs与zk的连接

this.zooKeeper=newZooKeeperWatcher(conf,REGIONSERVER+":"+

this.isa.getPort(),this);

 

创建master的跟踪器MasterAddressTracker,等待master的启动(在zk上注册)

this.masterAddressManager=newMasterAddressTracker(this.zooKeeper,this);

this.masterAddressManager.start();

blockAndCheckIfStopped(this.masterAddressManager);

 

创建cluster的跟踪器,等待cluster的启动,也就是master注册clusterid到zk后,表示集群已经启动

this.clusterStatusTracker=newClusterStatusTracker(this.zooKeeper,this);

this.clusterStatusTracker.start();

blockAndCheckIfStopped(this.clusterStatusTracker);

 

//Create the catalog tracker and start it;

创建metaregion是否在zk上注册的跟踪器,等待meta在zk上注册,通过assignMeta分配metaregion时注册

this.catalogTracker=newCatalogTracker(this.zooKeeper,this.conf,this);

catalogTracker.start();

 

//Retrieve clusterId

//Since cluster status is now up

//ID should have already been set by HMaster

try{

clusterId= ZKClusterId.readClusterIdZNode(this.zooKeeper);

if(clusterId==null){

this.abort("ClusterID has not been set");

}

LOG.info("ClusterId: "+clusterId);

}catch(KeeperExceptione) {

this.abort("Failedto retrieve Cluster ID",e);

}

.........此处省去一些Snapshot管理器的生成与TableLockManager的生成代码

 

//register watcher for recovering regions

如果配置有分布式日志重播,生成分布式日志重播监听,(recovering-regions)路径

if(this.distributedLogReplay){

this.recoveringRegionWatcher=newRecoveringRegionWatcher(this.zooKeeper,this);

}

}

InitializeThreads方法分析

privatevoidinitializeThreads()throwsIOException {

//Cache flushing thread.

生成memstore的刷新线程,

a.得到hbase.server.thread.wakefrequency的值,默认为10*1000ms,

检查memstore是否超过hbase.hregion.memstore.flush.size设置的flush大小的时间间隔

从cacheFlusher的队列中取出一个regionflush request的最大等待时间(可参见java的BlockingQueue,DelayQueue)

b.通过mbean得到最在的可用内存

c.通过hbase.regionserver.global.memstore.upperLimit得到memstore的最大内存使用限制的比例,

默认为0.4,并通过最大内存算出限制的内存大小

d.通过hbase.regionserver.global.memstore.lowerLimit得到memstore超过使用限制的内存后,

强制flush到限制的内存内,默认值为0.35,

也就是flush后的memstore占用的内存在max*0.35内就停止强制flush

e.通过hbase.hstore.blockingWaitTime配置一个等待时间，默认值90000ms,作用：

store中storefile的个数超过hbase.hstore.blockingStoreFiles配置的文件个数时,

在flush时发现store的个数超限时,把此region的flush等待hbase.hstore.blockingWaitTime/100

flushQueue(DelayQueue)通过超时时间进行队列的排序

f.通过hbase.hstore.flusher.count配置得到一个

MemStoreFlusher中真正用来处理flush操作的FlushHandler,默认值为1

 

this.cacheFlusher=newMemStoreFlusher(conf,this);

 

//Compaction thread

生成对storefile进行compact与split(compact后)的操作的线程

a.得到通过hbase.regionserver.regionSplitLimit配置的值,默认为integer.maxvalue,

表示如果当前rs中的region操作这个数后,不在进行split操作,设置为1可以不做split

b.得到通过hbase.regionserver.thread.compaction.large配置的

large(大)compact操作的线程个数,默认为1,

此值主要检查当前需要要compact的storefile大小,如果超过如下配置,表示需要large否则使用small

如果大小超过hbase.regionserver.thread.compaction.throttle配置的值

throttle的默认值为2*hbase.hstore.compaction.max(default=10)

*hbase.hregion.memstore.flush.size(default=128m)

c.得到通过hbase.regionserver.thread.compaction.small配置的small(小)compact线程个数,默认为1

c.1可以通过设置throttle的大小来控制large的发生,减少large的线程个数,加大small的线程个数

d.通过hbase.regionserver.thread.split配置split的线程个数,默认值为1

e.得到通过hbase.regionserver.thread.merge配置的major的线程个数,默认为1

this.compactSplitThread=newCompactSplitThread(this);

 

//Background thread to check forcompactions;needed if region has not gotten updates

//in a while. It will take care of not checking too frequently onstore-by-store basis.

生成通过hbase.server.thread.wakefrequency(10*1000ms)配置的定期检查region是否需要compact的检查线程,

如果需要进行compact,会在此处通过compact的线程触发compcat的请求

此实例中通过hbase.server.thread.wakefrequency(10*1000ms)配置majorcompact的优先级,

如果majorcompact的优先级大过此值,把compact的优先级设置为此值.

Store中通过hbase.server.compactchecker.interval.multiplier配置多少时间需要进行compact检查的间隔

默认为1000ms,如果multiplier

compactionChecker的检查周期为wakefrequency*multiplier

a.compaction检查时发起compact的条件是

如果一个store中所有的file个数减去在做(或发起compact请求)的个数，大于或等于

hbase.hstore.compaction.min配置的值,

老版本使用hbase.hstore.compactionThreshold进行配置,默认值为3

b.majorcompact的条件检查

通过hbase.hregion.majorcompaction配置major的检查周期,default=1000*60*60*24

通过hbase.hregion.majorcompaction.jitter配置major的浮动时间,默认为0.2,

也就是major的时间上下浮动4.8小时

b2.检查(当前时间-major配置时间>store最小的文件生成时间)表示需要major,

b2.1>store下是否只有一个文件,同时这个文件已经到了major的时间,

b2.1>检查ttl时间是否达到(intager.max表示没配置),达到ttl时间需要major,否则不做

b2.2>文件个数大于1,到达major的时间,需要major

this.compactionChecker=newCompactionChecker(this,this.threadWakeFrequency,this);

生成通过hbase.server.thread.wakefrequency(10*1000ms)配置定期检查region

是否有过期没有被flush过的region,如果有,触发此region的flush，

region的flush过期时间通过hbase.regionserver.optionalcacheflushinterval进行配置,

默认为3600000(一小时)，

this.periodicFlusher=newPeriodicMemstoreFlusher(this.threadWakeFrequency,this);

//Health checker thread.

Rs节点健康情况检查的线程,定期对rs进行健康检查,需要要配置自定义脚本

a.通过hbase.node.health.script.frequency配置检查的间隔时间,

默认为hbase.server.thread.wakefrequency(10*1000ms)

b.通过hbase.node.health.script.location配置节点检查的脚本路径,

c.通过hbase.node.health.script.timeout配置脚本执行的超时时间,默认为60000ms

d.通过配置检查的可控制失败次数(再一个连续的时间内),如果超过此值,停止rs

intsleepTime =this.conf.getInt(HConstants.HEALTH_CHORE_WAKE_FREQ,

HConstants.DEFAULT_THREAD_WAKE_FREQUENCY);

if(isHealthCheckerConfigured()){

healthCheckChore=newHealthCheckChore(sleepTime,this,getConfiguration());

}

RS中租约到期检查,检查间隔通过hbase.server.thread.wakefrequency(10*1000ms)配置

this.leases=newLeases(this.threadWakeFrequency);

 

//Create the thread to clean the moved regions list

定期120000ms检查一次在rs中的movedRegions中是否有移出时间在120000ms之前的region,

如果有直接从movedRegions中remove掉(region被移动到别的rs中)

movedRegionsCleaner= MovedRegionsCleaner.createAndStart(this);

 

//Setup RPC client for master communication

生成与master通信的rpcclient连接,通过MasterAddressManager跟踪zk中master注册的master地址

a.通过hbase.ipc.client.connection.maxidletime配置连接的最大等待时间,默认为10000ms

b.通过hbase.ipc.client.connect.max.retries配置连接的重试次数,默认为0

c.通过hbase.client.pause配置每次重试的间隔时间,默认为100ms

d.通过hbase.ipc.client.tcpnodelay配置tcp连接是否没有延时,默认为true

e.通过hbase.ipc.client.tcpkeepalive配置是否保持client连接,默认为true

f.通过ipc.ping.interval配置masterping的间隔时间,默认为60000ms=1分钟

g.通过hbase.ipc.client.failed.servers.expiry配置连接失败的master过期时间默认为2000ms,

过期后可以重新去连接此server,如果server连接失败会把server添加到一个列表中,

等待此server在列表中的时间超过配置的过期时间时,从列表称出,此时创建连接可以重新去连接此server

h.通过hbase.client.ipc.pool.size配置一个rpcClient中connection的连接池大小,默认为1

j.通过hbase.client.ipc.pool.type配置rpcClient连接池类型,

可配置Reusable(复用),ThreadLocal(本地线程),RoundRobin(轮循);,

默认为RoundRobin

k.如果hbase.security.authentication配置的认证为kerberos,

那么hbase.ipc.client.fallback-to-simple-auth-allowed值需要要配置为true,默认为false

---以下几个配置在IPCUtil中使用---

l.hbase.ipc.cellblock.decompression.buffersize.multiplier

配置对CallScanner中call(kv，已经压缩过)读取解压时buf相对压缩读取的buf的倍数,默认为3,

如读取内容为16kb,那么解压的buf为16*1024*3

z.hbase.ipc.cellblock.building.initial.buffersize

配置对CallScanner中的Call(kv)进行压缩的bufsize大小，默认为16k(16*1024)

o.通过hbase.client.rpc.compressor配置hbase使用的压缩算法,

rpcClient=newRpcClient(conf,clusterId,newInetSocketAddress(

this.isa.getAddress(),0));

this.pauseMonitor=newJvmPauseMonitor(conf);

pauseMonitor.start();

}

RS与Master通信

HMBASE的REGION分配

日志重播分析

数据写入/split/compact/flush分析

数据读取:get/scan