深入理解 ZooKeeper单机服务端的启动（二）

如下的ZooKeeperServer相关的属性
private FileTxnSnapLog txnLogFactory = null;
private ZKDatabase zkDb;
protected RequestProcessor firstProcessor
以及它可以构建DataTree

public void runFromConfig(ServerConfig config) throws IOException {
    LOG.info("Starting server");
    FileTxnSnapLog txnLog = null;
    try {
        // Note that this thread isn't going to be doing anything else,
        // so rather than spawning another thread, we will just call run() in this thread.
        // todo 请注意,当前线程不会做其他任何事情,因此我们只在当前线程中调用Run方法,而不是开启新线程
        // create a file logger url from the command line args
        // todo 根据命令中的args 创建一个logger文件
        final ZooKeeperServer zkServer = new ZooKeeperServer();
        // Registers shutdown handler which will be used to know the server error or shutdown state changes.
        // todo  注册一个shutdown handler, 通过他了解server发生的error或者了解shutdown 状态的更改
        final CountDownLatch shutdownLatch = new CountDownLatch(1);
        zkServer.registerServerShutdownHandler(
                new ZooKeeperServerShutdownHandler(shutdownLatch));
        // todo FileTxnSnapLog工具类, 与 文件快照相关
        txnLog = new FileTxnSnapLog(new File(config.dataLogDir), new File(config.dataDir));
        txnLog.setServerStats(zkServer.serverStats());
        zkServer.setTxnLogFactory(txnLog);
        zkServer.setTickTime(config.tickTime);
        zkServer.setMinSessionTimeout(config.minSessionTimeout);
        zkServer.setMaxSessionTimeout(config.maxSessionTimeout);
        // todo 创建Server上下文的工厂，工厂方法模式
        // todo ServerCnxnFactory是个抽象类,他有不同是实现, NIO版本的 Netty版本的
        cnxnFactory = ServerCnxnFactory.createFactory();
        // todo 建立socket,默认是NIOServerCnxnFactory（是一个线程）
        cnxnFactory.configure(config.getClientPortAddress(), config.getMaxClientCnxns());
        // todo 跟进这个方法
        cnxnFactory.startup(zkServer);

public FileTxnSnapLog(File dataDir, File snapDir) throws IOException {
    LOG.debug("Opening datadir:{} snapDir:{}", dataDir, snapDir);
    // todo 关联上指定数据文件和日志文件
    // todo 给FileTxnSnapLog赋值
    this.dataDir = new File(dataDir, version + VERSION);
    this.snapDir = new File(snapDir, version + VERSION);
    if (!this.dataDir.exists()) {
    ...
    .
    .
   // todo 将这两个文件封装进 FileTxnLog 给当前类维护的两种事务快照( TnxnSnap ) 赋值
    txnLog = new FileTxnLog(this.dataDir);
    snapLog = new FileSnap(this.snapDir);

Thread thread;
@Override
public void configure(InetSocketAddress addr, int maxcc) throws IOException {
    configureSaslLogin();
    // todo 把当前类作为线程
    thread = new ZooKeeperThread(this, "NIOServerCxn.Factory:" + addr);
    //todo 所以这里的这个线程是为了和JVM生命周期绑定，只剩下这个线程时已经没有意义了，应该关闭掉。
    thread.setDaemon(true);
    maxClientCnxns = maxcc;
    // todo 看到了NIO原生的代码,使用打开服务端的 Channel, 绑定端口,设置为非阻塞,注册上感兴趣的事件是 accept 连接事件
    this.ss = ServerSocketChannel.open();
    ss.socket().setReuseAddress(true);
    LOG.info("binding to port " + addr);
    ss.socket().bind(addr);
    ss.configureBlocking(false);
    ss.register(selector, SelectionKey.OP_ACCEPT);
}

NIOServerCnxnFactory factory;
    final SocketChannel sock;
    protected final SelectionKey sk;
    boolean initialized;
    ByteBuffer lenBuffer = ByteBuffer.allocate(4);
    ByteBuffer incomingBuffer = lenBuffer;
    LinkedBlockingQueue<ByteBuffer> outgoingBuffers = new LinkedBlockingQueue<ByteBuffer>();
    int sessionTimeout;
    protected final ZooKeeperServer zkServer;

public void runFromConfig(ServerConfig config) throws IOException {
        .
        .
        . 
        cnxnFactory.startup(zkServer);

public void startup(ZooKeeperServer zks) throws IOException,
        InterruptedException {
    // todo start(); ==> run() 开启线程
    start(); //todo 实现在上面, 到目前为止服务端已经可以接受客户端的请求了
    // todo 将ZKS 交给NIOServerCnxnFactory管理,意味着NIOServerCnxnFactory是目前来说,服务端功能最多的对象
    setZooKeeperServer(zks);
    // todo 因为是服务端刚刚启动,需要从从disk将数据恢复到内存
    zks.startdata();
    // todo 继续跟进
    zks.startup();
}

// todo 将数据加载进缓存中
public void startdata() 
throws IOException, InterruptedException {
    //check to see if zkDb is not null
    if (zkDb == null) {
        // todo 如果没初始化的话就初始化
        zkDb = new ZKDatabase(this.txnLogFactory);
    }  
    if (!zkDb.isInitialized()) {
        // todo 恢复数据
        loadData();
    }
}

public ZKDatabase(FileTxnSnapLog snapLog) {
    // todo 创建了DataTree 数据树的空对象
    dataTree = new DataTree();
    sessionsWithTimeouts = new ConcurrentHashMap<Long, Integer>();
    //todo 用初始化好了的存有关于系统事务日志将snaplog初始化
    this.snapLog = snapLog;
}

public void loadData() throws IOException, InterruptedException {
        // todo zkDatabase 已经初始化了
        if(zkDb.isInitialized()){
            // todo zxid = 最近的一次znode的事务id
            setZxid(zkDb.getDataTreeLastProcessedZxid());
        } else {
            //todo  zkDB 没有初始化就使用  zkDb.loadDataBase() ， 跟进去看， 他从快照中获取数据
            setZxid(zkDb.loadDataBase());
        }
        // Clean up dead sessions
        LinkedList<Long> deadSessions = new LinkedList<Long>();
        for (Long session : zkDb.getSessions()) {
            if (zkDb.getSessionWithTimeOuts().get(session) == null) {
                deadSessions.add(session);
            }
        }
        zkDb.setDataTreeInit(true);
        for (long session : deadSessions) {
            // XXX: Is lastProcessedZxid really the best thing to use?
            killSession(session, zkDb.getDataTreeLastProcessedZxid());
        }
    }

// todo 继续启动， 服务端和客户端建立连接后会保留一个session， 其中这个sessiion的生命周期倒计时就在下面的 createSessionTracker();
    public synchronized void startup() {
        if (sessionTracker == null) {
            // todo 创建session计时器
            createSessionTracker();
        }
        // todo 开启计时器
        startSessionTracker();
        // todo 设置请求处理器， zookeeper中存在不同的请求处理器, 就在下面
        setupRequestProcessors();
        //todo 是一个为应用程序、设备、系统等植入管理功能的框架。
        //todo JMX可以跨越一系列异构操作系统平台、系统体系结构和网络传输协议，灵活的开发无缝集成的系统、网络和服务管理应用
        registerJMX();
        // todo 修改状态  --> running
        setState(State.RUNNING);
        // todo 唤醒所有线程, 因为前面有一个线程等待处理器 睡了一秒
        notifyAll();
    }

protected void setupRequestProcessors() {
        // todo 下面的三个处理器的第二个参数是在指定 下一个处理器是谁
        RequestProcessor finalProcessor = new FinalRequestProcessor(this);
        RequestProcessor syncProcessor = new SyncRequestProcessor(this,
                finalProcessor);
        // todo  在服务端， 数据的处理  socket -> packet -> request -> queue
        // todo 然后由下面的requestprocessor 链 进行下一步处理request
        // todo  开启新线程， 服务端接收的客户端的请求都放在了 队列中，用处理器异步处理
        ((SyncRequestProcessor)syncProcessor).start();
        //todo  第一个处理器 ， 下一个处理器是 syncProcessor  最后一个处理器 finalProcessor
        firstProcessor = new PrepRequestProcessor(this, syncProcessor);
                // todo  开启新线程  服务端接收的客户端的请求都放在了 队列中，用处理器异步处理
        ((PrepRequestProcessor)firstProcessor).start();
    }