java架構之路-（分布式zookeeper）zookeeper集群配置和選舉機制詳解

　　上次博客我們說了一下zookeeper的配置文件，以及命令的使用https://www.cnblogs.com/cxiaocai/p/11597465.html。我們這次來說一下我們的zookeeper的集群配置和java的API相關操作。

集群：

　　一般情況下我們用zookeeper來做任務調度中心的，所以一定要做到高可用的，單機的不可能做到永不宕機，我們也不會信任他單機的永不宕機，這時我們就需要做集群處理，來實現我們的高可用。

配置集群時，我們盡可能采用奇數的服務器來配置，什么意思呢？盡力采用3，5，7，9台服務器來配置，原因是zookeeper會默認識別半數以上服務器正常運行，才認為zookeeper是正常運行的，比如我們現在部署4台zookeeper服務器，這時其中兩台宕機了，這時zookeeper會認為這個集群時不可用的，同理我們如果是5台服務器的情況，有兩台宕機了，可以正常運行，三台宕機了，才被認為是不可用的，這個很重要，包括后面的選舉機制也是這樣的。資金有限啊，我先用3台服務器搭建一下zookeeper集群。

1.集群配置

　　下載解壓什么的就不說了啊，上次都說過了，我們直接看下配置文件吧。和單機配置基本一致，我們看到dataDir=/tmp/zookeeper，也就是我們的數據存儲路徑，分別建立三個文件myid，內部輸入1-255的數字

 每台服務器別重復，切記一定建立在配置文件dataDir對應的目錄下，不然啟動會報找不到myid文件的錯誤，沒有對應/tmp/zookeeper目錄的，可以啟動一下zookeeper再關閉就有文件夾了，或者自己手動創建也行。

再每一個配置文件內加入配置

server.1=172.16.140.106:2888:3888

server.2=172.16.140.105:2888:3888

server.3=172.16.214.74:2888:3888

server.myid(myid文件的數字)=ip（與myid相對應的IP）：集群之間相互通訊的IP：選舉時通訊的IP。三分配置文件都是一樣的。

我們來分別啟動一下。說到這里我們的集群配置也就成功了。

啟動成功以后，我們分別輸入./bin/zkServer.sh status 我們可以看到我們的服務器角色

 

 2.角色：

leader 主節點，又名領導者。用於寫入數據，通過選舉產生，如果宕機將會選舉新的主節點。

follower 子節點，又名追隨者。用於實現數據的讀取。同時他也是主節點的備選節點，並用擁有投票權。

observer 次級子節點，又名觀察者。用於讀取數據，與fllower區別在於沒有投票權，不能選為主節點。並且在計算集群可用狀態時不會將observer計算入內。也就是我們的半數原則計算。

observer配置：

只要在集群配置中加上observer后綴即可，示例如下：

server.3=127.0.0.1:2889:3889:observer

選舉機制：

　　先說一個簡單的，投票機制的。假設我們現在有1，2，3，4，5五個follower要進行選舉。

 

 

 簡單流程就是這樣的，第一輪都認為自己很可以，自己要當選leader，但是選舉流程失敗了，還得繼續，接下來會把自己的票全盤拖出給自己臨近的id，1就會給2一票，2現在有了兩票了，發現還是不夠半數啊，半數是2.5啊，算了還得繼續，2又把自己的兩票都給了3，3這時獲得了3票了，大於半數了，當選leader。

每輪選舉結束后都會統一來處理，如果一輪投票就發現server1的zxid較大，那么直接server1會當選leader。

優先檢查ZXID。ZXID比較大的服務器優先作為Leader。

如果ZXID相同，那么就比較myid。myid較大的服務器作為Leader服務器。

留下一個思考題，5台服務器，如果啟動可以指定 4號為leader服務 。

javaAPI相關操作

　　maven的pom文件內加入

<dependency>
    <groupId>org.apache.zookeeper</groupId>
    <artifactId>zookeeper</artifactId>
    <version>3.5.5</version>
</dependency>

首先我們先初始化我們的Zookeeper的連接

connectString ->連接String連接串，包括ip+端口 ,集群模式下用逗號隔開192.168.0.149:2181,192.168.0.150:2181

sessionTimeout ->會話超時時間，類型int，該值不能超過服務端所設置的minSessionTimeout（默認2s）和maxSessionTimeout（默認60s）,單位毫秒

watcher -> 會話監聽器Watcher，服務端事件將會觸該監聽

sessionId -> 自定義會話ID long

sessionPasswd ->byte[] 會話密碼

canBeReadOnly ->boolean該連接是否為只讀的 

hostProvider ->HostProvider 服務端地址提供者，指示客戶端如何選擇某個服務來調用，默認采用StaticHostProvider實現

@Before
public void init() throws IOException {
    String conn = "47.111.109.3:2181"; // 連接字符串
    int sessionTimeout  = 4000;  //連接超時時間
    zooKeeper = new ZooKeeper(conn, sessionTimeout, new Watcher() {
        public void process(WatchedEvent watchedEvent) {

        }
    });
}

我們先來看看我們的新增節點，刪除節點等操作吧。

查看節點：

我們使用getData方法，添加三個參數，分別是路徑，是否監聽，和返回值(狀態stat)。

/**
 * 獲取數據
 *
 * @throws KeeperException
 * @throws InterruptedException
 */
@Test
public void getData() throws KeeperException, InterruptedException {
    byte[] data = zooKeeper.getData("/root", false, null);
    System.out.println(new String(data));
}

添加監聽：

/**
 * 添加監聽,結果在初始化里
 *
 * @throws KeeperException
 * @throws InterruptedException
 */
@Test
public void getWatchData() throws KeeperException, InterruptedException {
    byte[] data = zooKeeper.getData("/root", true, null);
    System.out.println(new String(data));
    Thread.sleep(Long.MAX_VALUE);
}

這個監聽是一次性的，而且結果在我們的初始化的watch里，初始化方法改為

@Before
public void init() throws IOException {
    String conn = "47.111.109.3:2181"; // 連接字符串
    int sessionTimeout = 4000;  //連接超時時間
    zooKeeper = new ZooKeeper(conn, sessionTimeout, new Watcher() {
        public void process(WatchedEvent watchedEvent) {
            System.out.println(watchedEvent.getPath());
        }
    });
}

永久監聽設置，我們只要將Watcher新建一下就可以了嗎...我們來看一下實現

/**
 * 永久監聽
 *
 * @throws KeeperException
 * @throws InterruptedException
 */
@Test
public void getWatchDataForever() throws KeeperException, InterruptedException {
    byte[] data = zooKeeper.getData("/root", new Watcher() {
        public void process(WatchedEvent watchedEvent) {
            try {
                zooKeeper.getData(watchedEvent.getPath(),this,null);
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(watchedEvent.getPath());
        }
    }, null);
    System.out.println(new String(data));
    Thread.sleep(Long.MAX_VALUE);
}

stat：

/**
 * Stat
 *
 * @throws KeeperException
 * @throws InterruptedException
 */
@Test
public void getDataStat() throws InterruptedException, KeeperException {
    Stat stat = new Stat();
    zooKeeper.getData("/root",false, stat);
    System.out.println(stat);
}

輸出結果和我們命令stat的輸入其實是完全一致的，只不過一個事16進制，一個事10進制的，可以自己對比一下。

獲取子節點 ：

/**
 * 獲取子節點
 *
 * @throws KeeperException
 * @throws InterruptedException
 */
@Test
public void getWatchChildDataForever() throws KeeperException, InterruptedException {
    List<String> children = zooKeeper.getChildren("/root", false);
    for (int i = 0; i < children.size(); i++) {
        System.out.println(children.get(i));
    }
}

刪除節點：

/**
 * 刪除節點
 *
 * @throws KeeperException
 * @throws InterruptedException
 */
@Test
public void deletePath() throws KeeperException, InterruptedException {
    zooKeeper.delete("/root/d2", 0);
}

創建節點：

/**
 * 創建節點
 *
 * @throws KeeperException
 * @throws InterruptedException
 */
@Test
public void createPath() throws KeeperException, InterruptedException {
    List<ACL> acl = new ArrayList<ACL>();
    int perm = ZooDefs.Perms.ADMIN | ZooDefs.Perms.CREATE | ZooDefs.Perms.READ;
    ACL aclObj = new ACL(perm, new Id("world", "anyone"));
    acl.add(aclObj);
    zooKeeper.create("/root/d4", "hello".getBytes(), acl, CreateMode.CONTAINER);
}

說到這我們的API和集群操作就差不多說完了。

這次代碼不多，就先不上傳了，寫完下次博客再一起上傳。

 

最進弄了一個公眾號，小菜技術，歡迎大家的加入