背景介紹

ZooKeeper可以以standalone,偽分散式和分散式三種方式部署.standalone模式下只有一臺機器作為伺服器,喪失了ZooKeeper高可用的特點.偽分散式是在一臺電腦上使用不同埠啟動多個ZooKeeper伺服器.分散式是使用多個機器,每臺機器上部署一個ZooKeeper伺服器,即使有伺服器宕機,只要少於半數,ZooKeeper叢集依然可以正常對外提供服務.
ZooKeeper以standalone模式啟動只需啟動對客戶端提供服務的元件,無需啟動叢集內部通訊元件,較為簡單,因此先從standalone模式開始介紹.

整體架構

Zookeeper整體架構如上圖,其中包括ServerCnxnFactory

啟動流程概述

standalone模式啟動主要包括如下幾個步驟:

1. 配置檔案解析
2. 建立並啟動歷史檔案清理器
3. 初始化資料管理器
4. 註冊shutdownHandler
5. 啟動Admin server
6. 建立並啟動網路IO管理器
7. 啟動ZooKeeperServer
8. 建立並啟動secureCnxnFactory
9. 建立並啟動ContainerManager

原始碼如下:

    protected void initializeAndRun(String[] args)
            throws
 
 ConfigException, IOException, AdminServerException {
        //1.解析配置檔案
        QuorumPeerConfig config = new QuorumPeerConfig();
        if (args.length == 1) {
            config.parse(args[0]);
        }

        // Start and schedule the the purge task
        //2.建立並啟動歷史檔案清理器(對事務日誌和快照資料檔案進行定時清理)
        DatadirCleanupManager purgeMgr = new
 
 DatadirCleanupManager(config
                .getDataDir(), config.getDataLogDir(), config
                .getSnapRetainCount(), config.getPurgeInterval());
        purgeMgr.start();

        if (args.length == 1 && config.isDistributed()) {
            //叢集啟動
            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);
        }
    }

    /**
     * Run from a ServerConfig.
     *
     * @param config ServerConfig to use.
     * @throws IOException
     * @throws AdminServerException
     */
    public void runFromConfig(ServerConfig config)
            throws IOException, AdminServerException {
        LOG.info("Starting server");
        FileTxnSnapLog txnLog = null;
        try {
            //3.建立ZooKeeper資料管理器
            txnLog = new FileTxnSnapLog(config.dataLogDir, config.dataDir);
            final ZooKeeperServer zkServer = new ZooKeeperServer(txnLog,
                    config.tickTime, config.minSessionTimeout, config.maxSessionTimeout, null);
            txnLog.setServerStats(zkServer.serverStats());

            //4.註冊shutdownHandler,在ZooKeeperServer的狀態變化時呼叫shutdownHandler的handle()
            final CountDownLatch shutdownLatch = new CountDownLatch(1);
            zkServer.registerServerShutdownHandler(
                    new ZooKeeperServerShutdownHandler(shutdownLatch));

            //5.啟動Admin server
            adminServer = AdminServerFactory.createAdminServer();
            adminServer.setZooKeeperServer(zkServer);
            adminServer.start();

            //6.建立並啟動網路IO管理器
            boolean needStartZKServer = true;
            if (config.getClientPortAddress() != null) {
                cnxnFactory = ServerCnxnFactory.createFactory();
                cnxnFactory.configure(config.getClientPortAddress(), config.getMaxClientCnxns(), false);
                //7.此方法除了啟動ServerCnxnFactory,還會啟動ZooKeeper
                cnxnFactory.startup(zkServer);
                // zkServer has been started. So we don't need to start it again in secureCnxnFactory.
                needStartZKServer = false;
            }
            //8.建立並啟動secureCnxnFactory 
            if (config.getSecureClientPortAddress() != null) {
                secureCnxnFactory = ServerCnxnFactory.createFactory();
                secureCnxnFactory.configure(config.getSecureClientPortAddress(), config.getMaxClientCnxns(), true);
                secureCnxnFactory.startup(zkServer, needStartZKServer);
            }
            //9.建立並啟動ContainerManager
            containerManager = new ContainerManager(zkServer.getZKDatabase(), zkServer.firstProcessor,
                    Integer.getInteger("znode.container.checkIntervalMs", (int) TimeUnit.MINUTES.toMillis(1)),
                    Integer.getInteger("znode.container.maxPerMinute", 10000)
            );
            containerManager.start();

            // Watch status of ZooKeeper server. It will do a graceful shutdown
            // if the server is not running or hits an internal error.
            //伺服器正常啟動時,執行到此處阻塞,只有server的state變為ERROR或SHUTDOWN時繼續執行後面的程式碼
            shutdownLatch.await();

            shutdown();

            if (cnxnFactory != null) {
                cnxnFactory.join();
            }
            if (secureCnxnFactory != null) {
                secureCnxnFactory.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();
            }
        }
    }

ZooKeeperstandalone啟動共有9個步驟,其中有些步驟還有子步驟,有些步驟還會啟動執行緒.因此下文中只會對一些簡單的步驟進行介紹,複雜的步驟留作日後補充,接下來我們就分別看下這9個步驟.

解析配置檔案

ZooKeeper啟動時會讀取配置檔案,預設讀取$ZK_HOME/conf/zoo.cfg,其將檔案解析為java.util.Properties,根據Properties中鍵值對的key設定相應value.

建立並啟動歷史檔案清理器

ZooKeeper雖然是一個記憶體資料庫,但是其通過快照和事務日誌提供了持久化的功能.在ZooKeeper啟動時,會根據快照和事務日誌恢復資料,重建記憶體資料庫;每次寫操作提交時,會在事務日誌中增加一條記錄,表明此次寫操作更改了哪些資料;在進行snapCount次事務之後,將記憶體資料庫所有節點的資料和當前會話資訊生成一個快照.
ZooKeeper的事務日誌類似於MySQL的redolog,若每次寫操作後直接將資料寫到磁碟上,則會存在大量的磁碟隨機讀寫,若是寫事務日誌,則將磁碟隨機讀寫轉換為順序讀寫.保證了資料的永續性的同時也兼顧了效能.
隨著時間的推移,會生成越來越多的快照和事務日誌檔案,為了定時清理無效日誌,DatadirCleanupManager啟動定時任務完成日誌檔案的清理.

相關配置

思考

配置項中只有snapRetainCount用於設定清理後保留的快照檔案個數,那清理快照檔案時會同時清理事務日誌檔案嗎?若會清理,清理之後會保留幾個事務日誌檔案呢?

答案:清理快照檔案時會同時清理事務日誌檔案,假如保留了3個快照檔案,其後綴名分別為100,200,300,則若事務日誌檔案中包含事務id>100的事務,則該事務日誌檔案被保留.則事務日誌檔案字尾>100的都會被保留,此外,字尾名<=100的事務日誌檔案中最新的事務日誌也被保留.因為即使該事務日誌檔案字尾<=100,但是可能其包含的事務中一部分id<=100,一部分>100,此時也需保留該檔案
事務日誌和快照檔案字尾名的含義見Zookeeper-持久化

建立ZooKeeper資料管理器

即初始化FileTxnSnapLog,FileTxnSnapLog組合了TxnLog和SnapShot,根據類名也可以看出,TxnLog負責處理事務日誌,SnapShot負責處理快照.FileTxnSnapLog是Zookeeper上層伺服器和底層資料儲存之間的對接層,提供一系列操作資料檔案的方法,如:

1. restore(DataTree, Map, PlayBackListener)
   啟動ZookeeperServer時呼叫此方法從磁碟上的快照和事務日誌中恢復資料
2. getLastLoggedZxid()
   獲取日誌中記載的最新的zxid
3. save(DataTree,ConcurrentHashMap, boolean syncSnap)
   將記憶體中的資料持久化到磁碟中

除此之外還有大量方法便於操作快照和事務日誌.

註冊shutdownhandler

在伺服器單機啟動結束後有一句shutdownLatch.await(),伺服器執行到此已經啟動完畢,主執行緒阻塞在此處.但伺服器退出時還需要做一些清理工作,因此註冊shutdownhandler,在ZooKeeperServer#setState(State)中呼叫此方法.

    /**
     * 當伺服器狀態變為`ERROR`或`SHUTDOWN`時喚醒shutdownLatch,執行後續的清理程式碼.
     * @param state new server state
     */
    void handle(State state) {
        if (state == State.ERROR || state == State.SHUTDOWN) {
            shutdownLatch.countDown();
        }
    }

啟動Admin server

AdminServer是3.5.0之後支援的特性,啟動了一個jettyserver,預設埠是8080,訪問此埠可以獲取Zookeeper執行時的相關資訊:

如伺服器的相關配置,統計資訊等.

相關配置

其配置如下

如果在啟動Zookeeper時提示Unable to start AdminServer, exiting abnormally,可能就是tomcat或其他軟體佔用了8080埠,需要修改AdminServer的預設埠.

建立並啟動網路IO管理器

ServerCnxnFactory是Zookeeper中的重要元件,負責處理客戶端與伺服器的連線.主要有兩個實現,一個是NIOServerCnxnFactory,使用Java原生NIO處理網路IO事件;另一個是NettyServerCnxnFactory,使用Netty處理網路IO事件.作為處理客戶端連線的元件,其會啟動若干執行緒監聽客戶端連線埠(即預設的9876埠).由於此元件非常複雜,日後單寫一篇部落格講解

啟動ZooKeeperServer

啟動Zookeeper會完成兩件事情,一是從磁碟上快照和事務日誌檔案將資料恢復到記憶體中,二是啟動會話管理器

恢復資料

    /**
     * 初始化ZkDatabase
     */
    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();
        }
    }

    /**
     * Restore sessions and data
     */
    public void loadData() throws IOException, InterruptedException {
        if (zkDb.isInitialized()) {
            setZxid(zkDb.getDataTreeLastProcessedZxid());
        } else {
            //1.由於zkDatabase尚未初始化,進入此分支(通過快照和事務日誌恢復資料)
            setZxid(zkDb.loadDataBase());
        }

        // 2.清理過期session,刪除其對應的node
        List<Long> deadSessions = new LinkedList<Long>();
        for (Long session : zkDb.getSessions()) {
            if (zkDb.getSessionWithTimeOuts().get(session) == null) {
                deadSessions.add(session);
            }
        }

        for (long session : deadSessions) {
            // XXX: Is lastProcessedZxid really the best thing to use?
            killSession(session, zkDb.getDataTreeLastProcessedZxid());
        }

        // 3.做一次快照
        takeSnapshot();
    }

啟動會話管理器

在介紹Zookeeper的回話之前,我們先回憶下Http中的會話.

由於HTTP協議是無狀態的協議，所以服務端需要記錄使用者的狀態時，就需要用某種機制來識具體的使用者，這個機制就是Session.典型的場景比如購物車，當你點選下單按鈕時，由於HTTP協議無狀態，所以並不知道是哪個使用者操作的，所以服務端要為特定的使用者建立了特定的Session，用於標識這個使用者，並且跟蹤使用者，這樣才知道購物車裡面有幾本書。這個Session是儲存在服務端的，有一個唯一標識。在服務端儲存Session的方法很多，記憶體、資料庫、檔案都有。叢集的時候也要考慮Session的轉移，在大型的網站，一般會有專門的Session伺服器叢集，用來儲存使用者會話，這個時候 Session 資訊都是放在記憶體的，使用一些快取服務比如Memcached之類的來放 Session。

session是一個抽象概念,開發者為了實現中斷和繼續等操作,將 user agent 和 server 之間一對一的互動,抽象為”會話”,進而衍生出”會話狀態”,也就是 session 的概念.
而session的實現一般是在服務端儲存的一個數據結構,用來跟蹤使用者的狀態,這個資料可以儲存在叢集,資料庫,檔案中,每一個session都有一個sessionId用來唯一標識session.客戶端在傳送請求時,將sessionId作為請求引數傳送給伺服器,伺服器就可根據sessionId找到儲存在伺服器中的session.

由於Zookeeper提供了臨時節點,Watcher通知等功能.自然需要儲存客戶端的狀態,會話管理器就是Zookeeper中用於管理會話的元件.由於此元件過於複雜,單獨介紹.

初始化zookeeper的請求處理鏈

類比於tomcat,tomcat處理請求時會構造pipeline和value,filter和filterChain兩個攔截過濾器處理請求,便於職責的解耦.Zookeeper會構造一個請求處理鏈用於處理客戶端傳送的請求.此元件過於複雜,單獨介紹.

註冊JMX

JMX的全稱為Java Management Extensions. 顧名思義，是管理Java的一種擴充套件。這種機制可以方便的管理、監控正在執行中的Java程式。常用於管理執行緒，記憶體，日誌Level，服務重啟，系統環境等

Zookeeper內部封裝了註冊JMX的邏輯,JMX註冊成功後,可以通過visualVM檢視和修改執行時屬性.JMX相關知識請查閱資料.

建立並啟動secureCnxnFactory

個人推測此元件應該和ServerCnxnFactory提供的功能類似,可能增加了認證的邏輯,目前沒有在網上關於此元件原始碼的資料,有時間檢視原始碼後補充.

建立並啟動ContainerManager

容器節點

Zookeeper中的節點型別有持久節點,持久順序節點,臨時節點,臨時順序節點,通過建立臨時順序節點,我們可以實現leader選舉,分散式鎖等功能,比如實現一個分散式鎖,我們的思路一般如下:

1. 建立一個持久節點,如”/lock”
2. 每一個想獲取鎖的程序在該節點下建立子節點,子節點型別為臨時順序節點
3. 建立了若干臨時順序節點中順序號最小的節點的執行緒獲得鎖;若程序未獲得鎖,則在順序號最小的節點上註冊監聽事件,監聽事件中包括競爭鎖的相關邏輯.當獲取鎖的程序釋放鎖(即刪除順序號最小的節點)時將回調監聽事件競爭鎖.(簡單介紹分散式鎖的實現思路,未解決羊群效應)

問題出現在第一步,為了實現分散式鎖的邏輯,我們必須建立一個父節點,且其型別為持久節點,但是當不需要分散式鎖時誰來刪除/lock節點呢?
為了解決這個問題,Zookeeper在3.6.0版本新增一種節點型別,即容器節點.其特點為:當容器節點的最後一個孩子節點被刪除之後，容器節點將被標註並在一段時間後刪除.
那麼在實現分散式鎖時,可以將/lock型別設定為容器節點,當沒有執行緒競爭分散式鎖時,/lock節點會被Zookeeper自動刪除.

屬性

ContainerManager中有兩個重要引數控制其行為:

注:上述屬性通過設定系統屬性配置,即在啟動QuorumPeerMain時新增-Dznode.container.checkIntervalMs=XXX

實現

為了能夠及時清理容器節點,通過Timer來執行定時任務,實現程式碼如下:

    /**
     * Manually check the containers. Not normally used directly
     */
    public void checkContainers()
            throws InterruptedException {
        //刪除兩個容器節點之間的最小間隔,預設:6ms
        long minIntervalMs = getMinIntervalMs();
        //遍歷待刪除的容器節點(同時會刪除過期的TTL節點)
        for (String containerPath : getCandidates()) {
            long startMs = Time.currentElapsedTime();

            ByteBuffer path = ByteBuffer.wrap(containerPath.getBytes());
            Request request = new Request(null, 0, 0,
                    ZooDefs.OpCode.deleteContainer, path, null);
            try {
                LOG.info("Attempting to delete candidate container: {}",
                        containerPath);
                //只是將刪除節點的請求傳送給PrepRequestProcessor,並未真正刪除該節點
                requestProcessor.processRequest(request);
            } catch (Exception e) {
                LOG.error("Could not delete container: {}",
                        containerPath, e);
            }
            //刪除一個容器節點所需時間
            long elapsedMs = Time.currentElapsedTime() - startMs;
            long waitMs = minIntervalMs - elapsedMs;
            //若刪除一個容器節點所需時間小於minIntervalMs,執行緒sleep.
            // 由於Timer內部只有一個執行緒,因此可以保證刪除兩個容器節點之間的時間間隔至少是minIntervalMs
            if (waitMs > 0) {
                Thread.sleep(waitMs);
            }
        }
    }

總結

作為一個伺服器,除了在主執行緒中進行初始化工作,還會開啟若干執行緒,為客戶端提供服務,在這裡,我們總結下Zookeeper單機啟動時啟動了多少執行緒:

1. 歷史檔案清理執行緒:通過Timer定時執行,執行週期為小時級別
2. Admin server:通過內建的jetty監聽8080埠,但由於對jetty不瞭解,不知一共啟動了多少個執行緒
3. ServerCnxnFactory:此元件負責管理客戶端的TCP連線,其有兩種實現,分別是原生NIO和基於Netty的實現
4. 會話管理器:啟動若干執行緒
5. 請求處理鏈:啟動若干執行緒
6. SecureServerCnxnFactory:與ServerCnxnFactory類似
7. ContainerManager:通過Timer定時執行,清理過期的容器節點和TTL節點,執行週期為分鐘級別

可以看出,有很多啟動執行緒的元件在此都未做介紹,正式因為啟動執行緒的元件是伺服器的重點,內容繁多,另開部落格介紹

參考