zk中ZooKeeperServer的作用是什么

zk中ZooKeeperServer的作用是什么，针对这个问题，这篇文章详细介绍了相对应的分析和解答，希望可以帮助更多想解决这个问题的小伙伴找到更简单易行的方法。

内部类

ChangeRecord 处理PrepRP和FinalRP之间的信息

static class ChangeRecord {

    ChangeRecord(long zxid, String path, StatPersisted stat, int childCount, List<ACL> acl) {
        this.zxid = zxid;
        this.path = path;
        this.stat = stat;
        this.childCount = childCount;
        this.acl = acl;
    }

    long zxid;

    String path;

    StatPersisted stat; /* Make sure to create a new object when changing */

    int childCount;

    List<ACL> acl; /* Make sure to create a new object when changing */

    ChangeRecord duplicate(long zxid) {
        StatPersisted stat = new StatPersisted();
        if (this.stat != null) {
            DataTree.copyStatPersisted(this.stat, stat);
        }
        return new ChangeRecord(zxid, path, stat, childCount, acl == null ? new ArrayList<>() : new ArrayList<>(acl));
    }

}


protected enum State {
    INITIAL,
    RUNNING,
    SHUTDOWN,
    ERROR
}


初始化函数
public ZooKeeperServer(FileTxnSnapLog txnLogFactory, int tickTime, int minSessionTimeout, int maxSessionTimeout, int clientPortListenBacklog, ZKDatabase zkDb, String initialConfig) {
    serverStats = new ServerStats(this);
    this.txnLogFactory = txnLogFactory;
    this.txnLogFactory.setServerStats(this.serverStats);
    this.zkDb = zkDb;
    this.tickTime = tickTime;
    setMinSessionTimeout(minSessionTimeout);
    setMaxSessionTimeout(maxSessionTimeout);
    this.listenBacklog = clientPortListenBacklog;

    listener = new ZooKeeperServerListenerImpl(this);

    readResponseCache = new ResponseCache();

    connThrottle = new BlueThrottle();

    this.initialConfig = initialConfig;

    this.requestPathMetricsCollector = new RequestPathMetricsCollector();

    this.initLargeRequestThrottlingSettings();

    LOG.info("Created server with tickTime " + tickTime
             + " minSessionTimeout " + getMinSessionTimeout()
             + " maxSessionTimeout " + getMaxSessionTimeout()
             + " clientPortListenBacklog " + getClientPortListenBacklog()
             + " datadir " + txnLogFactory.getDataDir()
             + " snapdir " + txnLogFactory.getSnapDir());

}

通过参数构造一个数据管理Log FileTxnSnapLog

public ZooKeeperServer(File snapDir, File logDir, int tickTime) throws IOException {
    this(new FileTxnSnapLog(snapDir, logDir), tickTime, "");
}

集群和单机中加载数据

集群调用顺序

Leader#lead

ZooKeeperServer#loadData

单机调用顺序

ServerCnxFactory#startUp

ZooKeeperServer#startdata

ZooKeeperServer#loadData

单机版的startData方法

public void startdata() throws IOException, InterruptedException {
    //check to see if zkDb is not null
    if (zkDb == null) {
        zkDb = new ZKDatabase(this.txnLogFactory);//实例化zkdatabase
    }
    if (!zkDb.isInitialized()) {
        loadData();//没有初始化就重新初始化
    }
}

//加载数据
public void loadData() throws IOException, InterruptedException {
    /*
     * When a new leader starts executing Leader#lead, it
     * invokes this method. The database, however, has been
     * initialized before running leader election so that
     * the server could pick its zxid for its initial vote.
     * It does it by invoking QuorumPeer#getLastLoggedZxid.
     * Consequently, we don't need to initialize it once more
     * and avoid the penalty of loading it a second time. Not
     * reloading it is particularly important for applications
     * that host a large database.
     *
     * The following if block checks whether the database has
     * been initialized or not. Note that this method is
     * invoked by at least one other method:
     * ZooKeeperServer#startdata
     */
     //加载信息
    if (zkDb.isInitialized()) {
        setZxid(zkDb.getDataTreeLastProcessedZxid());
    } else {
        setZxid(zkDb.loadDataBase());
    }

    // Clean up dead sessions
    //获取超时deadSessions
    List<Long> deadSessions = new ArrayList<>();
    for (Long session : zkDb.getSessions()) {
        if (zkDb.getSessionWithTimeOuts().get(session) == null) {
            deadSessions.add(session);
        }
    }
    //杀掉session
    for (long session : deadSessions) {
        // TODO: Is lastProcessedZxid really the best thing to use?
        killSession(session, zkDb.getDataTreeLastProcessedZxid());
    }

    // Make a clean snapshot 创建快照
    takeSnapshot();
}



删除会话
protected void killSession(long sessionId, long zxid) {
    zkDb.killSession(sessionId, zxid);
    if (LOG.isTraceEnabled()) {
        ZooTrace.logTraceMessage(
            LOG,
            ZooTrace.SESSION_TRACE_MASK,
            "ZooKeeperServer --- killSession: 0x" + Long.toHexString(sessionId));
    }
    if (sessionTracker != null) {
        sessionTracker.removeSession(sessionId);
    }
}




public synchronized void startup() {
    if (sessionTracker == null) {
        createSessionTracker();
    }
    //责任链处理
    startSessionTracker();
    //设置请求处理器
    setupRequestProcessors();

    startRequestThrottler();
    //注册jmx
    registerJMX();

    startJvmPauseMonitor();

    registerMetrics();
    //设置状态
    setState(State.RUNNING);

    requestPathMetricsCollector.start();

    localSessionEnabled = sessionTracker.isLocalSessionsEnabled();
    notifyAll();
}

关于zk中ZooKeeperServer的作用是什么问题的解答就分享到这里了，希望以上内容可以对大家有一定的帮助，如果你还有很多疑惑没有解开，可以关注亿速云行业资讯频道了解更多相关知识。