Zookeeper應用場景

 

• 數據發布與訂閱：發布訂閱模型，就是發布者將數據發布到ZK節點上，供訂閱者動態獲取數據，實現數據的集中管理和動態更新。
• 配置中心：在應用中，將全局的配置信息放到ZK上集中管理，在應用啟動的時候主動獲取一次配置。同時，在節點上注冊一個watcher，保證每次數據更新時會通知訂閱者更新配置信息。
• 元數據維護：在分布式搜索服務中，索引的元信息和服務器集群機器的節點狀態保存在ZK的指定節點上，供各個客戶端使用。
• 分布式日志收集系統：這個系統的核心工作是收集不同服務器的日志。收集器通常是按照應用來分配任務單元，因此在ZK上用應用名創建一個節點，將需要收集的服務器的IP注冊到這個節點的子結點上。
• 命名服務：客戶端應用能夠根據指定名字來獲取資源或服務的地址、提供者等信息。
• 分布式通知/協調：ZK中特有的watcher注冊與異步通知機制，能夠很好的實現分布式環境下不同系統之間的通知與協調，實現對數據變更的實時處理。使用方法通常是：不同系統對ZK上的同一個節點進行注冊，監聽節點的變化。任意一個系統對節點進行了更新，其它系統都能接到通知，並進行相應處理。
• 任務匯報：類似於任務分發系統。子任務啟動后，在ZK上注冊一個臨時節點，並定時將自己的進度寫入這個節點，這樣任務管理者可以實時查看任務進度。
• 服務器列表維護：能都動態監聽服務器的上下線
• 服務端代碼

package zookeeper;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;

import java.io.IOException;

public class Server {

    private static final String CONNECT_STRING = "172.17.23.79:2181,172.17.23.79:2182";
    private static final int TIME_OUT = 15000;

    private ZooKeeper zooKeeper;

    public Server() throws IOException {
        this.zooKeeper = new ZooKeeper(CONNECT_STRING, TIME_OUT, watchedEvent -> {

        });
    }

    /**
     * 創建臨時序列節點（服務器關閉時節點會自動刪除）
     * @param serverName
     * @return
     * @throws Exception
     */
    public String regsterServer(String serverName) throws Exception {
        String result = zooKeeper.create("/clusterServer/server", serverName.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println(serverName+" server start....");
        return result;
    }

    public static void main(String[] args) throws Exception {
        Server server = new Server();
        server.regsterServer(args[0]);
        //保持程序運行，防止程序停止運行導致節點自動刪除
        while (true) {

        }
    }
}

• • • 客戶端代碼

package zookeeper;

import org.apache.zookeeper.ZooKeeper;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
* 獲取並監聽注冊的服務器列表
*/
public class Client {

    private static final String CONNECT_STRING = "172.17.23.79:2181,172.17.23.79:2182";
    private static final int TIME_OUT = 15000;

    private ZooKeeper zooKeeper;

    public Client() throws IOException {
        this.zooKeeper = new ZooKeeper(CONNECT_STRING, TIME_OUT, watchedEvent -> {
            System.out.println("watcher works.");
            try {
                getServers();
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }

    public void getServers() throws Exception {
        List<String> servers = new ArrayList<>();
        List<String> children = zooKeeper.getChildren("/clusterServer", true, null);
        for (String child : children) {
            String server = getData("/clusterServer/"+child);
            servers.add(server);
        }
        System.out.println(servers);
    }

    public String getData(String path) throws Exception {
        byte[] data = zooKeeper.getData(path, true, null);
        return new String(data);
    }

    public static void main(String[] args) throws Exception {
        Client client = new Client();
        client.getServers();
        while (true) {

        }
    }
}

• 分布式鎖：
• 保持獨占：通常的做法是把ZK的節點看作一把鎖，通過create node的方式來實現。

                              具體的實現方式是：

• • 方法一：所有的客戶端都去創建/lock節點，最終創建成功的持有這把鎖。（同一個節點只能創建一次，再次創建會返回失敗信息）

/**
* 通過創建臨時節點，實現服務器之間的獨占鎖
*/
@Test
public void singleLock() {
    try {
        //參數：1，節點路徑； 2，要存儲的數據； 3，節點的權限； 4，節點的類型
        String nodePath = zooKeeper.create("/lock", "This is Lock.".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        //創建成功，則相當於擁有獨占鎖，可以進行以下邏輯
        //TODO 業務邏輯
        System.out.println(nodePath);
        //業務邏輯結束后，刪除節點，即釋放鎖資源
        zooKeeper.delete("/lock", -1);
    } catch (Exception e) {
        //創建節點失敗，重新調用，直至創建成功
        if (e instanceof KeeperException && "NODEEXISTS".equals(((KeeperException)e).code().name())) {
            System.out.println("Node exists.");
            singleLock();
        }else {
            e.printStackTrace();
        }
    }
}

• • • • 方法二：判斷/lock節點是否存在，如果存在，說明其它服務器已經持有鎖資源；如果不存在，則鎖資源處於空閑狀態，創建節點占有鎖資源。

/**
* 通過創建臨時節點，實現服務器之間的獨占鎖
*/
@Test
public void singleLock2() throws KeeperException, InterruptedException {
    Stat stat = zooKeeper.exists("/lock", false);
    //如果節點已經存在，等待其它服務器刪除節點。即：等待其它服務器釋放鎖資源
    while(stat != null) {  }


    //參數：1，節點路徑； 2，要存儲的數據； 3，節點的權限； 4，節點的類型
    String nodePath = zooKeeper.create("/lock", "This is Lock.".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
    //創建成功，則相當於擁有獨占鎖，可以進行以下邏輯
    //TODO 業務邏輯
    System.out.println(nodePath);
    //業務邏輯結束后，刪除節點，即釋放鎖資源
    zooKeeper.delete("/lock", -1);
}

• • 控制時序：/lock節點已存在，客戶端在它下面創建臨時有序節點/lock/{sessionId}-1 , /lock/{sessionId}-2 , /lock/{sessionId}-3 …..（通過節點屬性控制 CreateMode.EPHEMERAL_SEQUENTIAL控制），保證子節點創建的有序性，從而保證的客戶端的時序性。
    • 方式一：獲取鎖資源時，程序處於block狀態，直到獲取鎖資源。

/**
* 通過創建臨時時序節點，實現服務器之間的時序鎖
*/
@Test
public void lock() throws KeeperException, InterruptedException {
    //創建臨時時序節點
    String nodePath = zooKeeper.create("/lock/sublock", "This is sub lock.".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
    
    while(true) {
        List<String> children = zooKeeper.getChildren("/lock", false);
        //排序，並獲取序號最小的節點。（序號越小，表明請求時間越早，優先獲取鎖資源）
        children.sort(String::compareTo);
        if (nodePath.equals("/lock/"+children.get(0))){
            //TODO 業務邏輯
            System.out.println("TODO  Logic.");
            break;
        }
    }

    //業務邏輯結束后，刪除節點，即釋放鎖資源
    zooKeeper.delete(nodePath, -1);
}

• • • 方式二：通過watcher監聽服務器列表變化，判斷當前服務器是否獲取鎖資源，程序不會block

package zookeeper;


import org.apache.zookeeper.*;


import java.io.IOException;
import java.util.Collections;
import java.util.List;


/**
*
* 通過Zookeeper實現服務器之間的時序鎖
*
*/
public class SeqLock {

    private static final String CONNECT_STRING = "172.17.23.79:2181,172.17.23.79:2182";

    private static ZooKeeper zooKeeper;
    private String thispath;

    public SeqLock() throws IOException {
        //超時時間單位：毫秒
        zooKeeper = new ZooKeeper(CONNECT_STRING, 15000, event -> {
            //監聽“/lock”的子節點變化。如果有服務器釋放鎖，判斷自己是否獲取鎖
            if (event.getType() == Watcher.Event.EventType.NodeChildrenChanged
                    && event.getPath().startsWith("/lock")) {
                try {
                    List<String> children = zooKeeper.getChildren("/lock", false);
                    if (children.size() > 0) {
                        Collections.sort(children);
                        String fistNode = "/lock/"+children.get(0);
                        if (fistNode.equals(thispath)){
                            doSomethingAndDelNode();
                        }
                    }


                } catch (KeeperException | InterruptedException e) {
                    e.printStackTrace();
                }


            }
        });
    }


    /**
     * 通過創建臨時時序節點注冊時序鎖，並監聽服務器列表
     */
    public void lock() throws KeeperException, InterruptedException {
        thispath = zooKeeper.create("/lock/sublock", "This is sub lock.".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
        System.out.println(thispath);
        List<String> children = zooKeeper.getChildren("/lock", false);
        //如果只有一個子節點，說明鎖資源被當前服務器持有。如果子節點不止一個，說明鎖資源已經被其它服務器持有
        if(children.size() == 1) {
            doSomethingAndDelNode();
        }
    }

    private void doSomethingAndDelNode() throws InterruptedException, KeeperException {
        //TODO 業務邏輯
        System.out.println("TODO  Logic.");
        //業務邏輯結束后，刪除節點，即釋放鎖資源
        zooKeeper.delete(thispath, -1);
    }

    public static void main(String[] args)  {
        try {
            SeqLock lock = new SeqLock();
            lock.lock();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

 

參考文獻：

• ZooKeeper典型應用場景

鏈接   ：https://pan.baidu.com/s/1lohZ98K33z_Hf_8Y370Nuw
提取碼：kydw