Zookeeper入門看這篇就夠了

作者：牧小農(nóng) 2021-09-30 07:59:06

Zookeeper 是一個分布式應(yīng)用程序的分布式開源協(xié)調(diào)服務(wù)。是Apache Hadoop 的一個子項目，主要用來解決分布式應(yīng)用中經(jīng)常遇到的一些數(shù)據(jù)管理問題，例如：統(tǒng)一命名服務(wù)、狀態(tài)同步服務(wù)、集群管理、分布式應(yīng)用配置項的管理等。

[[426770]]

簡介

Zookeeper 工作原理

ZooKeeper 核心是原子廣播，該機(jī)制保證了各個Server之間的同步，實(shí)現(xiàn)這個機(jī)制的協(xié)議叫做 Zab協(xié)議，Zab協(xié)議有兩個模式，他們分別是 “恢復(fù)模式 & 廣播模式”。

恢復(fù)模式

Zab協(xié)議會讓ZK集群進(jìn)入崩潰恢復(fù)模式的情況如下：

(1)當(dāng)服務(wù)框架在啟動過程中(2)當(dāng)Leader服務(wù)器出現(xiàn)網(wǎng)絡(luò)中斷，崩潰退出與重啟等異常情況。(3)當(dāng)集群中已經(jīng)不存在過半的服務(wù)器與Leader服務(wù)器保持正常通信。

在所有的follower服務(wù)器中選舉一臺為Leader，當(dāng)leader被選舉出來，集群中有多數(shù)服務(wù)與新的Leader完成狀態(tài)同步之后就會退出恢復(fù)模式，用來保證至少有一半的follower能和Leader保持?jǐn)?shù)據(jù)一致，當(dāng)多數(shù)的follower集群與leader數(shù)據(jù)保持一致的時候，就會進(jìn)入消息廣播模式。

狀態(tài)同步保證了 Leader 和 Server具有相同的系統(tǒng)狀態(tài)，所謂的狀態(tài)同步其實(shí)就是數(shù)據(jù)的同步。

一旦leader 已經(jīng)和多數(shù)的follower進(jìn)行狀態(tài)同步之后，它就開始廣播消息，并且進(jìn)入廣播模式，這時候當(dāng)一個server加入Zookeeper 服務(wù)中，它會在恢復(fù)模式下啟動，發(fā)現(xiàn)leader，并和leader進(jìn)行狀態(tài)同步，同步結(jié)束后，它也參與消息廣播，Zookeeper服務(wù)一直維持在 Broadcast狀態(tài)，直到leader崩潰了或者leader失去了大部分的followers支持。

廣播模式

消息廣播模式，Zab協(xié)議消息廣播過程使用的是原子廣播協(xié)議，類似于一個二階段提交，但是又有點(diǎn)不一樣，并不是所有的follower節(jié)點(diǎn)都需要返回ack才進(jìn)行一致性事務(wù)完成，只需要多數(shù)以上即可。

針對每個客戶端的事務(wù)請求，leader服務(wù)器會為其生成對應(yīng)的事務(wù)Proposal，并將其發(fā)送給集群中其余所有的機(jī)器，然后再分別收集各自的選票，最后進(jìn)行事務(wù)提交。

leader 接收到消息請求后，將消息賦予一個全局唯一的 64 位自增的 Id，我們通常稱之為zxid，通過 zxid 的大小比較即可實(shí)現(xiàn)有序的特性。
leader 通過隊列保證發(fā)送的順序性，將帶有zxid的消息作為一個提案(proposal)分發(fā)給所有follower
當(dāng)follower 接收到proposal，先將proposal寫到本地事務(wù)日志，寫事務(wù)成功后再向Leader 回一個ACK確認(rèn)
當(dāng)leader 接收到多數(shù)的ack確認(rèn)后，leader 會向所有follower 發(fā)送 commit 命令，同意會在本地執(zhí)行該消息。
當(dāng)follower 收到消息 commit 命令后，就會執(zhí)行該消息。

消息廣播模式流程示意圖如下：

首先客戶端會輪詢Zookeeper集群中的各個節(jié)點(diǎn)，當(dāng)輪詢到一臺是follower，如果是讀的請求，follower會返回請求結(jié)果，如果是增刪改操作，follower 會向leader生成事務(wù)請求，針對客戶端的事務(wù)請求，針對客戶端的事務(wù)請求，leader會為這個生成對應(yīng)的事務(wù)Proposal，然后發(fā)送集群中所有follower服務(wù)器，然后分別在收集各自的選票，最后進(jìn)行事務(wù)提交。

Zab協(xié)議的二階段提交，在提交過程中移除了中斷提交過程的操作，對于Zookeeper集群來說，超過半數(shù)反饋Ack確認(rèn)就代表事務(wù)成功，這種方式無法完成所有節(jié)點(diǎn)事務(wù)一致性問題，所以Zab協(xié)議采用恢復(fù)模式來解決數(shù)據(jù)不一致的問題。

消息廣播協(xié)議是基于具有FIFO特性的TCP協(xié)議進(jìn)行通訊，因此可以保證消息廣播過程中的接收和發(fā)送的順序性。

事務(wù)ID

為了保證事務(wù)的順序一致性，Zookeeper 采用了遞增的事務(wù)ID號(zxid)來標(biāo)識事務(wù)，所有的操作(proposal)都會在被提出時加上zxid，zxid是一個64位的數(shù)字，他高32位是epoch用來標(biāo)識leader關(guān)系是否發(fā)生變化，每當(dāng)有新的leader 被選舉出來，都會有一個新的epoch，標(biāo)識當(dāng)前屬于哪個leader的領(lǐng)導(dǎo)。

對于Zookeeper 來說，每次的變化都會產(chǎn)生一個唯一的事務(wù)id，zxid(ZooKeeper Transaction Id)通過zxid ，可以確定更新操作的先后順序，如果說 zxid1 小于 zxid2，說明 zxid1比zxid先發(fā)生。

Zookeeper 模型

Zookeeper 是一個目錄樹結(jié)構(gòu)，名稱是由斜杠 (/) 分隔的一系列路徑元素。ZooKeeper 命名空間中的每個節(jié)點(diǎn)都由路徑標(biāo)識。

ZooKeeper 層級樹狀結(jié)構(gòu)

根節(jié)點(diǎn) / 包含兩個節(jié)點(diǎn)(/modele1 & /module2)，其中節(jié)點(diǎn) /module1 包含三個子節(jié)點(diǎn)(/module1/app1 & /module1/app2 & /module1/app3)，在Zookeeper 中，節(jié)點(diǎn)以絕對路徑表示，不存在相對路徑，出了根節(jié)點(diǎn)以外，其他節(jié)點(diǎn)不能以 / 結(jié)尾。

特性

資源共享： 例如存儲空間，計算能力，數(shù)據(jù)，和服務(wù)等等

擴(kuò)展性： 從軟件和硬件上增加系統(tǒng)的規(guī)模

并發(fā)性： 多個用戶同時訪問

性能： 確保當(dāng)負(fù)載增加的時候，系統(tǒng)想要時間不會有影響

容錯性： 盡管一些組件暫時不可用了，整個系統(tǒng)仍然是可用的

API抽象： 系統(tǒng)的獨(dú)立組件對用戶隱藏，僅僅暴露服務(wù)

Zookeeper的角色

領(lǐng)導(dǎo)者(leader) ：負(fù)責(zé)進(jìn)行投票的發(fā)起和決議，更新系統(tǒng)狀態(tài)

學(xué)習(xí)者(learner) ：包括跟隨者(follower)和觀察者(observer)，follower用于接受客戶端請求并想客戶端返回結(jié)果，在選主過程中參與投票

Observer ：可以接受客戶端連接，將寫請求轉(zhuǎn)發(fā)給leader，但observer不參加投票過程，只同步leader的狀態(tài)，observer的目的是為了擴(kuò)展系統(tǒng)，提高讀取速度

客戶端(client) ：請求發(fā)起方

保證

順序一致性： 客戶端的更新將按發(fā)送順序應(yīng)用。

原子性： 更新成功或失敗，沒有部分結(jié)果。

統(tǒng)一視圖： 無論服務(wù)器連接到哪個服務(wù)器，客戶端都將看到相同的服務(wù)視圖。即，即使客戶端故障轉(zhuǎn)移到具有相同會話的不同服務(wù)器，客戶端也永遠(yuǎn)不會看到系統(tǒng)的舊視圖。

可靠性： 一旦應(yīng)用更新了，它將從那時起一直存在，直到客戶端覆蓋更新。

及時性： 系統(tǒng)的客戶視圖保證在特定時間范圍內(nèi)是最新的。

Znode 節(jié)點(diǎn)

Znode有兩種類型：持久節(jié)點(diǎn)和臨時節(jié)點(diǎn) ，Znode的類型在創(chuàng)建的之后就不能在進(jìn)行修改了。

臨時節(jié)點(diǎn)

臨時節(jié)點(diǎn)在客戶端會話結(jié)束的時候，Zookeeper 會將臨時節(jié)點(diǎn)(znode)刪除，并且臨時節(jié)點(diǎn)不能有子節(jié)點(diǎn)。利用臨時節(jié)點(diǎn)的特性，我們可以使用臨時節(jié)點(diǎn)來進(jìn)行集群管理以及發(fā)現(xiàn)服務(wù)的上下線等。

創(chuàng)建臨時節(jié)點(diǎn)命令：create-e/module1/app1 app1創(chuàng)建一個臨時節(jié)點(diǎn)為 “/module1/app1” ，數(shù)據(jù)為 “app1”

持久節(jié)點(diǎn)

持久節(jié)點(diǎn)不依賴于客戶端會話，只有當(dāng)客戶端明確要刪除持久節(jié)點(diǎn)(znode)的時候才會被刪除

創(chuàng)建臨時節(jié)點(diǎn)命令：create/module1 module1創(chuàng)建一個臨時節(jié)點(diǎn)為 “/module1” ，數(shù)據(jù)為 “module1”

順序節(jié)點(diǎn)

ZooKeeper 中還提供了一種順序節(jié)點(diǎn)的類型，每次創(chuàng)建順序節(jié)點(diǎn)時候，ZooKeeper 都會在路徑后面自動添加10為的數(shù)據(jù)中，例如0000000001 計數(shù)器會保證在同一父節(jié)點(diǎn)下唯一，創(chuàng)建節(jié)點(diǎn)的時候會添加順序，常見分布式鎖。

順序節(jié)點(diǎn)只是節(jié)點(diǎn)的一種特性，也就說不管是持久節(jié)點(diǎn)還是臨時節(jié)點(diǎn) 都可以設(shè)置為順序節(jié)點(diǎn)，所以Znode類型可以理解為 4種類型：

持久節(jié)點(diǎn)
臨時節(jié)點(diǎn)
持久順序節(jié)點(diǎn)
臨時順序節(jié)點(diǎn)

創(chuàng)建順序節(jié)點(diǎn)命令(加上 “-s”參數(shù))：create-s/module1/app app我們會看到 Created/module1/app0000000001意思是我們創(chuàng)建了一個持久順序節(jié)點(diǎn)“/module1/app0000000001” 如果再執(zhí)行上面命令會生成節(jié)點(diǎn) “/module1/app0000000002”，同理如果我們 create-s后面添加 -e 參數(shù)，就表示我們創(chuàng)建了一個臨時節(jié)點(diǎn)。

節(jié)點(diǎn)數(shù)據(jù)

創(chuàng)建節(jié)點(diǎn)的時候，我們可以指定節(jié)點(diǎn)中存儲的數(shù)據(jù)，ZooKeeper可以保證讀寫都是原子操作，而且每次讀寫操作都是對數(shù)據(jù)的完整讀取或者完成寫入，不提供對數(shù)據(jù)的部分讀取或者寫入操作。
ZooKeeper 雖然提供了節(jié)點(diǎn)存儲數(shù)據(jù)的功能，但是我們并不能把它當(dāng)成一個數(shù)據(jù)庫，重點(diǎn)不要把Zookeeper 當(dāng)成數(shù)據(jù)庫用，因?yàn)閆ookeeper 規(guī)定了節(jié)點(diǎn)的數(shù)據(jù)大小不能超過1M，所以我們不能在節(jié)點(diǎn)上存儲過多的數(shù)據(jù)，盡可能保證小的數(shù)據(jù)量，因?yàn)閿?shù)據(jù)過大，會導(dǎo)致ZK的性能下降。
如果確實(shí)需要存儲大量的數(shù)據(jù)，一般可以在分布式數(shù)據(jù)庫或者Redis保存這部分?jǐn)?shù)據(jù)，然后在Znode中保留數(shù)據(jù)庫中的索引。

Zookeeper單機(jī)模式安裝

java 環(huán)境

配置JAVA環(huán)境，檢驗(yàn)環(huán)境 java-version

下載安裝Zookeeper

下載地址：https://zookeeper.apache.org/releases.html

下載解壓Zookeeper

cd /usr/local/ 
wget https://dlcdn.apache.org/zookeeper/zookeeper-3.7.0/apache-zookeeper-3.7.0-bin.tar.gz 
tar -zxvf apache-zookeeper-3.7.0-bin.tar.gz 
cd zookeeper-3.7.0/

重命名配置文件 zoo_sample.cfg

cp conf 
/ 
zoo_sample 
. 
cfg conf 
/ 
zoo 
. 
cfg

啟動ZK

./ 
bin 
/ 
zkServer 
. 
sh start

連接ZK客戶端

./ 
bin 
/ 
zkCli 
. 
sh

當(dāng)我們看到下圖的信息的時候，表示我們啟動成功

Zookeeper命令

基本命令

create : 在樹中的某個位置創(chuàng)建一個節(jié)點(diǎn)
delete : 刪除一個節(jié)點(diǎn)存在：測試節(jié)點(diǎn)是否存在于某個位置
get data : 從節(jié)點(diǎn)讀取數(shù)據(jù)
set data：將數(shù)據(jù)寫入節(jié)點(diǎn)
get children : 檢索節(jié)點(diǎn)的子節(jié)點(diǎn)列表
sync : 等待數(shù)據(jù)被傳播

操作Zookeeper

查看Zookeeper中包含的key

ls  
/

創(chuàng)建一個新的Znode 創(chuàng)建成功以后我們可以使用 ls/查看我們創(chuàng)建的內(nèi)容

create /zkMxn muxiaonong ls /[zkMxn, zookeeper]

get命令獲取創(chuàng)建Znode的內(nèi)容

get /zkMxn

set 命令來對 zk 所關(guān)聯(lián)的字符串進(jìn)行設(shè)置

set /zkMxn mxn666

刪除Znode

delete /zkMxn

Java Api操作 ZK

1. 導(dǎo)入Jar包

<dependency>  
 
<groupId>org.apache.zookeeper</groupId>  
 
<artifactId>zookeeper</artifactId> 
 
<version>3.6.3</version> 
 
</dependency> 
 
<!--junit單元測試-->  
 
<dependency>  
 
<groupId>junit</groupId>  
 
<artifactId>junit</artifactId>  
 
<version>4.13.2</version>  
 
<scope>runtime</scope>  
 
</dependency>

2. API操作Zookeeper

創(chuàng)建Zookeeper對象

public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher) throws IOException { 
      this(connectString, sessionTimeout, watcher, false); 
  }

connectString：連接的地址，包括主機(jī)名和端口號，多個的話用逗號隔開

sessionTimeout：等待客戶端通信的最長時間，客戶端如果超過這個時間沒有和服務(wù)端進(jìn)行通信，那么就認(rèn)為該客戶端已經(jīng)終止，一般設(shè)置值為 5-10秒，單位為毫秒

watcher：監(jiān)聽器，用于接收會話事件的接口，需要自己定義，實(shí)現(xiàn)process()方法

連接Zookeeper

Zookeeper zkClient = ""; 
String connectStr = "192.168.2.1:2181"; 
zkClient = new ZooKeeper(connectStr, 5000, new Watcher() { 
    @Override 
    public void process(WatchedEvent watchedEvent) { } 
});

創(chuàng)建節(jié)點(diǎn)

public ZooKeeper(String connectString, int sessionTimeout, Watcher watcher) throws IOException { 
      this(connectString, sessionTimeout, watcher, false); 
  }

path：節(jié)點(diǎn)路徑
data：節(jié)點(diǎn)數(shù)據(jù)
acl：節(jié)點(diǎn)權(quán)限，例如： ZooDefs.Ids.OPEN_ACL_UNSAFE

OPENACLUNSAFE：完全開發(fā)，采用world驗(yàn)證模式，由于每個ZK連接都有world驗(yàn)證模式，所以當(dāng)我們節(jié)點(diǎn)設(shè)置了該參數(shù)時，對所有連接開放 CREATORALLACL：創(chuàng)建該Znode連接的擁有所有權(quán)限，這里采用的是auth驗(yàn)證模式，用sessionID做驗(yàn)證，如果設(shè)置了該參數(shù)，只有創(chuàng)建改Znode節(jié)點(diǎn)的連接才能對這個Znode進(jìn)行任何操作 READACLUNSAFE：所有的客戶端都可讀，這里采用world驗(yàn)證模式，和第一條同理，所有連接都可以讀取該znode

createMode：節(jié)點(diǎn)類型，例如： CreateMode.PERSISTENT PERSISTENT：持久節(jié)點(diǎn) PERSISTENTSEQUENTIAL：持久有序節(jié)點(diǎn) EPHEMERAL：短暫節(jié)點(diǎn) EPHEMERALSEQUENTIAL：短暫有序節(jié)點(diǎn)

完整APIDemo：

import lombok.extern.slf4j.Slf4j; 
import org.apache.zookeeper.*; 
import org.apache.zookeeper.data.Stat; 
import java.io.IOException; 
import java.io.UnsupportedEncodingException; 
import java.util.concurrent.CountDownLatch; 
 
/** @Author mxn 
  * @Description //TODO ZooKeeper Java API測試 
  * @Date 10:22 2021/9/29 
  * @Param 
  * @return 
 **/ 
@Slf4j 
public class ZookeeperTest { 
 
    // IP 和端口 
    private final static String ipAddress = "192.168.2.123:2181"; 
 
 
    public static void main(String[] args) { 
        ZookeeperTest test = new ZookeeperTest(); 
 
        String key = "/zkMxn"; 
        String value = "wo is muxiaonong"; 
        //創(chuàng)建Znode 
        test.add(key,value); 
//        獲取節(jié)點(diǎn)數(shù)據(jù) 
//        test.get(key); 
        //修改節(jié)點(diǎn)數(shù)據(jù) 
//        test.modify(key,"wo is zhuzhuxia"); 
        //刪除節(jié)點(diǎn) 
//        test.delete(key); 
 
 
    } 
 
    /** 
     * @return 
     * @Author mxn 
     * @Description //TODO 獲取ZooKeeper連接 
     * @Date 10:22 2021/9/29 
     * @Param 
     **/ 
    public static ZooKeeper getConntection() { 
        ZooKeeper zooKeeper = null; 
        try { 
            final CountDownLatch countDownLatch = new CountDownLatch(1); 
            //watch機(jī)制(回調(diào))，監(jiān)聽是否連接成功 
            zooKeeper = new ZooKeeper(ipAddress, 5000, new Watcher() { 
                @Override 
                public void process(WatchedEvent watchedEvent) { 
                    if (Event.KeeperState.SyncConnected == watchedEvent.getState()) { 
                        //如果受收到了服務(wù)端的響應(yīng)事件，連接成功 
                        countDownLatch.countDown(); 
                    } 
                } 
            }); 
 
            countDownLatch.await(); 
            log.info("zookeeper狀態(tài)：{}",zooKeeper.getState());//CONNECTED 
 
        } catch (IOException e) { 
            e.printStackTrace(); 
        } catch (InterruptedException e) { 
            e.printStackTrace(); 
        } 
        return zooKeeper; 
    } 
 
   /** @Author lyy 
     * @Description //TODO 關(guān)閉ZooKeeper連接 
     * @Date 14:57 2021/9/29 
     * @Param  
     * @return  
    **/ 
    public static void closeConnection(ZooKeeper zooKeeper) { 
        try { 
//            zooKeeper.close(); 
        } catch (Exception e) { 
            e.printStackTrace(); 
        } 
    } 
 
     
    /** @Author lyy 
      * @Description //TODO 添加節(jié)點(diǎn) 
      * @Date 13:36 2021/9/29 
      * @Param  
      * @return  
     **/ 
    public void add(String key ,String value) { 
        ZooKeeper zooKeeper = ZookeeperTest.getConntection(); 
        try { 
            //參數(shù)類型 
            //1.key 
            //2.value 
            //3.對應(yīng)的ACL,當(dāng)前節(jié)點(diǎn)的權(quán)限控制 
            //4.設(shè)置當(dāng)前節(jié)點(diǎn)類型 
            zooKeeper.create(key, value.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); 
            ZookeeperTest.closeConnection(zooKeeper); 
        } catch (KeeperException e) { 
            e.printStackTrace(); 
        } catch (InterruptedException e) { 
            e.printStackTrace(); 
        } 
    } 
 
   /** @Author lyy 
     * @Description //TODO 獲取節(jié)點(diǎn)信息 
     * @Date 14:57 2021/9/29 
     * @Param  
     * @return  
    **/ 
    public void get(String key) { 
        ZooKeeper zooKeeper = ZookeeperTest.getConntection(); 
        Stat stat = new Stat(); 
        String data = null; 
        try { 
            byte[] bytes = zooKeeper.getData(key, null, stat); 
            data = new String(bytes, "gbk"); 
            log.info("當(dāng)前節(jié)點(diǎn)信息:{}",data); 
            ZookeeperTest.closeConnection(zooKeeper); 
        } catch (KeeperException e) { 
            e.printStackTrace(); 
        } catch (InterruptedException e) { 
            e.printStackTrace(); 
        } catch (UnsupportedEncodingException e) { 
            e.printStackTrace(); 
        } 
    } 
 
    /** @Author lyy 
      * @Description //TODO 修改節(jié)點(diǎn)信息 
      * @Date 14:57 2021/9/29 
      * @Param  
      * @return  
     **/ 
    public void modify(String key,String newValue) { 
        ZooKeeper zooKeeper = ZookeeperTest.getConntection(); 
        Stat stat = new Stat(); 
        //version樂觀鎖概念,此處需要獲取version信息，則需要先get拿到節(jié)點(diǎn)信息 
        try { 
            //獲取節(jié)點(diǎn)(修改需要version信息) 
            zooKeeper.getData(key, null, stat); 
            //再修改 
            zooKeeper.setData(key, newValue.getBytes(), stat.getVersion()); 
            ZookeeperTest.closeConnection(zooKeeper); 
        } catch (KeeperException e) { 
            e.printStackTrace(); 
        } catch (InterruptedException e) { 
            e.printStackTrace(); 
        } 
    } 
 
   /** @Author lyy 
     * @Description //TODO 刪除節(jié)點(diǎn) 
     * @Date 14:57 2021/9/29 
     * @Param  
     * @return  
    **/ 
    public void delete(String key) { 
        ZooKeeper zooKeeper = ZookeeperTest.getConntection(); 
        Stat stat = new Stat(); 
        try { 
            //獲取節(jié)點(diǎn)(刪除需要version信息) 
            zooKeeper.getData(key, null, stat); 
            //刪除節(jié)點(diǎn) 
            zooKeeper.delete(key, stat.getVersion()); 
            ZookeeperTest.closeConnection(zooKeeper); 
        } catch (KeeperException e) { 
            e.printStackTrace(); 
        } catch (InterruptedException e) { 
            e.printStackTrace(); 
        } 
 
    } 
}

總結(jié)

到這里我們對Zookeeper 大概有個入門級的了解了，不過Zookeeper遠(yuǎn)遠(yuǎn)比我們這里講述的功能多，如何用Zookeeper實(shí)現(xiàn)集群管理、分布式鎖，隊列等等，小農(nóng)會在后面的文章中進(jìn)行講解，關(guān)注我，后續(xù)精彩內(nèi)容第一時間推送。

責(zé)任編輯：姜華來源：牧小農(nóng)