zookeeper day2

1 zk 事件監(jiān)聽機制

1.1 watcher概念

zk 提供了消息發(fā)布/消息訂閱功能盲再。多個訂閱者同時監(jiān)聽某一個主題對象。當該主題對象發(fā)生改變（節(jié)點內(nèi)容改變，子節(jié)點列表改變等），會實時页屠、主動通知所有訂閱者。

zk采用Watcher機制來實現(xiàn)發(fā)布訂閱功能蓖柔。
該機制會在主題對象發(fā)生變化后異步通知客戶端辰企，因此客戶端不必再訂閱后輪詢阻塞，從而減輕客戶端壓力况鸣。

watcher機制與觀察者模式類似牢贸。可用看作是觀察者模式在分布式場景中的實現(xiàn)镐捧。

1.2 watcher架構(gòu)
由三部分組成：

  • zk 服務端
  • zk 客戶端
  • 客戶端的ZKWatchManager對象

1 客戶端首先將watch注冊到服務端潜索，同時將watcher對象保存到客戶端的watcher管理器中。
2 當zk服務端監(jiān)聽的數(shù)據(jù)狀態(tài)發(fā)生變化后懂酱，服務端會主動通知客戶端竹习。
3 接著客戶端的watcher管理器會觸發(fā)相關(guān)的watcher來回調(diào)相應的處理邏
輯。

image.png
image.png

1.4 watcher的接口設計

Watcher是一個接口玩焰，任何實現(xiàn)了接口的類就是一個新的watcher由驹。watcher內(nèi)部包含了兩個枚舉類： KeeperState芍锚、EventType

image.png
image.png
image.png

image.png
image.png
package watcher;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

import java.util.concurrent.CountDownLatch;

public class ZkConnectionWatcher implements Watcher {
    final static CountDownLatch latch = new CountDownLatch(1);
    static ZooKeeper zk;

    public static void main(String[] args) {
        try {
            zk = new ZooKeeper("localhost:2181", 5000, new ZkConnectionWatcher());
            // 阻塞等待連接創(chuàng)建
            latch.countDown();
            // 會話id
            System.out.println(zk.getSessionId());

            // 驗證鑒權(quán)失敗的事件監(jiān)聽
            // 添加授權(quán)用戶
            zk.addAuthInfo("digest", "wh:12344".getBytes());
            byte[] res = zk.getData("/wh/node1", false, null);
            System.out.println(new String(res));
            Thread.sleep(10000);
            zk.close();
            System.out.println("all done...");
        } catch (Exception e) {

        }
    }

    public void process(WatchedEvent event) {
        try {
            // 監(jiān)聽連接事件類型
            if (event.getType() == Event.EventType.None) {
                if (event.getState() == Event.KeeperState.SyncConnected) {
                    System.out.println("conncetioned ...");
                    latch.countDown();
                } else if (event.getState() == Event.KeeperState.Disconnected) {
                    System.out.println("connection duakai...");
                } else if (event.getState() == Event.KeeperState.Expired) {
                    System.out.println("connection timeout");
                    zk = new ZooKeeper("localhost:2181", 5000, new ZkConnectionWatcher());

                } else if (event.getState() == Event.KeeperState.AuthFailed) {
                    System.out.println("認證失敗");
                }
            }
        } catch (Exception e) {

        }
    }
}
image.png
package set;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Id;
import org.apache.zookeeper.data.Stat;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

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

public class ZkSet1 {
    ZooKeeper zk;

    @Before
    public void before() throws IOException, InterruptedException {
        final CountDownLatch latch = new CountDownLatch(1);
        zk = new ZooKeeper("localhost:2181", 5000, new Watcher() {
            public void process(WatchedEvent event) {
                if (event.getState().equals(Event.KeeperState.SyncConnected)) {
                    System.out.println("connectioned ...");
                    latch.countDown();
                }
                System.out.println("path=" + event.getPath());
                System.out.println("eventType=" + event.getType());
            }
        });
        latch.await();
    }

    @After
    public void after() throws InterruptedException {
        zk.close();
    }

    @Test
    public void watcherExist1() throws KeeperException, InterruptedException {
        // arg1 : node path
        // arg2 : 使用連接對象里的watcher對象
        zk.exists("/wh/node1", true);
        Thread.sleep(10000);
        System.out.println("all done...");
    }

    @Test
    public void watcherExist2() throws KeeperException, InterruptedException {
        // arg1 : node path
        // arg2 : 自定義的watcher對象
        zk.exists("/wh/node1", new Watcher() {
            public void process(WatchedEvent event) {
                System.out.println("自定義的watcher...");
                System.out.println("path=" + event.getPath());
                System.out.println("eventType=" + event.getType());
            }
        });
        Thread.sleep(10000);
        System.out.println("all done...");
    }

    @Test
    public void watcherExist3() throws KeeperException, InterruptedException {
        // 驗證watcher是一次性的
        Watcher watcher = new Watcher() {
            public void process(WatchedEvent event) {
                System.out.println("自定義的watcher...");
                System.out.println("path=" + event.getPath());
                System.out.println("eventType=" + event.getType());
            }
        };
        // arg1 : node path
        // arg2 : 使用連接對象里的watcher對象
        zk.exists("/wh/node1", watcher);
        Thread.sleep(10000);
        System.out.println("all done...");
    }

    @Test
    public void watcherExist4() throws KeeperException, InterruptedException {
        // 驗證watcher是一次性的
        // 改進昔园，可用重復使用
        Watcher watcher = new Watcher() {
            public void process(WatchedEvent event) {
                System.out.println("自定義的watcher...");
                System.out.println("path=" + event.getPath());
                System.out.println("eventType=" + event.getType());
                try {
                    zk.exists("/wh/node1", this);
                } catch (KeeperException e) {
                    e.printStackTrace();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        // arg1 : node path
        // arg2 : 使用連接對象里的watcher對象
        zk.exists("/wh/node1", watcher);
        Thread.sleep(10000);
        System.out.println("all done...");
    }

    @Test
    public void watcherExist5() throws KeeperException, InterruptedException {
        // 注冊多個watcher
        zk.exists("/wh/node1", new Watcher() {
            public void process(WatchedEvent event) {
                System.out.println("自定義的watcher1...");
                System.out.println("path=" + event.getPath());
                System.out.println("eventType=" + event.getType());
            }
        });

        zk.exists("/wh/node1", new Watcher() {
            public void process(WatchedEvent event) {
                System.out.println("自定義的watcher2...");
                System.out.println("path=" + event.getPath());
                System.out.println("eventType=" + event.getType());
            }
        });
        Thread.sleep(10000);
        System.out.println("all done...");
    }

}
image.png
類似1.6.1 exists
image.png
image.png
類似1.6.1 exists
image.png
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import watcher.ZkConnectionWatcher;

import java.util.concurrent.CountDownLatch;

public class ZkConfigCenter implements Watcher  {

    final static CountDownLatch latch = new CountDownLatch(1);
    static ZooKeeper zk;

    private String url;
    private String name;
    private String pwd;



    public void process(WatchedEvent event) {
        try {
            // 監(jiān)聽連接事件類型
            if (event.getType() == Event.EventType.None) {
                if (event.getState() == Event.KeeperState.SyncConnected) {
                    System.out.println("conncetioned ...");
                    latch.countDown();
                } else if (event.getState() == Event.KeeperState.Disconnected) {
                    System.out.println("connection duakai...");
                } else if (event.getState() == Event.KeeperState.Expired) {
                    System.out.println("connection timeout");
                    zk = new ZooKeeper("localhost:2181", 5000, new ZkConnectionWatcher());

                } else if (event.getState() == Event.KeeperState.AuthFailed) {
                    System.out.println("認證失敗");
                } else if (event.getType() == Event.EventType.NodeDataChanged) {
                    initValue();
                }
            }
        } catch (Exception e) {

        }
    }


    public static void main(String[] args) throws InterruptedException {
        ZkConfigCenter configCenter = new ZkConfigCenter();
        for (int i=0;i<10;i++) {
            Thread.sleep(5000);
            System.out.println("url = " + configCenter.getUrl());
            System.out.println("name = " + configCenter.getName());
            System.out.println("pwd = " + configCenter.getPwd());
        }
    }

    public ZkConfigCenter() {
        initValue();
    }

    public void initValue() {
        try {
            zk = new ZooKeeper("localhost:2181", 5000, new ZkConnectionWatcher());
            String url = new String(zk.getData("/wh/config/url", true, null));
            String name = new String(zk.getData("/wh/config/name", true, null));
            String pwd = new String(zk.getData("/wh/config/pwd", true, null));
        } catch (Exception e) {

        }
    }




    public String getUrl() {
        return url;
    }

    public void setUrl(String url) {
        this.url = url;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public String getPwd() {
        return pwd;
    }

    public void setPwd(String pwd) {
        this.pwd = pwd;
    }
}
image.png
import org.apache.zookeeper.*;
import watcher.ZkConnectionWatcher;

import java.util.concurrent.CountDownLatch;

public class GlobalUniqueId implements Watcher {
    final static CountDownLatch latch = new CountDownLatch(1);
    static ZooKeeper zk;

    private String nodePath = "/uniqueid";

    public void process(WatchedEvent event) {
        try {
            // 監(jiān)聽連接事件類型
            if (event.getType() == Event.EventType.None) {
                if (event.getState() == Event.KeeperState.SyncConnected) {
                    System.out.println("conncetioned ...");
                    latch.countDown();
                } else if (event.getState() == Event.KeeperState.Disconnected) {
                    System.out.println("connection duakai...");
                } else if (event.getState() == Event.KeeperState.Expired) {
                    System.out.println("connection timeout");
                    zk = new ZooKeeper("localhost:2181", 5000, new ZkConnectionWatcher());

                } else if (event.getState() == Event.KeeperState.AuthFailed) {
                    System.out.println("認證失敗");
                } else if (event.getType() == Event.EventType.NodeDataChanged) {
                }
            }
        } catch (Exception e) {

        }
    }


    public GlobalUniqueId() {
        try {
            zk =  new ZooKeeper("localhost:2181", 5000, new ZkConnectionWatcher());
            latch.countDown();
        } catch (Exception ex) {

        }
    }

    public String getUniqueId() {
        String id = "";
        try {
            // 創(chuàng)建臨時有序節(jié)點
            id = zk.create(nodePath, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

            // /uniqueid000000001
            id = id.substring(9);
        } catch (Exception ex) {

        }
        return id;
    }

    public static void main(String[] args) {
        GlobalUniqueId globalUniqueId = new GlobalUniqueId();
        for (int i=0;i<10;i++) {
            System.out.println(globalUniqueId.getUniqueId());
        }
    }

}
image.png
import javafx.scene.SubScene;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * 利用臨時有序節(jié)點來實現(xiàn)分布式鎖
 */
public class ZkLock {

    // zk url
    String zkUrl = "localhost:2181";

    final CountDownLatch latch = new CountDownLatch(1);
    // zk 配置信息
    ZooKeeper zooKeeper;
    private static  final String LOCK_NODE_PATH = "/Locks";
    private static  final String LOCK_NODE_NAME = "/Lock_";
    private String lockPath;

    public ZkLock () {
        try {
            zooKeeper = new ZooKeeper(zkUrl, 5000, new Watcher() {
                public void process(WatchedEvent event) {
                    if (event.getType() == Event.EventType.None) {
                        System.out.println("connectioned ....");
                        latch.countDown();
                    }
                }
            });
            latch.await();
        } catch (Exception e) {

        }
    }

    // 獲取鎖
    public void tryAccquireLock() throws KeeperException, InterruptedException {
        // 創(chuàng)建鎖節(jié)點
        createLock();
        // 嘗試獲取鎖
        accquireLock();
    }

    // 創(chuàng)建鎖節(jié)點
    public void createLock() throws KeeperException, InterruptedException {
        // 1
        Stat stat = zooKeeper.exists(LOCK_NODE_PATH, false);
        if (stat == null) {
            zooKeeper.create(LOCK_NODE_PATH, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,
                    CreateMode.PERSISTENT);
        }
        // 2 創(chuàng)建臨時有序節(jié)點
        lockPath = zooKeeper.create(LOCK_NODE_PATH + "/" + LOCK_NODE_NAME,
                new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
    }

    // 監(jiān)視上一個節(jié)點是否被刪除
    Watcher watcher = new Watcher() {
        public void process(WatchedEvent event) {
            if (event.getType() == Event.EventType.NodeDeleted) {
                synchronized (this) {
                    notifyAll();
                }
            }
        }
    };


     // 嘗試獲取鎖
    public void accquireLock() throws InterruptedException, KeeperException {
        // 獲取Locks下的所有子節(jié)點
        List<String> list = zooKeeper.getChildren(LOCK_NODE_PATH, false);
        Collections.sort(list);
        // Locks/Lock_000000001
        int index = list.indexOf(lockPath.substring(LOCK_NODE_PATH.length()+1));
        if(index == 0) {
            System.out.println("拿到鎖");
        } else {
            // 上一個節(jié)點的路徑
            String path = list.get(index -1);
            Stat stat = zooKeeper.exists(LOCK_NODE_PATH + "/" + path, watcher);
            if (stat == null) {
                accquireLock();
            } else {
                synchronized (watcher) {
                    watcher.wait();
                }
                accquireLock();
            }
        }
    }

    // shifang suo
    public void releaseLock() throws InterruptedException, KeeperException {
        zooKeeper.delete(this.lockPath, -1);
        zooKeeper.close();
        System.out.println("鎖已經(jīng)釋放" + this.lockPath);
    }
}




import org.apache.zookeeper.KeeperException;
import org.omg.PortableInterceptor.SYSTEM_EXCEPTION;

/**
 * 測試分布式鎖
 */
public class TickSeller {
    private void sell() {
        System.out.println("sell begin...");
        // 線程隨機休眠5000，模擬現(xiàn)實中的費時操作
        int mils = 5000;
        try {
            // 代表復雜邏輯執(zhí)行
            Thread.sleep(mils);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("sell end...");
    }

    public void sellWithLock() throws KeeperException, InterruptedException {
        ZkLock lock= new ZkLock();
        // 獲取鎖
        lock.tryAccquireLock();
        sell();
        lock.releaseLock();
    }

    public static void main(String[] args) throws KeeperException, InterruptedException {
        TickSeller seller = new TickSeller();
        for (int i=0;i<10;i++) {
            seller.sellWithLock();
        }
    }
}

2 zk 集群搭建

image.png
image.png
image.png

3 一致性協(xié)議 Zab協(xié)議

image.png

image.png
image.png
image.png

4 zk的leader選舉

image.png

image.png

image.png

5 observer角色及其配置

image.png

6 Java API連接zk集群

image.png

image.png
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
  • 人面猴
    序言：七十年代末并炮，一起剝皮案震驚了整個濱河市默刚，隨后出現(xiàn)的幾起案子，更是在濱河造成了極大的恐慌逃魄，老刑警劉巖荤西，帶你破解...
    沈念sama閱讀 216,651評論 6贊 501
  • 死咒
    序言：濱河連續(xù)發(fā)生了三起死亡事件，死亡現(xiàn)場離奇詭異，居然都是意外死亡邪锌，警方通過查閱死者的電腦和手機勉躺，發(fā)現(xiàn)死者居然都...
    沈念sama閱讀 92,468評論 3贊 392
  • 救了他兩次的神仙讓他今天三更去死
    文/潘曉璐 我一進店門，熙熙樓的掌柜王于貴愁眉苦臉地迎上來觅丰，“玉大人饵溅，你說我怎么就攤上這事「咎眩” “怎么了蜕企？”我有些...
    開封第一講書人閱讀 162,931評論 0贊 353
  • 道士緝兇錄：失蹤的賣姜人
    文/不壞的土叔 我叫張陵，是天一觀的道長冠句。 經(jīng)常有香客問我轻掩，道長，這世上最難降的妖魔是什么懦底？ 我笑而不...
    開封第一講書人閱讀 58,218評論 1贊 292
  • ?港島之戀（遺憾婚禮）
    正文 為了忘掉前任唇牧，我火速辦了婚禮，結(jié)果婚禮上基茵，老公的妹妹穿的比我還像新娘奋构。我一直安慰自己，他們只是感情好拱层，可當我...
    茶點故事閱讀 67,234評論 6贊 388
  • 惡毒庶女頂嫁案：這布局不是一般人想出來的
    文/花漫 我一把揭開白布弥臼。 她就那樣靜靜地躺著，像睡著了一般根灯。 火紅的嫁衣襯著肌膚如雪径缅。 梳的紋絲不亂的頭發(fā)上，一...
    開封第一講書人閱讀 51,198評論 1贊 299
  • 城市分裂傳說
    那天烙肺，我揣著相機與錄音纳猪，去河邊找鬼。 笑死桃笙，一個胖子當著我的面吹牛氏堤，可吹牛的內(nèi)容都是我干的。 我是一名探鬼主播搏明，決...
    沈念sama閱讀 40,084評論 3贊 418
  • 雙鴛鴦連環(huán)套：你想象不到人心有多黑
    文/蒼蘭香墨 我猛地睜開眼鼠锈，長吁一口氣：“原來是場噩夢啊……” “哼！你這毒婦竟也來了星著？” 一聲冷哼從身側(cè)響起购笆，我...
    開封第一講書人閱讀 38,926評論 0贊 274
  • 萬榮殺人案實錄
    序言：老撾萬榮一對情侶失蹤，失蹤者是張志新（化名）和其女友劉穎虚循，沒想到半個月后同欠，有當?shù)厝嗽跇淞掷锇l(fā)現(xiàn)了一具尸體样傍，經(jīng)...
    沈念sama閱讀 45,341評論 1贊 311
  • ?護林員之死
    正文 獨居荒郊野嶺守林人離奇死亡，尸身上長有42處帶血的膿包…… 初始之章·張勛 以下內(nèi)容為張勛視角 年9月15日...
    茶點故事閱讀 37,563評論 2贊 333
  • ?白月光啟示錄
    正文 我和宋清朗相戀三年铺遂，在試婚紗的時候發(fā)現(xiàn)自己被綠了衫哥。 大學時的朋友給我發(fā)了我未婚夫和他白月光在一起吃飯的照片。...
    茶點故事閱讀 39,731評論 1贊 348
  • 活死人
    序言：一個原本活蹦亂跳的男人離奇死亡襟锐，死狀恐怖炕檩，靈堂內(nèi)的尸體忽然破棺而出，到底是詐尸還是另有隱情捌斧，我是刑警寧澤笛质，帶...
    沈念sama閱讀 35,430評論 5贊 343
  • ?日本核電站爆炸內(nèi)幕
    正文 年R本政府宣布，位于F島的核電站捞蚂，受9級特大地震影響妇押，放射性物質(zhì)發(fā)生泄漏。R本人自食惡果不足惜姓迅，卻給世界環(huán)境...
    茶點故事閱讀 41,036評論 3贊 326
  • 男人毒藥：我在死后第九天來索命
    文/蒙蒙 一敲霍、第九天 我趴在偏房一處隱蔽的房頂上張望。 院中可真熱鬧丁存，春花似錦肩杈、人聲如沸。這莊子的主人今日做“春日...
    開封第一講書人閱讀 31,676評論 0贊 22
  • 一樁弒父案扩然，背后竟有這般陰謀
    文/蒼蘭香墨 我抬頭看了看天上的太陽。三九已至聋伦，卻和暖如春夫偶，著一層夾襖步出監(jiān)牢的瞬間，已是汗流浹背觉增。 一陣腳步聲響...
    開封第一講書人閱讀 32,829評論 1贊 269
  • 情欲美人皮
    我被黑心中介騙來泰國打工兵拢， 沒想到剛下飛機就差點兒被人妖公主榨干…… 1. 我叫王不留，地道東北人逾礁。 一個月前我還...
    沈念sama閱讀 47,743評論 2贊 368
  • 代替公主和親
    正文 我出身青樓说铃，卻偏偏與公主長得像，于是被迫代替她去往敵國和親嘹履。 傳聞我的和親對象是個殘疾皇子腻扇，可洞房花燭夜當晚...
    茶點故事閱讀 44,629評論 2贊 354