本文只是從總體流程來分析replication過程，很多細(xì)節(jié)沒有提及，下一篇文章準(zhǔn)備多分析分析細(xì)節(jié)。 replicationSource啟動過程 org.apache.hadoop.hbase.regionserver.HRegionServer#startServiceThreads -> org.apache.hadoop.hbase.replication.regionserver.Replication#startReplicationService -> ?//初始化replicationManager org.apache.hadoop.hbase.replication.regionserver.ReplicationSourceManager#init -> //在init階段for循環(huán)把所有的replicationPeers添加到source里，即每個replicationPeer對應(yīng)一個source，也就是可以添加多個slave cluster，replicationPeers從zookeeper /hbase/replication/peers目錄取 org.apache.hadoop.hbase.replication.regionserver.ReplicationSourceManager#addSource -> //在addSource階段生成ReplicationSource并啟動ReplicationSource，ReplicationSource本身是一個線程 org.apache.hadoop.hbase.replication.regionserver.ReplicationSource#startup //ReplicationSource線程啟動，進(jìn)入while循環(huán)工作 replicationSource大致工作流程

while(isAlive())進(jìn)行主體循環(huán)

從WAL文件獲取List<WAL.Entry>

通過調(diào)用shipEdits方法發(fā)送數(shù)據(jù)

調(diào)用replicationEndpoint replicate方法發(fā)送數(shù)據(jù)

最終調(diào)用admin.replicateWALEntry通過rpc發(fā)送數(shù)據(jù)

? regionserver如何從slave cluster中選取regionserver當(dāng)做復(fù)制節(jié)點(diǎn)

replication過程需要連接peer（slave cluster），首先要獲取這個peer所有活著的regionservers

拿到所有regionservers信息之后，開始選擇哪些regionservers作為replication的對象

選哪些regionservers當(dāng)做sink由peer活著的regionserver個數(shù)*ratio（默認(rèn)值0.1）決定，regionservers先shuffle打亂順序后再截取

如果選擇的sink（regionserver）個數(shù)為0，一直等待peer上線，也就是slave cluster沒有啟動的情況

下面源碼可以解釋如何選擇regionserver當(dāng)做sink

private void connectToPeers() {getRegionServers();int sleepMultiplier = 1;// Connect to peer cluster first, unless we have to stopwhile (this.isRunning() && replicationSinkMgr.getSinks().size() == 0) {replicationSinkMgr.chooseSinks();if (this.isRunning() && replicationSinkMgr.getSinks().size() == 0) {if (sleepForRetries("Waiting for peers", sleepMultiplier)) {sleepMultiplier++; //倍數(shù)最多為默認(rèn)配置的300倍，也就是每次sleep最長間隔是300秒 }}}}void chooseSinks() {List<ServerName> slaveAddresses = endpoint.getRegionServers();Collections.shuffle(slaveAddresses, random);int numSinks = (int) Math.ceil(slaveAddresses.size() * ratio);sinks = slaveAddresses.subList(0, numSinks);lastUpdateToPeers = System.currentTimeMillis();badReportCounts.clear();}/*** Do the sleeping logic* @param msg Why we sleep* @param sleepMultiplier by how many times the default sleeping time is augmented* @return True if <code>sleepMultiplier</code> is &lt; <code>maxRetriesMultiplier</code>*/protected boolean sleepForRetries(String msg, int sleepMultiplier) {try {if (LOG.isTraceEnabled()) {LOG.trace(msg + ", sleeping " + sleepForRetries + " times " + sleepMultiplier);}Thread.sleep(this.sleepForRetries * sleepMultiplier);} catch (InterruptedException e) {LOG.debug("Interrupted while sleeping between retries");}return sleepMultiplier < maxRetriesMultiplier;}this.maxRetriesMultiplier = this.conf.getInt("replication.source.maxretriesmultiplier", 300); this.ratio = conf.getFloat("replication.source.ratio", DEFAULT_REPLICATION_SOURCE_RATIO);

?

??? 總結(jié)

每個slave cluster對應(yīng)一個replicationSource線程，各個slave復(fù)制互不干擾

每個replicationSource是單線程進(jìn)行傳輸數(shù)據(jù)，改成多線程并發(fā)傳可能更好

數(shù)據(jù)是通過rpc發(fā)送過去，調(diào)用slave cluster regionserver RSRpcServices的replicateWALEntry方法

?

轉(zhuǎn)載于:https://www.cnblogs.com/yueweimian/p/6520390.html

創(chuàng)作挑戰(zhàn)賽新人創(chuàng)作獎勵來咯，堅(jiān)持創(chuàng)作打卡瓜分現(xiàn)金大獎

總結(jié)

以上是生活随笔為你收集整理的hbase replication原理分析的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。