2. 程式人生 > >Zookeeper客戶端Curator使用指南

Zookeeper客戶端Curator使用指南

阿新 • • 發佈：2018-12-14

what is Curator ?

Curator是zookeeper分散式協調服務的java客戶端庫，它包裝了一系列操作zk的高階API和實用庫，是的操作zk變得更加容易和可靠。例如使用原生zk的API實現分散式鎖的話，程式碼量多，複雜，使用Curator後就相對簡單的多，很多底層的api都直接封裝好了，開箱即用，學習成本低。

Getting Started

1、使用Curator之前，你需要引入maven依賴

<dependency>
   <groupId>org.apache.curator</groupId>
   <artifactId>curator-framework</artifactId>
   <version>2.8.0</version>
</dependency>

<dependency>
   <groupId>org.apache.curator</groupId>
   <artifactId>curator-recipes</artifactId>
   <version>2.8.0</version>
</dependency>

2、例項化Curator，你可以通過CuratorFrameworkFactory類提供的來產生一個CuratorFramework物件

CuratorFrameworkFactory.newClient(zookeeperConnectionString, retryPolicy)

zookeeperConnectionString就是連線的ip:埠資訊，retryPolicy是重試策略，Curator提供了三種常用的重試策略，這裡不詳述

RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3)
CuratorFramework client = CuratorFrameworkFactory.newClient(zookeeperConnectionString, retryPolicy);
client.start();

或者你也可以使用鏈式呼叫來例項化curator

CuratorFramework curatorFramework = CuratorFrameworkFactory.builder()
        .connectString(zookeeperConnectionString)
        .sessionTimeoutMs(5000)
        .connectionTimeoutMs(5000)
        .retryPolicy(retryPolicy)
        .build();

curatorFramework.start();

How to use Curator API operate ZK ?

curator操作zk的api主要包括 節點的增刪改查、節點判斷、節點監聽等，下面的程式碼演示瞭如何使用基本的curator api

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.retry.RetryUntilElapsed;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
* 類描述：zookeeper客戶端curator使用demo
*/
public class CuratorDemo {

    private static final Logger logger = LoggerFactory.getLogger(CuratorDemo.class);

    private static final String NODE_PATH = "/node_8";
    private static final String CONNECT_TOSTRING = "10.200.121.46:2181";

    /*建立執行緒池，供給非同步使用curator時呼叫*/
    public static ExecutorService executorService = Executors.newCachedThreadPool();

    public static void main(String[] args) throws Exception {
        try {
        
/*重試策略一：重試三次，每重試一次，重試的間隔時間會越來越大

        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
*/
/*重試策略二：最多重試三次，每次重試間隔1s

        RetryPolicy retryPolicy1 = new RetryNTimes(3,1000);
*/

/*重試策略三：最大重試時間總和不超過5s，每次重試間隔為1s*/
            RetryPolicy retryPolicy2 = new RetryUntilElapsed(5000, 1000);

/*
        
/*      方式一建立zookeeper連線
        CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(CONNECT_TOSTRING,5000,5000,retryPolicy2);
*/

            /*方式二建立zookeeper連線*/
            CuratorFramework curatorFramework = CuratorFrameworkFactory.builder()
                    .connectString(CONNECT_TOSTRING)
                    .sessionTimeoutMs(5000)
                    .connectionTimeoutMs(5000)
                    .retryPolicy(retryPolicy2)
                    .build();

            curatorFramework.start();

        /*建立節點資料*/        
 curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(NODE_PATH, "456".getBytes());

        /*刪除節點（包含子節點）*/            curatorFramework.delete().guaranteed().deletingChildrenIfNeeded().withVersion(-1).forPath(NODE_PATH);

        /*獲取子節點*/
        List<String> strings = curatorFramework.getChildren().forPath(NODE_PATH);

        /*獲取節點資料內容*/
        byte[] bytes = curatorFramework.getData().forPath(NODE_PATH);
        System.out.println(new String(bytes));
        
       /*獲取節點資料內容+狀態資訊*/
       Stat stat = new Stat();
        byte[] result = curatorFramework.getData().storingStatIn(stat).forPath(NODE_PATH);
        System.out.println(new String(result));

        /*修改節點資料內容*/
        curatorFramework.setData().forPath(NODE_PATH, "123".getBytes());

       /*判斷節點是否存在*/
        Stat stat1 = curatorFramework.checkExists().forPath(NODE_PATH);

        /*非同步操作，以判斷節點是否存在為例，注意使用執行緒池以便節省單個執行緒的建立銷燬開銷，及最後執行緒的關閉*/
        curatorFramework.checkExists().inBackground(new BackgroundCallback() {
                @Override
                public void processResult(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception {

                    Object context = curatorEvent.getContext();   //這裡的上下文就是 傳遞進去的"123456"

                }
            }, "12345", executorService).forPath(NODE_PATH);

          /*設定節點事件監聽*/
          final NodeCache nodeCache = new NodeCache(curatorFramework, NODE_PATH);
            nodeCache.start();
            nodeCache.getListenable().addListener(new NodeCacheListener() {
                @Override
                public void nodeChanged() throws Exception {
                    byte[] result = nodeCache.getCurrentData().getData();
                    logger.info("事件監聽result=" + new String(result));
                }
            });

            /*設定子節點事件監聽*/
            final PathChildrenCache childrenCache = new PathChildrenCache(curatorFramework, NODE_PATH, true);
            childrenCache.start();
            childrenCache.getListenable().addListener(new PathChildrenCacheListener() {
                @Override
                public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                    PathChildrenCacheEvent.Type type = pathChildrenCacheEvent.getType();
                    switch (type) {
                        case CHILD_ADDED:
                            logger.info("");
                        case CHILD_UPDATED:
                            logger.info("");
                        case CHILD_REMOVED:
                            logger.info("");
                        default:
                            break;
                    }
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            executorService.shutdown();
        }
    }

}

使用Curator實現Master選舉

Curator中包裝的master選舉包含兩種，Leader Latch和Leader Election，原理採用的是zk的節點特性，即多個客戶端同時建立同一節點，zk保證只有一個客戶端能建立成功，成功的客戶端即為master節點，再master節點的機器上執行業務。Leader  Latcher裡面即包裝了zk建立節點、設定監聽，對應的操作均包裝在LeaderLatch類中。LeaderLatch將隨機選出一個master

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.leader.LeaderLatch;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.springframework.context.annotation.Lazy;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

/**
* 類描述：基於Curator的Leader Latch實現的master選舉
* 建立人：simonsfan
*/
@Component
public class CuratorLeaderLatch {

    private static CuratorFramework curatorFramework;

    private static LeaderLatch leaderLatch;

    private static final String path = "/root/leaderlatch";

    private static final String connectStr = "10.200.121.46:2181,10.200.121.159:2181,10.200.121.168:2181";

    static {
        curatorFramework = CuratorFrameworkFactory
                .builder()
                .connectString(connectStr)
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .connectionTimeoutMs(5000)
                .sessionTimeoutMs(5000)
                .build();

        curatorFramework.start();
        leaderLatch = new LeaderLatch(curatorFramework, path.concat("/testtast"));
    }

    @Lazy
    @Scheduled(cron = "")
    public void testTask() {
        //是master節點的執行業務流程，使用leaderLatch.hasLeadership()方法判斷是否為leader，true表示是master節點
        if (!leaderLatch.hasLeadership()) return;
        //TODO something

    }
    
}

使用Curator實現分散式鎖

這裡講的的是Shared Reentrant Lock（共享可重入鎖，推薦使用，Curator還封裝了其他型別的鎖：共享不可重入鎖之類的）：全域性同步的、公平的分散式共享重入式鎖，可保證在任意同一時刻，只有一個客戶端持有鎖。使用到的類是InterProcessMutex

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
* 類描述：Curator實現的分散式鎖
* 建立人：simonsfan
*/
public class DistributedLock {

    private static CuratorFramework curatorFramework;

    private static InterProcessMutex interProcessMutex;

    private static final String connectString = "10.200.121.46:2181,10.200.121.43:2181,10.200.121.167:2181";

    private static final String root = "/root";

    private static ExecutorService executorService;

    private String lockName;

    public String getLockName() {
        return lockName;
    }

    public void setLockName(String lockName) {
        this.lockName = lockName;
    }

    static {
        curatorFramework = CuratorFrameworkFactory.builder().connectString(connectString).connectionTimeoutMs(5000).sessionTimeoutMs(5000).retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();
        executorService = Executors.newCachedThreadPool();
        curatorFramework.start();
    }

    public DistributedLock(String lockName) {
        this.lockName = lockName;
        interProcessMutex = new InterProcessMutex(curatorFramework, root.concat(lockName));
    }

    /*上鎖*/
    public void tryLock() {
        int count = 0;
        try {
            while (!interProcessMutex.acquire(1, TimeUnit.SECONDS)) {
                count++;
                if (count > 3) {
                    break;
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /*釋放*/
    public void releaseLock() {
        try {
            if (interProcessMutex != null) {
                interProcessMutex.release();
            }
            curatorFramework.delete().inBackground(new BackgroundCallback() {
                @Override
                public void processResult(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception {

                }
            }, executorService).forPath(root.concat(lockName));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}

相關推薦

Zookeeper客戶端Curator使用指南

what is Curator ? Curator是zookeeper分散式協調服務的java客戶端庫，它包裝了一系列操作zk的高階API和實用庫，是的操作zk變得更加容易和可靠。例如使用原生zk的API實現分散式鎖的話，程式碼量多，複雜，使用Curator

7.5 zookeeper客戶端curator的基本使用

serve server 超時 one c-c tlist result 強制 car 使用zookeeper原生API實現一些復雜的東西比較麻煩。所以，出現了兩款比較好的開源客戶端，對zookeeper的原生API進行了包裝：zkClient和curator。後者是Net

zookeeper客戶端curator簡易使用

zookeeper客戶端curator簡易使用 寫在前面：目前Curator有2.x.x和3.x.x兩個系列的版本，支援不同版本的Zookeeper。其中Curator 2.x.x相容Zookeeper的3.4.x和3.5.x。而Curator 3.x.x只相容Zookeeper

Zookeeper(八)-zookeeper客戶端Curator

zookeeper的原生api相對來說比較繁瑣，比如：對節點新增監聽事件，當監聽觸發後，我們需要再次手動新增監聽，否則監聽只生效一次；再比如，斷線重連也需要我們手動程式碼來判斷處理等等。對於curator的介紹，從網上百度了一段：Curator是Netflix開源的一套zoo

Zookeeper客戶端Curator使用詳解

前提 最近剛好用到了zookeeper，做了一個基於SpringBoot、Curator、Bootstrap寫了一個視覺化的Web應用： 歡迎使用和star。 簡介 Curator是Netflix公司開源的一套zookeeper客戶端框架，解決了很多Zookeep

ZooKeeper客戶端curator元件介紹

Curator framework提供了高階API， 極大的簡化了ZooKeeper的使用。 它在ZooKeeper基礎上增加了很多特性，可以管理與ZOoKeeper的連線和重試機制。這些特性包括： 自動連線管理 ** 有些潛在的錯誤情況需要讓ZooKeeper client重建連線和重試。Curato

Zookeeper客戶端Curator使用介紹

簡介 Curator是Netflix公司開源的一套zookeeper客戶端框架，解決了很多Zookeeper客戶端非常底層的細節開發工作，包括連線重連、反覆註冊Watcher和NodeExistsException異常等等。Patrixck Hunt（Zook

Zookeeper分散式及客戶端Curator的API簡單使用

最近公司專案中使用了分散式Zookeeper及Dubbo，為了弄清楚這些框架在專案中的使用，在我業餘時間中學習了一些Zookeeper的簡單用法，分享出來，若有不足之處，望大家給與建議...... 一、什麼是分散式系統？ 我的理解：將原有的系統拆分為多個子系統組成一個龐大的系統，這個龐大

zookeeper開源客戶端Curator介紹(六)

上一篇文章介紹了zookeeper原生API的使用,使用過原生API不得不說,有很多的問題,比如：不能遞迴建立和刪除節點、Watcher只能使用一次、還有很多可以解決分散式應用問題的api(比如分散式鎖,leader選舉等)，但由於ZooKeeper提供的原始

zookeeper開源客戶端Curator典型應用場景之-服務註冊與發現(十一)

隨著業務增加,以前簡單的系統已經變得越來越複雜，單純的提升伺服器效能也不是辦法,而且程式碼也是越來越龐大，維護也變得越來越困難，這一切都催生了新的架構設計風格 – 微服務架構的出現。 微服務給我們帶來了很多好處，例如：獨立可擴充套件、易維護。但是隨著應用的分解

zookeeper開源客戶端Curator典型應用場景之-Barrier屏障(十三)

什麼是Barrier Barrier是這樣的：Barrier是一個同步點，每一個程序到達此點都要等待，直到某一個條件滿足，然後所有的節點繼續進行。 比如：賽跑大家都知道,所有比賽人員都會在起跑線外等待,直到教練員的槍響之後,所有參賽者立刻開始賽跑。 JDK的併

zookeeper開源客戶端Curator典型應用場景之-分散式計數器(十四)

之前我們瞭解了基於Corator的分散式鎖之後,我們就很容易基於其實現一個分散式計數器，顧名思義，計數器是用來計數的, 利用ZooKeeper可以實現一個叢集共享的計數器。 只要使用相同的path就可以得到最新的計數器值， 這是由ZooKeeper的一致性保證

Zookeeper開源客戶端Curator之基本功能講解

簡介 Curator是Netflix公司開源的一套Zookeeper客戶端框架。瞭解過Zookeeper原生API都會清楚其複雜度。Curator幫助我們在其基礎上進行封裝、實現一些開發細節，包括接連重連、反覆註冊Watcher和NodeExistsExcept

zookeeper開源客戶端Curator典型應用場景之-訊息佇列(十二)

Curator框架也有分散式佇列實現。 利用ZK的PERSISTENT SEQUENTIAL(持久順序)節點，可以保證放入到佇列中的專案是按照順序排隊的。並且宕機重啟並不丟失訊息， 如果單一的消費者從佇列中取資料， 那麼它是先入先出的，這也是佇列的特點。 如果

zookeeper開源客戶端Curator典型應用場景之-Master選舉(十)

在生產環境中，一般要保證服務的高可用，有時候只需要選出一臺機器來執行，其餘機器處於備用狀態，比如，在分散式系統中很常見的一個問題就是定時任務的執行。如果多臺機器同時執行相同的定時任務，業務複雜則可能出現災難性的後果。我使用的是噹噹網的elastic-job分散

zookeeper客戶端命令行查看dubbo服務的生產者和消費者

ide 切換 microsoft ons -s sof 生產 size zkcli 假設zookeeper安裝在192.168.5.130這臺服務器上,現在我們通過命令行查看dubbo在zookeeper註冊服務的生產者和消費者信息 首先通過命令切換到/usr/zookee

zookeeper客戶端命令詳解

需要 des absolut 時間 its 認證 del 就會 ons 　　今天同事突然向看一下zookeeper中都創建了哪些節點，而我本人對zookeeper的客服端命令了解的很少，有些操作竟然不知道怎麽用，於是乎就索性整理一下zookeeper客服端命令的使用，並再此

zookeeper客戶端相關命令

server 同步 window iss lpad dig 技術分享 localhost gpo windows環境： 本機 直接 點機zkcli.cmd linux環境： 連接到zookeeper server ./zkCli.sh -server localh

Zookeeper客戶端對比選擇_4

del factor 異步 當前 lin pac call 深入 客戶端 Zookeeper客戶端對比選擇 本文思維導圖 使用框架的好處是自帶一套實用的API，但是Zookeeper雖然非常強大，但是社區卻安靜的可怕，版本更新較慢，下面會先從zookeeper原生API的

2018-07-13期 Zookeeper客戶端基本操作【本人親自反復驗證通過分享】

RoCE 分享 自動刪除 ast state println await roc 可用 說明：本文檔主要給大家介紹ZK客戶端API基本操作一、新建Java工程1、New->Project->Java Project填寫JAVA工程名字，選擇JDK版本等2、為新建