2. 程式人生 > >ZooKeeper系列(9)：ZooKeeper實現分布式Barrier和Queue

ZooKeeper系列(9)：ZooKeeper實現分布式Barrier和Queue

阿新 • • 發佈：2019-01-20
nod zookeeper instant zook conf protected tint 說了 this

1. 快速開始

1.1概述：

Zookeeper是Hadoop的一個子項目，它是分布式系統中的協調系統，可提供的服務主要有：配置服務、名字服務、分布式同步、組服務等。
1.2 使用常見

1.2.1 統一配置

把配置放在ZooKeeper的節點中維護，當配置變更時，客戶端可以收到變更的通知，並應用最新的配置。

1.2.2，集群管理

集群中的節點，創建ephemeral的節點，一旦斷開連接，ephemeral的節點會消失，其它的集群機器可以收到消息。

1.2.3 分布式鎖

多個客戶端發起節點創建操作，只有一個客戶端創建成功，從而獲得鎖。

1.3 安裝和配置

通過官方下載鏈接zookeeper 進行下載，解壓後進入conf目錄，新建一個zoo.conf文件，配置內容如下：

tickTime=2000    
dataDir=/Users/lsq/Documents/zookeeper/zookeeper0/data
dataLogDir=/Users/lsq/Documents/zookeeper/zookeeper0/dataLog
clientPort=4399
initLimit=5
syncLimit=2

tickTime: ZooKeeper基本時間單位（ms）
initLimit: 指定了啟動zookeeper時，zookeeper實例中的隨從實例同步到領導實例的初始化連接時間限制，超出時間限制則連接失敗（以tickTime為時間單位）；
syncLimit: 指定了zookeeper正常運行時，主從節點之間同步數據的時間限制，若超過這個時間限制，那麽隨從實例將會被丟棄

dataDir： zookeeper存放數據的目錄；
clientPort： 用於連接客戶端的端口

接下來進入bin目錄啟動ZooKeeper實例以及客戶端連接：

./zkServer.sh start
./zkCli.sh -server localhost:4399

接下來看看集群如何配置，其實跟單機差不多，這裏我們把剛剛下載的Zookeeper復制多兩份，一共是三個，配置信息如下：

tickTime=2000    
dataDir=/Users/lsq/Documents/zookeeper/zookeeper0/data
dataDir=/Users/lsq/Documents/zookeeper/zookeeper0/dataLog
clientPort
 
=4399
initLimit=5
syncLimit=2
server.1=127.0.0.1:8880:9990
server.2=127.0.0.1:8881:9991
server.3=127.0.0.1:8882:9992

三個文件夾下面的zoo.conf都是這個格式，需要修改dataDir，dataDir，clientPort，
然後在dataDir所指向的目錄下面新建一個myid文件，對應server.x，比如第一個文件夾下面的myid就填入一個1，第二個就填入一個2，以此類推。接著依次啟動即可。可以采用下面的命令

echo "1" > myid

2.使用java來操作ZooKeeper實例
一門技術最重要的就算實戰了，接下來的內容將圍繞這一部分來講。
首先是Znode的創建和刪除
Znode有兩種類型：短暫的和持久的。短暫的znode在創建的客戶端與服務器端斷開（無論是明確的斷開還是故障斷開）連接時，該znode都會被刪除；相反，持久的znode則不會

技術分享圖片 
public class CreateGroup implements Watcher {
    //會話延時
    private static final int SESSION_TIMEOUT = 1000;
    //zk對象
    private ZooKeeper zk = null;
    //同步計數器
    private CountDownLatch countDownLatch = new CountDownLatch(1);
    //客戶端連接到服務器時會觸發觀察者進行調用
    public void process(WatchedEvent event) {
        if(event.getState() == KeeperState.SyncConnected){
            countDownLatch.countDown();//計數器減一
        }
    }

    public void connect(String hosts) throws IOException, InterruptedException {
        zk = new ZooKeeper(hosts, SESSION_TIMEOUT, this);
        countDownLatch.await();//阻塞程序繼續執行
    }
    //創建GROUP
    public void create(String groupName) throws KeeperException, InterruptedException{
        String path = "/" + groupName;
        //允許任何客戶端對該znode進行讀寫,以及znode進行持久化
        String createPath = zk.create(path, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        System.out.println("Created "+createPath);
    }
    //關閉zk
    public void close() throws InterruptedException{
        if(zk != null){
            try {
                zk.close();
            } catch (InterruptedException e) {
                throw e;
            }finally{
                zk = null;
                System.gc();
            }
        }
    }

    //測試主類
    public static void main(String args[]){
        String host = "127.0.0.1:4399";
        String groupName = "test";
        CreateGroup createGroup = new CreateGroup();
        try {
            createGroup.connect(host);
            createGroup.create(groupName);
            createGroup.close();
            createGroup = null;
            System.gc();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }   
    }
}
View Code

接下來把創建和銷毀分離出來作為一個獨立的類,以後相關操作可以直接使用

技術分享圖片 
public class ConnetctionWatcher implements Watcher {

    private static final int SESSION_TIMEOUT = 5000;

    protected ZooKeeper zk = null;
    private CountDownLatch countDownLatch = new CountDownLatch(1);

    public void process(WatchedEvent event) {
        KeeperState state = event.getState();

        if(state == KeeperState.SyncConnected){
            countDownLatch.countDown();
        }
    }
    public void connection(String hosts) throws IOException, InterruptedException {
        zk = new ZooKeeper(hosts, SESSION_TIMEOUT, this);
        countDownLatch.await();
    }
    public void close() throws InterruptedException {
        if (null != zk) {
            try {
                zk.close();
            } catch (InterruptedException e) {
                throw e;
            }finally{
                zk = null;
                System.gc();
            }
        }
    }
}
View Code

接下來我們看看節點如何刪除

技術分享圖片 
public class DeleteGroup extends ConnetctionWatcher {
    public void delete(String groupName) {
        String path = "/" + groupName;

        try {
            List<String> children = zk.getChildren(path, false);

            for(String child : children){
                zk.delete(path + "/" + child, -1);
            }
            zk.delete(path, -1);//版本號為-1，
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
View Code

3. 利用java實現分布式Barrier
Barrier是一種控制和協調多個任務觸發次序的機制。簡單來說就是用一個屏障把將要執行的任務攔住，等待所有任務都處於可運行狀態才放開屏障，其實在單機上我們可以利用CyclicBarrier來實現這個機制，但是在分布式環境下，我們可以利用ZooKeeper可以派上用場，我們可以利用一個Node來作為Barrier的實體，然後要Barrier的任務通過調用exists檢測是否Node存在，當需要打開Barrier時候，刪除這個Node，這樣ZooKeeper的watch機制會通知到各個任務可以開始執行。接下來看代碼：

技術分享圖片 
public class Barrier extends SyncPrimitive {
    int size;
    String name;

    Barrier(String address, String root, int size) {
        super(address);
        this.root = root;
        this.size = size;
        //創建Barrier的Node
        if (zk != null) {
            try {
                Stat s = zk.exists(root, false);
                if (s == null) {
                    zk.create(root, new byte[0], Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
                }
            } catch (KeeperException e) {
                System.out.println("Keeper exception when instantiating queue: " + e.toString());
            } catch (InterruptedException e) {
                System.out.println("Interrupted exception");
            }
        }
        try {
            name = new String(InetAddress.getLocalHost().getCanonicalHostName().toString());
        } catch (UnknownHostException e) {
            System.out.println(e.toString());
        }

    }

    /**
     * 加入Barrier等待
     */

    boolean enter() throws KeeperException, InterruptedException{
        zk.create(root + "/" + name, new byte[0], Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);
        while (true) {
            synchronized (mutex) {
                List<String> list = zk.getChildren(root, true);
                if (list.size() < size) {
                    mutex.wait();
                } else {
                    return true;
                }
            }
        }
    }

    /**
     * 一直等待知道指定數量節點到達
     */

    boolean leave() throws KeeperException, InterruptedException{
        zk.delete(root + "/" + name, 0);
        while (true) {
            synchronized (mutex) {
                List<String> list = zk.getChildren(root, true);
                    if (list.size() > 0) {
                        mutex.wait();
                    } else {
                        return true;
                    }
                }
            }
    }
}
View Code

父類代碼如下

技術分享圖片 
public class SyncPrimitive implements Watcher {
    static ZooKeeper zk = null;
    static Integer mutex;
    //根節點
    String root;
    SyncPrimitive(String address) {
        if(zk == null){
            try {
                System.out.println("Starting ZK:");
                zk = new ZooKeeper(address, 3000, this);
                mutex = new Integer(-1);
                System.out.println("Finished starting ZK: " + zk);
            } catch (IOException e) {
                System.out.println(e.toString());
                zk = null;
            }
        }
        //else mutex = new Integer(-1);
    }

    synchronized public void process(WatchedEvent event) {
        synchronized (mutex) {
            System.out.println("Process: " + event.getType());
            mutex.notify();
        }
    }

    public static void queueTest(String args[]) {
        Queue q = new Queue(args[1], "/app1");

        System.out.println("Input: " + args[1]);
        int i;
        Integer max = new Integer(args[2]);

        if (args[3].equals("p")) {
            System.out.println("Producer");
            for (i = 0; i < max; i++)
                try{
                    q.produce(10 + i);
                } catch (KeeperException e){

                } catch (InterruptedException e){

                }
        } else {
            System.out.println("Consumer");

            for (i = 0; i < max; i++) {
                try{
                    int r = q.consume();
                    System.out.println("Item: " + r);
                } catch (KeeperException e){
                    i--;
                } catch (InterruptedException e){

                }
            }
        }
    }

    public static void barrierTest(String args[]) {
        Barrier b = new Barrier(args[1], "/b1", new Integer(args[2]));
        try{
            boolean flag = b.enter();
            System.out.println("Entered barrier: " + args[2]);
            if(!flag) System.out.println("Error when entering the barrier");
        } catch (KeeperException e){

        } catch (InterruptedException e){

        }
        Random rand = new Random();
        int r = rand.nextInt(100);
        for (int i = 0; i < r; i++) {
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {

            }
        }
        try{
            b.leave();
        } catch (KeeperException e){
        } catch (InterruptedException e){
        }
        System.out.println("Left barrier");
    }
    //測試用的主類
    public static void main(String args[]) {
        /*
        args =new String[] {"qTest","localhost:4399","3","c"};
        if (args[0].equals("qTest"))
            queueTest(args);
        else
            barrierTest(args);
         */
    }
}
View Code

4. 分布式隊列（Queue）
在分布式環境下，實現Queue需要高一致性來保證，那麽我們可以這樣來設計。把一個Node當成一個隊列，然後children用來存儲內容，利用ZooKeeper提供的順序遞增的模式（會自動在name後面加入一個遞增的數字來插入新元素）。於是在offer時候我們可以使用create，take時候按照順序把children第一個delete就可以了。ZooKeeper保證了各個server上數據是一致的。廢話不多說了，直接看代碼

技術分享圖片 
/**
 * 一個消費者－生產者模式的消息隊列
 */
public class Queue extends SyncPrimitive {

    Queue(String address, String name) {
        super(address);
        this.root = name;
        if (zk != null) {
            try {
                Stat s = zk.exists(root, false);
                if (s == null) {
                    zk.create(root, new byte[0], Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
                }
            } catch (KeeperException e) {
                System.out.println("Keeper exception when instantiating queue: " + e.toString());
            } catch (InterruptedException e) {
                System.out.println("Interrupted exception");
            }
        }
    }

    /**
     * 隊列中插入數據
     */

    boolean produce(int i) throws KeeperException, InterruptedException{
        ByteBuffer b = ByteBuffer.allocate(4);
        byte[] value;

        b.putInt(i);
        value = b.array();
        zk.create(root + "/element", value, Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT_SEQUENTIAL);

        return true;
    }


    /**
     * 把元素從隊列中移除
     */
    int consume() throws KeeperException, InterruptedException{
        int retvalue = -1;
        Stat stat = null;

        //得到現在隊列中首個可用的節點
        while (true) {
            synchronized (mutex) {
                List<String> list = zk.getChildren(root, true);
                if (list.size() == 0) {
                    System.out.println("Going to wait");
                    mutex.wait();
                } else {
                    Integer min = new Integer(list.get(0).substring(7));
                    for(String s : list){
                        Integer tempValue = new Integer(s.substring(7));
                        //System.out.println("Temporary value: " + tempValue);
                        if(tempValue < min) min = tempValue;
                    }
                    System.out.println("Temporary value: " + root + "/element" + min);
                    byte[] b = zk.getData(root + "/element" + min, false, stat);
                    zk.delete(root + "/element" + min, 0);
                    ByteBuffer buffer = ByteBuffer.wrap(b);
                    retvalue = buffer.getInt();

                    return retvalue;
                }
            }
        }
    }
}
View Code

ZooKeeper系列(9)：ZooKeeper實現分布式Barrier和Queue

相關推薦

ZooKeeper系列(9)：ZooKeeper實現分布式Barrier和Queue

nod zookeeper instant zook conf protected tint 說了 this 1. 快速開始 1.1概述： Zookeeper是Hadoop的一個子項目，它是分布式系統中的協調系統，可提供的服務主要有：配置服務、名字服務、分布式同步、組服

阿裏JAVA面試題剖析：一般實現分布式鎖都有哪些方式？使用 Redis 如何設計分布式鎖？

自己 ini 單位 nts ast 客戶端 this 失敗 獲取 面試原題 一般實現分布式鎖都有哪些方式？使用 redis 如何設計分布式鎖？使用 zk 來設計分布式鎖可以嗎？這兩種分布式鎖的實現方式哪種效率比較高？ 面試官心理分析 其實一般問問題，都是這麽問的，先問問你

go任務調度9（op實現分布式樂觀鎖）

eat range roc cee shadow 上鎖 import package urn package main import ( "go.etcd.io/etcd/clientv3" "time" "fmt" "context" )

基於zookeeper簡單實現分布式鎖

lean bool 思想 common factor exists play nec try 這裏利用zookeeper的EPHEMERAL_SEQUENTIAL類型節點及watcher機制。來簡單實現分布式鎖。 主要思想： 1、開啟10個線程。在disLocks節點下

Zookeeper實現分布式集群監控

rpath 集群 用戶 mave 0ms unsafe 2個 his void Zookeeepr實現分布式集群監控 Zookeeper中節點有兩種：臨時節點和永久節點 從類型上看節點又可以分為四種節點類型：PERSIST，PERSIST_SEQUENTIAL,EPHEME

ZooKeeper實現分布式鎖

data tag 找到 ont 控制 ear create rac oct 全部代碼下載：CSDN鏈接：CSDN鏈接;寫文章不易，歡迎大家采我的文章，以及給出有用的評論，當然大家也可以關註一下我的github；多謝； 1.分布式鎖的由來： 在程序開發過程中不得不考慮的就

zookeeper學習實踐1-實現分布式鎖

ont 技術 start ria 開放 有序 stat throws edev 引言 ZooKeeper是一個分布式的，開放源碼的分布式應用程序協調服務，是Google的Chubby一個開源的實現，是Hadoop和Hbase的重要組件。它是一個為分布式應用提供一致性服務的軟

ZooKeeper連接並創建節點以及實現分布式鎖操作節點排序輸出最小節點Demo

false bytes roc 永遠 err index public nal kde class LockThread implements Runnable { private DistributedLock lock; public Lo

【原創】redis庫存操作，分布式鎖的四種實現方式[連載一]--基於zookeeper實現分布式鎖

zookeepe operation iat 並發 method logger 方案 nag 概念 一、背景 在電商系統中，庫存的概念一定是有的，例如配一些商品的庫存，做商品秒殺活動等，而由於庫存操作頻繁且要求原子性操作，所以絕大多數電商系統都用Redis來實現庫存的加減，

基於zookeeper實現分布式鎖

ber mave and bool 就是 tar 資源 service() 分布式系 什麽是分布式鎖 分布式鎖一般用在分布式系統或者多個應用中，用來控制同一任務是否執行或者任務的執行順序。在項目中，部署了多個tomcat應用，在執行定時任務時就會遇到同一任務可能執行多次的情

zookeeper 實現分布式鎖安全用法

() 之間 void 寫入 機房 串行化 client .data 無法 背景 ConnectionLoss 鏈接丟失 SessionExpired 會話過期 繞開 zookeeper broker 進行狀態通知 leader 選舉與zkNode 斷開 做好冪等 靜態擴容、

利用Zookeeper實現分布式鎖及服務註冊中心

建立連接 oid 等待 某個文件 改名 解壓 als 中文字符 blog 原文:利用Zookeeper實現分布式鎖及服務註冊中心對於Zookeeper的定義以及原理，網上已經有很多的優秀文章對其進行了詳細的介紹，所以本文不再進行這方面的闡述。 本文主要介紹一些基本的準備工

zookeeper — 實現分布式鎖

lean sta 會話期 競爭 all zookeeper scribe handle 獲取鎖 一.前言 在之前的文章中介紹過分布式鎖的特點和利用Redis實現簡單的分布式鎖。但是分布式鎖的實現還有很多其他方式，但是萬變不離其宗，始終遵循一個特點：同一時刻只能有一個操作獲取

基於zookeeper實現分布式鎖（續）

src 9.png 分享 per 分享圖片 技術分享 zookeeper 圖片 http 測試代碼： 效果圖： 基於zookeeper實現分布式鎖（續）

精華【分布式、微服務、雲架構、dubbo+zookeeper+springmvc+mybatis+shiro+redis】分布式大型互聯網企業架構！

net ios 系統數據庫 權限分配 容器 移動 activit str 重復 平臺簡介 Jeesz是一個分布式的框架，提供項目模塊化、服務化、熱插拔的思想，高度封裝安全性的Java EE快速開發平臺。 Jeesz本身集成Dubbo服務管控、

精華【分布式微服務雲架構dubbo+zookeeper+springmvc+mybatis+shiro+redis】分布式大型互聯網企業架構！

數據字典 https sql log redmine websphere 搜索 pri 數據庫表 摘要: 開發工具 1.Eclipse IDE：采用Maven項目管理，模塊化。 2.代碼生成：通過界面方式簡單配置，自動生成相應代碼，目前包括三種生成方式（增刪改查）：單表、

DCS實踐幹貨：使用Redis實現分布式鎖

需要 icloud CA 互聯 論壇 讀寫 無序 color log 場景介紹 很多互聯網場景（如商品秒殺，論壇回帖蓋樓等），需要用加鎖的方式，以對某種資源進行順序訪問控制。如果應用服務集群部署，則涉及到對分布式應用加鎖。當前分布式加鎖主要有三種方式：（磁盤）數據庫、緩存數

zookeeper系列之：zookeeper簡介淺談

一、zookeeper的定義 　　開啟zookeeper官網，赫然一行大字，寫著：“Apache ZooKeeper致力於開發和維護實現高度可靠的分散式協調的開源伺服器”。什麼意思呢？就是Apache ZooKeeper的目標是開發和維護開源伺服器，這伺服器是幹什麼的呢？是做分散式協調的。這伺服器的特點是什麼

[zookeeper]zookeeper系列七：zookeeper選舉及資料一致性

ZAB協議 ZAB(Zookeeper Atomic Broadcast)協議，即Zookeeper原子訊息廣播協議，協議內容大致如下： 所有事物的請求必須由全域性唯一的伺服器來協調處理，這樣的伺服器

[zookeeper]zookeeper系列三：zookeeper中watcher的使用及原理

watcher解決的問題 在進入watcher之前我們先試想在應用伺服器叢集中可能存在的兩個問題： 因為叢集中有很多機器，當某個通用的配置發生變化後，怎麼讓自動的讓所有伺服器的配置統一生效？ 當叢集中某個節點宕機，如何讓叢集中的其他節點知道？ 為了解決