1 服務器端整體概覽圖

概覽圖

ServerCnxnFactory：負責與client之間的網絡交互，支持NIO(默認)以及Netty

SessionTrackerImpl：會話管理器

DatadirCleanupManager：定期清理存在磁盤上的log文件和snapshot文件

PreRequestProcessor，SyncRequestProcessor，FinalRequestProcessor：請求處理流程，責任鏈模式

LearnerHandler：Leader與Learner之間的交互

FileTxnSnapLog：存儲在磁盤上的日志文件

DataTree：體現在內存中的存儲結構

Sessions：Session的相關信息存儲

2 單機版服務器啟動流程

執行QuorumPeerMain的main方法，其中先創建一個QuorumPeerMain對象

調用initializeAndRun方法

protected void initializeAndRun(String[] args)

throws ConfigException, IOException

{

QuorumPeerConfig config = new QuorumPeerConfig();

if (args.length == 1) {

config.parse(args[0]);

}

// Start and schedule the the purge task

DatadirCleanupManager purgeMgr = new DatadirCleanupManager(config

.getDataDir(), config.getDataLogDir(), config

.getSnapRetainCount(), config.getPurgeInterval());

purgeMgr.start();

if (args.length == 1 && config.servers.size() > 0) {

runFromConfig(config);

} else {

LOG.warn("Either no config or no quorum defined in config, running "

+ " in standalone mode");

// there is only server in the quorum -- run as standalone

ZooKeeperServerMain.main(args);

}

}

2.1

args實際上就是zoo.cfg中的配置，如下

2.2

創建DatadirCleanupManager實例，參數有snapDir，dataLogDir，snapRetainCount(要保存snapshot文件的個數)，purgeInterval(定期清理的頻率，單位為小時)，snapRetainCount與purgeInterval在zoo.cfg中均可以配置

調用DatadirCleanupManager的start方法，里面主要依賴PurgeTask，這也是一個線程，其run方法

PurgeTxnLog的purge方法：

public static void purge(File dataDir, File snapDir, int num) throws IOException {

// snapshot文件保存的數量小于3,拋異常

if (num < 3) {

throw new IllegalArgumentException(COUNT_ERR_MSG);

}

// 根據dataDir和snapDir創建FileTxnSnapLog

FileTxnSnapLog txnLog = new FileTxnSnapLog(dataDir, snapDir);

// 根據給定數量獲取最近的文件

List snaps = txnLog.findNRecentSnapshots(num);

// 獲取數目

int numSnaps = snaps.size();

if (numSnaps > 0) {

// 第二個參數是最近的snapshot文件

purgeOlderSnapshots(txnLog, snaps.get(numSnaps - 1));

}

}

找最近n個snapshot文件：

public List findNRecentSnapshots(int n) throws IOException {

List files = Util.sortDataDir(snapDir.listFiles(), SNAPSHOT_FILE_PREFIX, false);

int count = 0;

List list = new ArrayList();

for (File f: files) {

if (count == n)

break;

if (Util.getZxidFromName(f.getName(), SNAPSHOT_FILE_PREFIX) != -1) {

count++;

list.add(f);

}

}

return list;

}

purgeOlderSnapshots:

static void purgeOlderSnapshots(FileTxnSnapLog txnLog, File snapShot) {

// 從snapshot文件名中獲取zxid

final long leastZxidToBeRetain = Util.getZxidFromName(

snapShot.getName(), PREFIX_SNAPSHOT);

/**

* 我們刪除名稱中帶有zxid且小于leastZxidToBeRetain的所有文件。

* 該規則適用于快照文件和日志文件，

* zxid小于X的日志文件可能包含zxid大于X的事務。

* 更準確的說,命名為log.(X-a)的log文件可能包含比snapshot.X文件更新的事務,

* 如果在該間隔中沒有其他以zxid開頭即(X-a，X]的日志文件

*/

final Set retainedTxnLogs = new HashSet();

//獲取快照日志，其中可能包含比給定zxid更新的事務,這些日志需要保留下來

retainedTxnLogs.addAll(Arrays.asList(txnLog.getSnapshotLogs(leastZxidToBeRetain)));

/**

* Finds all candidates for deletion, which are files with a zxid in their name that is less

* than leastZxidToBeRetain. There's an exception to this rule, as noted above.

*/

class MyFileFilter implements FileFilter{

private final String prefix;

MyFileFilter(String prefix){

this.prefix=prefix;

}

public boolean accept(File f){

if(!f.getName().startsWith(prefix + "."))

return false;

if (retainedTxnLogs.contains(f)) {

return false;

}

long fZxid = Util.getZxidFromName(f.getName(), prefix);

if (fZxid >= leastZxidToBeRetain) {

return false;

}

return true;

}

}

// add all non-excluded log files

List files = new ArrayList();

File[] fileArray = txnLog.getDataDir().listFiles(new MyFileFilter(PREFIX_LOG));

if (fileArray != null) {

files.addAll(Arrays.asList(fileArray));

}

// add all non-excluded snapshot files to the deletion list

fileArray = txnLog.getSnapDir().listFiles(new MyFileFilter(PREFIX_SNAPSHOT));

if (fileArray != null) {

files.addAll(Arrays.asList(fileArray));

}

// remove the old files

for(File f: files)

{

final String msg = "Removing file: "+

DateFormat.getDateTimeInstance().format(f.lastModified())+

"\t"+f.getPath();

LOG.info(msg);

System.out.println(msg);

if(!f.delete()){

System.err.println("Failed to remove "+f.getPath());

}

}

}

先獲取到那些需要保留的文件，之后再去刪除這些不在保留文件之內的文件。

2.3

判斷集群是單機啟動還是集群啟動，集群走runFromConfig(config)，單機走ZooKeeperServerMain.main(args)(其實單機版最終走的是ZooKeeperServerMain的runFromConfig，)

runFromConfig方法：

public void runFromConfig(ServerConfig config) throws IOException {

LOG.info("Starting server");

FileTxnSnapLog txnLog = null;

try {

final ZooKeeperServer zkServer = new ZooKeeperServer();

// Registers shutdown handler which will be used to know the

// server error or shutdown state changes.

final CountDownLatch shutdownLatch = new CountDownLatch(1);

zkServer.registerServerShutdownHandler(

new ZooKeeperServerShutdownHandler(shutdownLatch));

// 構建FileTxnSnapLog對象

txnLog = new FileTxnSnapLog(new File(config.dataLogDir), new File(

config.dataDir));

zkServer.setTxnLogFactory(txnLog);

zkServer.setTickTime(config.tickTime);

zkServer.setMinSessionTimeout(config.minSessionTimeout);

zkServer.setMaxSessionTimeout(config.maxSessionTimeout);

// 構建與client之間的網絡通信服務組件

// 這里可以通過zookeeper.serverCnxnFactory配置NIO還是Netty

cnxnFactory = ServerCnxnFactory.createFactory();

cnxnFactory.configure(config.getClientPortAddress(),

config.getMaxClientCnxns());

cnxnFactory.startup(zkServer);

// Watch status of ZooKeeper server. It will do a graceful shutdown

// if the server is not running or hits an internal error.

shutdownLatch.await();

shutdown();

cnxnFactory.join();

if (zkServer.canShutdown()) {

zkServer.shutdown(true);

}

} catch (InterruptedException e) {

// warn, but generally this is ok

LOG.warn("Server interrupted", e);

} finally {

if (txnLog != null) {

txnLog.close();

}

}

}

初始化網絡服務組件后，cnxnFactory.startup(zkServer);

這里以默認網絡服務組件為例NIOServerCnxnFactory

@Override

public void start() {

// ensure thread is started once and only once

if (thread.getState() == Thread.State.NEW) {

thread.start();

}

}

@Override

public void startup(ZooKeeperServer zks) throws IOException,

InterruptedException {

//調用上面的start方法,實際調用thread的start

//也就調用了該類的run方法,啟動網絡服務

start();

//這是ZooKeeperServer

setZooKeeperServer(zks);

zks.startdata();

zks.startup();

}

2.4

zks.startdata()方法：

public void startdata()

throws IOException, InterruptedException {

//check to see if zkDb is not null

if (zkDb == null) {

zkDb = new ZKDatabase(this.txnLogFactory);

}

if (!zkDb.isInitialized()) {

//loadData進行初始化

loadData();

}

}

ZKDatabase在內存中維護了zookeeper的sessions， datatree和commit logs集合。 當zookeeper server啟動的時候會將txnlogs和snapshots從磁盤讀取到內存中

ZKDatabase的loadData()方法：

//如果zkDb已經初始化,設置zxid(內存當中DataTree最新的zxid)

if(zkDb.isInitialized()){

setZxid(zkDb.getDataTreeLastProcessedZxid());

}

else {

// 沒有初始化,就loadDataBase

setZxid(zkDb.loadDataBase());

}

public long loadDataBase() throws IOException {

long zxid = snapLog.restore(dataTree, sessionsWithTimeouts, commitProposalPlaybackListener);

initialized = true;

return zxid;

}

loadDataBase()內部調用的是FileTxnSnapLog的restore方法

2.5

zks.startup()方法：

public synchronized void startup() {

if (sessionTracker == null) {

// 創建會話管理器

createSessionTracker();

}

// 啟動會話管理器

startSessionTracker();

// 設置請求處理器

setupRequestProcessors();

// 注冊jmx

registerJMX();

setState(State.RUNNING);

notifyAll();

}

總結

以上是生活随笔為你收集整理的服务器启动文件,[Zookeeper] 服务端之单机版服务器启动的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。