轉自：https://blog.csdn.net/qq_41112238/article/details/105240421

基本概念
大數(shù)據(jù)生態(tài)系統(tǒng)里很多組件的命名都是某種動物，例如Hadoop是??，hive是??，zookeeper就是動物園管理者，是管理大數(shù)據(jù)生態(tài)系統(tǒng)各組件的管理員。

zookeeper是經典的分布式數(shù)據(jù)一致性解決方案，致力于為分布式應用提供一個高性能，高可用，且具有嚴格順序訪問控制能力的分布式協(xié)調存儲服務。

應用場景
維護配置信息
Java編程經常會遇到配置項，例如數(shù)據(jù)庫的user、password等，通常配置信息會放在配置文件中，再把配置文件放在服務器上。當需要修改配置信息時，要去服務器上修改對應的配置文件，但在分布式系統(tǒng)中很多服務器都需要使用該配置文件，因此必須保證該配置服務的高可用性和各臺服務器上配置的一致性。通常會將配置文件部署在一個集群上，但一個集群涉及的服務器數(shù)量是很龐大的，如果一臺臺服務器逐個修改配置文件是效率很低且危險的，因此需要一種服務可以高效快速且可靠地完成配置項的更改工作。
zookeeper就可以提供這種服務，使用Zab一致性協(xié)議保證一致性。hbase中客戶端就是連接zookeeper獲得必要的hbase集群的配置信息才可以進一步操作。在開源消息隊列Kafka中，也使用zookeeper來維護broker的信息。在dubbo中也廣泛使用zookeeper管理一些配置來實現(xiàn)服務治理。

分布式鎖服務
一個集群是一個分布式系統(tǒng)，由多臺服務器組成。為了提高并發(fā)度和可靠性，在多臺服務器運行著同一種服務。當多個服務在運行時就需要協(xié)調各服務的進度，有時候需要保證當某個服務在進行某個操作時，其他的服務都不能進行該操作，即對該操作進行加鎖，如果當前機器故障，釋放鎖并fall over到其他機器繼續(xù)執(zhí)行。

集群管理
zookeeper會將服務器加入/移除的情況通知給集群中其他正常工作的服務器，以及即使調整存儲和計算等任務的分配和執(zhí)行等，此外zookeeper還會對故障的服務器做出診斷并嘗試修復。

生成分布式唯一ID
在過去的單庫單表系統(tǒng)中，通常使用數(shù)據(jù)庫字段自帶的auto_increment熟悉自動為每條記錄生成一個唯一的id。但分庫分表后就無法依靠該屬性來標識一個唯一的記錄。此時可以使用zookeeper在分布式環(huán)境下生成全局唯一性id。每次要生成一個新id時，創(chuàng)建一個持久順序結點，創(chuàng)建操作返回的結點序號，即為新id，然后把比自己結點小的刪除。

設計目標
高性能
將數(shù)據(jù)存儲在內存中，直接服務于客戶端的所有非事務請求，尤其適合讀為主的應用場景
高可用
一般以集群方式對外提供服務，每臺機器都會在內存中維護當前的服務狀態(tài)，每臺機器之間都保持通信。只要集群中超過一般機器都能正常工作，那么整個集群就能夠正常對外服務。
嚴格順序訪問
對于客戶端的每個更新請求，zookeeper都會生成全局唯一的遞增編號，這個編號反應了所有事務操作的先后順序。
數(shù)據(jù)結構
zookeeper的數(shù)據(jù)結點可以視為樹狀結構（或者目錄），樹中各節(jié)點成為znode，一個znode可以有多個子結點。zookeeper結點在結構上表現(xiàn)為樹狀，使用路徑來定位某個znode。
znode兼具文件和目錄兩種特點，既像文件一樣維護著數(shù)據(jù)、元信息、ACL、時間戳等數(shù)據(jù)結構，又像目錄一樣可以作為路徑標識的一部分。

znode大體上分為三部分（使用get命令查看）
①結點的數(shù)據(jù)
②結點的子結點
③結點的狀態(tài)

結點類型（在創(chuàng)建時被確定且不能更改）
①臨時結點：生命周期依賴于創(chuàng)建它的會話，會話結束結點將被自動刪除。臨時結點不允許擁有子結點。
②持久化結點：生命周期不依賴于會話，只有在客戶端顯式執(zhí)行刪除操作時才被刪除。

安裝（我恨linux）
使用centos8虛擬機，先創(chuàng)建一個zookeeper用戶

zookeeper是基于jdk的，先安裝jdk1.8：安裝JDK

安裝zookeeper（我是真的煩Linux 找個下載地址找一年 還慢死）：

wget http://mirrors.hust.edu.cn/apache/zookeeper/zookeeper-3.4.14/zookeeper-3.4.14.tar.gz
1
30kb/s 慢慢等吧

終于下載完了，tar -xzvf zookeeper-3.4.14.tar.gz進行解壓

然后進入zookeeper的conf配置目錄，復制zoo_sample.cfg 命名zoo.cfg。然后回到上級目錄，創(chuàng)建一個data目錄。

然后進入conf目錄，使用vi命令編輯剛才復制的zoo.cfg配置文件，打開后按a進入編輯模式。

修改datadir為剛才創(chuàng)建的data目錄，然后按esc，再按:到文件末尾，使用wq保存退出。

臥槽…居然成功啦
先cd ..返回zookeeper根目錄，再cd bin進入bin目錄，./zkServer.sh start 啟動zookeeper，等待一段時間后，./zkServer.sh status查看zookeeper是否成功啟動，Mode顯示的是當前是單機狀態(tài)

通過客戶端連接本機zookeeper服務./zkCli.sh，登錄成功后會打印日志信息

關閉zookeeper：

基本命令
新增結點
//-s為有序結點，-e為臨時結點
create [-s] [-e] path data
1
2
先啟動服務器

確實已經啟動

通過客戶端連接服務器

創(chuàng)建持久化結點
path是/hadoop 數(shù)據(jù)是“123456” 不加選項默認是持久化結點

通過get讀取數(shù)據(jù)

持久化結點與當前會話無關，quit退出會話后再次登陸，get數(shù)據(jù)還在

創(chuàng)建持久化有序結點
創(chuàng)建路徑為/a，會自動補為/a0000000001

創(chuàng)建臨時結點
臨時結點在會話結束后會被刪除

quit結束會話再次登陸，不能讀取到數(shù)據(jù)

創(chuàng)建臨時有序結點（用于分布式鎖）

更新結點
更新結點的命令是set，可以直接進行修改

修改/hadoop的值從123456變?yōu)閕 want offer

成功

也可以基于版本號更改，類似CAS機制，當傳入數(shù)據(jù)版本號和當前不一致時拒絕修改，初始版本號dataVersion為0，每次修改后會加1

當前版本為1，修改時使用1版本號，修改完變?yōu)?

版本號為2時，使用3會失敗

刪除結點
使用delete命令，同樣可以傳入版本號，如果版本號不符合也不會執(zhí)行
版本號為2時，使用3刪除失敗，使用2刪除成功

如果當前結點下有子結點，不能刪除，如果要刪除需要使用rmr

查看結點
使用get查看結點的屬性和數(shù)據(jù)

cZxid 數(shù)據(jù)結點創(chuàng)建時的事務id
ctime 數(shù)據(jù)結點創(chuàng)建時間
mZxid 數(shù)據(jù)結點最后一次更新時的事務id
mtime 數(shù)據(jù)結點最后一次更新的時間
pZxid 子結點最后一次修改的事務id
cversion 子結點的更改次數(shù)
dataVersion 結點數(shù)據(jù)更改次數(shù)
aclVersion 結點ACL的更改次數(shù)
ephemeralOwner 如果是臨時結點，表示會話的sessionID；如果是持久結點值為0
dataLength 數(shù)據(jù)內容長度
numChildren 子結點數(shù)

使用stat只返回屬性 沒有數(shù)據(jù)

查看結點列表
使用ls path或ls2 path，ls2是ls的增強，除了列出子結點還有當前結點的屬性

監(jiān)聽器（維護配置信息）
使用get path watch，監(jiān)聽器只能使用一次
左邊客戶端使用監(jiān)聽器，右邊客戶端更改了數(shù)據(jù)，左邊監(jiān)聽到了數(shù)據(jù)的更改

也可以使用stat path watch

還可以使用ls/ls2 path watch 可以監(jiān)聽到子結點

ACL權限控制
類似linux針對文件的權限控制
acl權限控制使用scheme：id：permission標識，
①scheme ：權限模式

world 只有一個用戶，代表登陸zookeeper的所有人（默認）
ip 對客戶端使用ip地址認證
auth 使用已添加認證的用戶認證
digest 使用用戶名：密碼方式認證
②id ：授權對象
③permission ：授予的權限

create 簡寫c 可以創(chuàng)建子結點
delete 簡寫d 可以刪除子結點（僅下一級）
read 簡寫r 可以讀取結點數(shù)據(jù)以及顯示子結點
write 簡寫w 可以設置結點數(shù)據(jù)
admin 簡寫a 可以設置節(jié)點訪問控制列表權限
例如setAcl /hadoop ip:192.168.2.142:crwda 表示ip地址為192.168.2.142的客戶端對/hadoop有全部權限
getAcl path 讀取權限
setAcl path acl 設置權限
addauth schema suth 添加認證用戶

world授權模式
setAcl path world:anyone:<acl>
例如取消c創(chuàng)建權限，就不能再創(chuàng)建子結點

取消d權限，不能刪除子結點

ip授權模式
命令setAcl path ip:<ip>:<acl>
要用兩個虛擬機，這個就不演示了…
主要就是限制客戶端ip地址的

auth模式
addauth digest <user>:<password>添加認證用戶
setAcl <path> auth:<user>:<acl> 授權
1
2

另一臺客戶端如果沒有添加該用戶就不能讀取

digest授權模式
setAcl <path> digest:<user>:<password>:<acl>
這里的密碼是經過SHA1及BASE64處理的密文
通過echo -n sjh:sjh2019. | openssl dgst -binary -sha1 | openssl base64生成密文，復制該結果

添加結點/node4，使用digest模式授權，因為之前添加過sjh密碼sjh019.的用戶，所以可以訪問

多種授權模式
同一個結點可以使用多種授權模式，用逗號隔開就行，例

超級管理員
假設超級管理員的賬號是super:admin，先要為其生成密文：

得到密文xQJmxLMiHGwaqBvst5y6rkB6HQs=
在zookeeper目錄下/bin/zkServer.sh服務器腳本文件，找到這一行

"-Dzookeeper.DigestAuthenticationProvider.superDigest=super:xQJmxLMiHGwaqBvst5y6rkB6HQs="加入這一句

修改完按esc ，按：， 輸入wq保存退出
重啟

使用客戶端登陸，創(chuàng)建/node6，取消創(chuàng)建子結點權限

此時不能創(chuàng)建子結點

添加超級管理員用戶，此時可以成功添加子結點

IDEA操作zookeeper
操作流程
連接到zookeeper服務器（好害怕連不上- - 。。。昨天GitHub連上了，不知道今天能不能再撞一次運氣。。連不上這篇文章也就到此為止了。。。）
定期向服務器發(fā)送心跳，否則會過期
會話處于活動狀態(tài)就可以獲取/設置znode
所有任務完成后，斷開連接
連接到zookeeper
創(chuàng)建一個新的Java工程，導入一下jar包

媽的連了快倆小時終于搞定了?。?！

導完jar包后還需要導入一個log4j文件，創(chuàng)建一個連接zookeeper的測試類

public class zookeeperTest {

public static void main(String[] args) {
    ZooKeeper zooKeeper = null;
    try{
        //創(chuàng)建一個計數(shù)器對象
        CountDownLatch countDownLatch=new CountDownLatch(1);
        //第一個參數(shù)是服務器ip和端口號，第二個參數(shù)是客戶端與服務器的會話超時時間單位ms，第三個參數(shù)是監(jiān)視器對象
        zooKeeper=new ZooKeeper("192.168.2.142:2181", 5000, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                if(event.getState()==Event.KeeperState.SyncConnected){
                    System.out.println("連接創(chuàng)建成功");
                    //通知主線程解除阻塞
                    countDownLatch.countDown();
                }
            }
        });
        //主線程阻塞，等待連接對象的創(chuàng)建成功
        countDownLatch.await();
        System.out.println("會話編號"+zooKeeper.getSessionId());
    }catch (Exception e){
        e.printStackTrace();
    }finally {
        if(zooKeeper!=null) {
            try {
                zooKeeper.close();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
由于連接是異步的，所以用countdownlatch確保連接成功再繼續(xù)
一直失敗！
最后關了防火墻。。。就成功了?。?！

創(chuàng)建結點
在Linux客戶端先創(chuàng)建一個/create結點

在IDEA創(chuàng)建一個子結點/node1

public class zkCreate {

private static final String IP="192.168.2.142:2181";
private static ZooKeeper zooKeeper;

@Before
public void connect() throws Exception{
    //創(chuàng)建一個計數(shù)器對象
    CountDownLatch countDownLatch=new CountDownLatch(1);
    //第一個參數(shù)是服務器ip和端口號，第二個參數(shù)是客戶端與服務器的會話超時時間單位ms，第三個參數(shù)是監(jiān)視器對象
    zooKeeper=new ZooKeeper(IP, 5000, new Watcher() {
        @Override
        public void process(WatchedEvent event) {
            if(event.getState()==Event.KeeperState.SyncConnected){
                System.out.println("連接創(chuàng)建成功");
                //通知主線程解除阻塞
                countDownLatch.countDown();
            }
        }
    });
    //主線程阻塞，等待連接對象的創(chuàng)建成功
    countDownLatch.await();
}

@After
public void close() throws Exception{
    zooKeeper.close();
}

@Test
public void create1() throws Exception{
    //同步創(chuàng)建結點
    // 參數(shù)1 結點路徑
    // 參數(shù)2 結點數(shù)據(jù)
    // 參數(shù)3權限列表 OPEN_ACL_UNSAFE代表world方式授權 cdrwa
    // 參數(shù)4 結點類型 persistent表示持久化結點
    zooKeeper.create("/create/node1","i want offer".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}

}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
運行后在Linux客戶端查看

創(chuàng)建一個只有讀權限的數(shù)據(jù)

@Test
public void create2() throws Exception{
    //  OPEN_ACL_UNSAFE代表world方式授權 r只能讀
    zooKeeper.create("/create/node2","i want offer".getBytes(), ZooDefs.Ids.READ_ACL_UNSAFE, CreateMode.PERSISTENT);
}

1
2
3
4
5
運行后查詢

自定義設置權限列表
使用world模式，設置有創(chuàng)建和刪除權限

@Test
public void create3() throws Exception{
    //自定義方式設置權限
    List<ACL> acls=new ArrayList<>();
    Id id=new Id("world","anyone");
    acls.add(new ACL(ZooDefs.Perms.CREATE,id));
    acls.add(new ACL(ZooDefs.Perms.DELETE,id));
    zooKeeper.create("/create/node4","i want offer".getBytes(), acls, CreateMode.PERSISTENT);
}

1
2
3
4
5
6
7
8
9
ip模式授權

@Test
public void create4() throws Exception{
//ip方式設置權限
List<ACL> acls=new ArrayList<>();
Id id=new Id("ip","192.168.2.142");
acls.add(new ACL(ZooDefs.Perms.CREATE,id));
zooKeeper.create("/create/node4","i want offer".getBytes(), acls, CreateMode.PERSISTENT);
}
1
2
3
4
5
6
7
8
auth方式授權

@Test
public void create5() throws Exception{
//auth方式設置權限
zooKeeper.addAuthInfo("digest","sjh:sjh2019.".getBytes());
zooKeeper.create("/create/node5","i want offer".getBytes(), ZooDefs.Ids.CREATOR_ALL_ACL, CreateMode.PERSISTENT);
}
1
2
3
4
5
6
digest方式授權

@Test
public void create6() throws Exception{
//digest方式設置權限
List<ACL> acls=new ArrayList<>();
Id id=new Id("digest","sjh:base64和sha1加密后的密碼");
acls.add(new ACL(ZooDefs.Perms.CREATE,id));
zooKeeper.create("/create/node5","i want offer".getBytes(), acls, CreateMode.PERSISTENT);
}
1
2
3
4
5
6
7
8
創(chuàng)建持久化有序結點

@Test
public void create7() throws Exception{
//持久化有序結點
String s = zooKeeper.create("/create/node7", "i want offer".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
System.out.println(s);
}
1
2
3
4
5
6
運行結果：

創(chuàng)建臨時結點

@Test
public void create8() throws Exception{
//創(chuàng)建臨時結點
String s = zooKeeper.create("/create/node8", "i want offer".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
System.out.println(s);
}
1
2
3
4
5
6
創(chuàng)建臨時有序結點

@Test
public void create9() throws Exception{
//創(chuàng)建臨時結點
String s = zooKeeper.create("/create/node9", "i want offer".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
System.out.println(s);
}
1
2
3
4
5
6
異步創(chuàng)建結點

 @Test
public void create10() throws Exception{
    //異步創(chuàng)建結點
    zooKeeper.create("/create/node11", "i want offer".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT, new AsyncCallback.StringCallback() {
        @Override
        public void processResult(int rc, String path, Object ctx, String name) {
            System.out.println("創(chuàng)建狀態(tài): "+rc);//0表示創(chuàng)建成功
            System.out.println("path: "+path);//結點路徑
            System.out.println("name: "+name);//結點路徑
            System.out.println("ctx: "+ctx);//上下文
        }
    },"context");
}

1
2
3
4
5
6
7
8
9
10
11
12
13
運行結果，由于是異步的，可能還沒有輸出就結束了

調用sleep方法讓線程休眠，等待zookeeper操作執(zhí)行完畢

成功：

更新結點
先查詢/create/node1結點的當前值為i wangt offer

同步更新結點：

@Test
public void set1() throws Exception{
    //同步更新結點
    //第一個參數(shù) 結點路徑
    //第二個參數(shù) 要修改的值
    //第三個參數(shù) 數(shù)據(jù)版本 -1代表版本號不參與更新
    zooKeeper.setData("/create/node1","2020GetGoodOffer".getBytes(),-1);
}

1
2
3
4
5
6
7
8
運行完成后，再次查詢該結點，值已經更改為2020GetGoodOffer

異步修改結點

@Test
public void set2() throws Exception{
//異步更新結點
//第一個參數(shù) 結點路徑
//第二個參數(shù) 要修改的值
//第三個參數(shù) 數(shù)據(jù)版本 -1代表版本號不參與更新
//第四個參數(shù) 匿名回調函數(shù)
//第五個參數(shù) 上下文參數(shù)
zooKeeper.setData("/create/node1", "2020 Get Offer?。?！".getBytes(), -1, new AsyncCallback.StatCallback() {
@Override
public void processResult(int rc, String path, Object ctx, Stat stat) {
System.out.println("rc: "+rc);//0表示成功
System.out.println("path: "+path);
System.out.println("ctx: "+ctx);
System.out.println("stat: "+stat);
}
},"context");
Thread.sleep(1000);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
結果：

刪除結點
同步刪除

@Test
public void del1() throws Exception{
//同步刪除數(shù)據(jù)
//第一個參數(shù)表示刪除結點的路徑
//第二個參數(shù)表示刪除結點的數(shù)據(jù)版本 -1表示刪除時不考慮版本信息
zooKeeper.delete("/create/node1",-1);
}
1
2
3
4
5
6
7
此時數(shù)據(jù)已不存在

異步刪除

@Test
public void del2() throws Exception{
//異步刪除數(shù)據(jù)
zooKeeper.delete("/create/node2", -1, new AsyncCallback.VoidCallback() {
@Override
public void processResult(int rc, String path, Object ctx) {
System.out.println("rc: "+rc);//0表示成功
System.out.println("path: "+path);
System.out.println("ctx: "+ctx);
}
},"context");
Thread.sleep(1000);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
運行結果

查看結點
同步查看結點

@Test
public void get1() throws Exception{
//同步讀取數(shù)據(jù)
//第一個參數(shù)是路徑
//第二個參數(shù)是watch,先填false（以后在講）
//第三個參數(shù)用于獲取結點屬性
Stat stat = new Stat();
byte[] data = zooKeeper.getData("/create/node3", false, stat);
System.out.println(new String(data));//數(shù)據(jù)
System.out.println(stat);//屬性
}
1
2
3
4
5
6
7
8
9
10
11
結果

異步查看結點

@Test
public void get2() throws Exception{
//異步讀取數(shù)據(jù)
//第一個參數(shù)是路徑
//第二個參數(shù)是watch,先填false（以后在講）
//第三個參數(shù)是匿名回調函數(shù)
Stat stat = new Stat();
zooKeeper.getData("/create/node3", false, new AsyncCallback.DataCallback() {
@Override
public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
System.out.println("rc: "+rc);//0表示成功
System.out.println("path: "+path);//結點路徑
System.out.println("ctx: "+ctx);//上下文參數(shù)
System.out.println(new String(data));//數(shù)據(jù)
System.out.println(stat);//屬性
}
},"context");
Thread.sleep(1000);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
運行結果：

查看子結點
創(chuàng)建測試數(shù)據(jù)

同步

@Test
public void getChild1() throws Exception{
    //同步
    //第一個參數(shù)是父路徑
    //第二個參數(shù)是watch,先填false（以后在講）
    List<String> children = zooKeeper.getChildren("/create/father", false);
    for(String str:children)
        System.out.println(str);
}

1
2
3
4
5
6
7
8
9
運行結果：

異步

@Test
public void getChild2() throws Exception{
//異步
//第一個參數(shù)是父路徑
//第二個參數(shù)是watch,先填false（以后在講）
//第三個參數(shù)是匿名回調函數(shù)
zooKeeper.getChildren("/create/father", false, new AsyncCallback.ChildrenCallback() {
@Override
public void processResult(int rc, String path, Object ctx, List<String> children) {
System.out.println("rc: "+rc);//0表示成功
System.out.println("path: "+path);//結點路徑
System.out.println("ctx: "+ctx);//上下文參數(shù)
System.out.println(children);//子結點信息
}
},"context");
Thread.sleep(1000);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
運行結果

檢查結點是否存在
同步

@Test
public void exists1() throws Exception{
    //同步判斷
    //第一個參數(shù)是路徑
    //第二個參數(shù)是watch,先填false（以后在講）
    Stat exists = zooKeeper.exists("/create/null", false);
    System.out.println(exists==null?"不存在":"存在");
}

1
2
3
4
5
6
7
8
運行結果：

異步

@Test
public void exists2() throws Exception{
//異步判斷
//第一個參數(shù)是路徑
//第二個參數(shù)是watch,先填false（以后在講）
//第三個參數(shù)是匿名回調函數(shù)
//第四個參數(shù)上下文
zooKeeper.exists("/create/null", false, new AsyncCallback.StatCallback() {
@Override
public void processResult(int rc, String path, Object ctx, Stat stat) {
System.out.println("rc: "+rc);//0表示成功
System.out.println("path: "+path);//結點路徑
System.out.println("ctx: "+ctx);//上下文參數(shù)
System.out.println(stat==null?"不存在":"存在");
}
},"context");
Thread.sleep(1000);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
運行結果：