[TOC] # Zookeeper命令行以及API使用 ## 1 Zookeeper命令行操作 ### 1 客戶端連接 ~~~ 運行 zkCli.sh –server <ip>進入命令行工具 ~~~  ### 2 查看znode路徑 ~~~ ls /mygirls ~~~ ### 3 獲取znode數據 ~~~ get /mygirls ~~~ ### 4 監聽znode事件 ~~~ ls /mygirls watch ## 就對一個節點的子節點變化事件注冊了監聽 get /mygirls watch ## 就對一個節點的數據內容變化事件注冊了監聽 ~~~ > 注意： 監聽器只生效一次 > 監聽器的工作機制，其實是在客戶端會專門創建一個監聽線程，在本機的一個端口上等待zk集群發送過來事件 ## 2 Zookeeper 客戶端API ### 1 基本使用 > org.apache.zookeeper.Zookeeper是客戶端入口主類，負責建立與server的會話 > 它提供以下幾類主要方法 ： | 功能 | 描述 | | --- | --- | | create | 在本地目錄樹中創建一個節點 | | delete | 刪除一個節點 | | exists | 測試本地是否存在目標節點 | | get/set data | 從目標節點上讀取 / 寫數據 | | get/set ACL| | 獲取 / 設置目標節點訪問控制列表信息 | | get children | 檢索一個子節點上的列表 | | sync | 等待要被傳送的數據 | ### 2 增刪改查znode數據 ~~~ public class SimpleDemo { // 會話超時時間，設置為與系統默認時間一致 private static final int SESSION_TIMEOUT = 30000; // 創建 ZooKeeper 實例 ZooKeeper zk; // 創建 Watcher 實例 Watcher wh = new Watcher() { public void process(org.apache.zookeeper.WatchedEvent event) { System.out.println(event.toString()); } }; // 初始化 ZooKeeper 實例 private void createZKInstance() throws IOException { zk = new ZooKeeper("weekend01:2181", SimpleDemo.SESSION_TIMEOUT, this.wh); } private void ZKOperations() throws IOException, InterruptedException, KeeperException { System.out.println("/n1. 創建 ZooKeeper 節點 (znode ： zoo2, 數據： myData2 ，權限： OPEN_ACL_UNSAFE ，節點類型： Persistent"); zk.create("/zoo2", "myData2".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); System.out.println("/n2. 查看是否創建成功： "); System.out.println(new String(zk.getData("/zoo2", false, null))); System.out.println("/n3. 修改節點數據 "); zk.setData("/zoo2", "shenlan211314".getBytes(), -1); System.out.println("/n4. 查看是否修改成功： "); System.out.println(new String(zk.getData("/zoo2", false, null))); System.out.println("/n5. 刪除節點 "); zk.delete("/zoo2", -1); System.out.println("/n6. 查看節點是否被刪除： "); System.out.println(" 節點狀態： [" + zk.exists("/zoo2", false) + "]"); } private void ZKClose() throws InterruptedException { zk.close(); } public static void main(String[] args) throws IOException, InterruptedException, KeeperException { SimpleDemo dm = new SimpleDemo(); dm.createZKInstance(); dm.ZKOperations(); dm.ZKClose(); } } ~~~ ### 3 監聽znode > Zookeeper的監聽器工作機制  > 監聽器是一個接口，我們的代碼中可以實現Wather這個接口，實現其中的process方法，方法中即我們自己的業務邏輯  > 監聽器的注冊是在獲取數據的操作中實現： > getData(path,watch?)監聽的事件是：節點數據變化事件 > getChildren(path,watch?)監聽的事件是：節點下的子節點增減變化事件 ## 3 Zookeeper 應用案例 ### 1 案例1——服務器上下線動態感知 #### 1.1 需求描述 > 某分布式系統中，主節點可以有多臺，可以動態上下線 > 任意一臺客戶端都能實時感知到主節點服務器的上下線 #### 1.2 設計思路  #### 1.3 代碼開發 > 1、客戶端實現 ~~~ public class AppClient { private String groupNode = "sgroup"; private ZooKeeper zk; private Stat stat = new Stat(); private volatile List<String> serverList; /** * 連接zookeeper */ public void connectZookeeper() throws Exception { zk = new ZooKeeper("localhost:4180,localhost:4181,localhost:4182", 5000, new Watcher() { public void process(WatchedEvent event) { // 如果發生了"/sgroup"節點下的子節點變化事件, 更新server列表, 并重新注冊監聽 if (event.getType() == EventType.NodeChildrenChanged && ("/" + groupNode).equals(event.getPath())) { try { updateServerList(); } catch (Exception e) { e.printStackTrace(); } } } }); updateServerList(); } /** * 更新server列表 */ private void updateServerList() throws Exception { List<String> newServerList = new ArrayList<String>(); // 獲取并監聽groupNode的子節點變化 // watch參數為true, 表示監聽子節點變化事件. // 每次都需要重新注冊監聽, 因為一次注冊, 只能監聽一次事件, 如果還想繼續保持監聽, 必須重新注冊 List<String> subList = zk.getChildren("/" + groupNode, true); for (String subNode : subList) { // 獲取每個子節點下關聯的server地址 byte[] data = zk.getData("/" + groupNode + "/" + subNode, false, stat); newServerList.add(new String(data, "utf-8")); } // 替換server列表 serverList = newServerList; System.out.println("server list updated: " + serverList); } /** * client的工作邏輯寫在這個方法中 * 此處不做任何處理, 只讓client sleep */ public void handle() throws InterruptedException { Thread.sleep(Long.MAX_VALUE); } public static void main(String[] args) throws Exception { AppClient ac = new AppClient(); ac.connectZookeeper(); ac.handle(); } } ~~~ > 2、服務器端實現 ~~~ public class AppServer { private String groupNode = "sgroup"; private String subNode = "sub"; /** * 連接zookeeper * @param address server的地址 */ public void connectZookeeper(String address) throws Exception { ZooKeeper zk = new ZooKeeper( "localhost:4180,localhost:4181,localhost:4182", 5000, new Watcher() { public void process(WatchedEvent event) { // 不做處理 } }); // 在"/sgroup"下創建子節點 // 子節點的類型設置為EPHEMERAL_SEQUENTIAL, 表明這是一個臨時節點, 且在子節點的名稱后面加上一串數字后綴 // 將server的地址數據關聯到新創建的子節點上 String createdPath = zk.create("/" + groupNode + "/" + subNode, address.getBytes("utf-8"), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); System.out.println("create: " + createdPath); } /** * server的工作邏輯寫在這個方法中 * 此處不做任何處理, 只讓server sleep */ public void handle() throws InterruptedException { Thread.sleep(Long.MAX_VALUE); } public static void main(String[] args) throws Exception { // 在參數中指定server的地址 if (args.length == 0) { System.err.println("The first argument must be server address"); System.exit(1); } AppServer as = new AppServer(); as.connectZookeeper(args[0]); as.handle(); } } ~~~ ### 2 案例2——分布式共享鎖 #### 1、需求描述 > 在我們自己的分布式業務系統中，可能會存在某種資源，需要被整個系統的各臺服務器共享訪問，但是只允許一臺服務器同時訪問 #### 2、設計思路  #### 3、代碼開發 ~~~ public class DistributedClientMy { // 超時時間 private static final int SESSION_TIMEOUT = 5000; // zookeeper server列表 private String hosts = "spark01:2181,spark02:2181,spark03:2181"; private String groupNode = "locks"; private String subNode = "sub"; private boolean haveLock = false; private ZooKeeper zk; // 當前client創建的子節點 private volatile String thisPath; /** * 連接zookeeper */ public void connectZookeeper() throws Exception { zk = new ZooKeeper("spark01:2181", SESSION_TIMEOUT, new Watcher() { public void process(WatchedEvent event) { try { // 子節點發生變化 if (event.getType() == EventType.NodeChildrenChanged && event.getPath().equals("/" + groupNode)) { // thisPath是否是列表中的最小節點 List<String> childrenNodes = zk.getChildren("/" + groupNode, true); String thisNode = thisPath.substring(("/" + groupNode + "/").length()); // 排序 Collections.sort(childrenNodes); if (childrenNodes.indexOf(thisNode) == 0) { doSomething(); thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); } } } catch (Exception e) { e.printStackTrace(); } } }); // 創建子節點 thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); // wait一小會, 讓結果更清晰一些 Thread.sleep(new Random().nextInt(1000)); // 監聽子節點的變化 List<String> childrenNodes = zk.getChildren("/" + groupNode, true); // 列表中只有一個子節點, 那肯定就是thisPath, 說明client獲得鎖 if (childrenNodes.size() == 1) { doSomething(); thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); } } /** * 共享資源的訪問邏輯寫在這個方法中 */ private void doSomething() throws Exception { try { System.out.println("gain lock: " + thisPath); Thread.sleep(2000); // do something } finally { System.out.println("finished: " + thisPath); // 將thisPath刪除, 監聽thisPath的client將獲得通知 // 相當于釋放鎖 zk.delete(this.thisPath, -1); } } public static void main(String[] args) throws Exception { DistributedClientMy dl = new DistributedClientMy(); dl.connectZookeeper(); Thread.sleep(Long.MAX_VALUE); } } ~~~