學習目標：

什么是zookeeper
應用場景
基本的操作
安裝部署
shell客戶端
java端
基本原理
選舉機制
資料一致性
資料的讀寫流程

1 zookeeper簡介

zookeeper是一個底層的集群協調工具，（比如：NN和DN之間的狀態感應；監控通知）！

具備基本的功能有，記錄用戶的狀態資料（寫），回傳用戶的資料（讀），監控和通知，zookeeper為了高可用和安全性是一個集群！ 3臺（每個節點的資料完全一致）！

2 應用場景

寫資料
讀取資料
監控通知

2.1 服務狀態感知

2.2 分布式鎖

2.3 分布式配置同步

2.4 統一域名

分布式中統一CMD指令

分布式中服務器的狀態感知

3 操作

3.1 安裝

zookeeper在部署的時候選擇奇數臺！要求zk集群半數以上的機器在線才能正常作業！

上傳解壓

配置

在zk安裝目錄下創建檔案夾 zkData mkdir zkData /opt/apps/zookeeper-3.4.6/zkData
修改conf/下的配置模板組態檔名 mv zoo_sample.cfg zoo.cfg

vi zoo.cfg

dataDir=/opt/apps/zookeeper-3.4.6/zkData

# Set to "0" to disable auto purge feature
#autopurge.purgeInterval=1
server.1=linux01:2888:3888
server.2=linux02:2888:3888
server.3=linux03:2888:3888

在zkData檔案夾下創建myid檔案 echo 1 > zkData/myid

同步安裝包

scp -r  zookeeper-3.4.6/  linux02:$PWD
scp -r  zookeeper-3.4.6/  linux03:$PWD

修改各個節點myid的值 linux02【2】 linux03【3】

啟動集群

在各個節點上啟動ZK服務

/opt/apps/zookeeper-3.4.6/bin/zkServer.sh  start

查看行程 jps

[root@linux01 bin]# jps
1697 QuorumPeerMain
1720 Jps

查看集群狀態在每個節點執行

/opt/apps/zookeeper-3.4.6/bin/zkServer.sh  status

-------------------------------------------------------
Mode: follower

Mode: leader

Mode: follower

3.2 撰寫一鍵啟動腳本

#!/bin/bash
# zk啟停腳本 
for hostname in linux01  linux02  linux03
do 
echo "連接${hostname}... ...正在執行 $1"
ssh ${hostname} "source  /etc/profile ;/opt/apps/zookeeper-3.4.6/bin/zkServer.sh  $1 ;exit"
done

3.3 shell操作

zookeeper中記錄資料以Tree節點的形式存盤資料的/類似于目錄樹!

目錄叫znode : 節點的組織資料是 K V

基本操作指令

# 客戶端連接 
bin/zkCli.sh 連接到本地zk服務
bin/zkCli.sh -server  linux02:2181  連接到執行節點的服務
quit  退出客戶端      ctrl+c
-------------------------------------------
help 幫助命令  查看系統支持的所有的命令
ZooKeeper -server host:port cmd args
        stat path [watch]
        set path data [version]
        ls path [watch] * 
        delquota [-n|-b] path
        ls2 path [watch]  * 
        setAcl path acl
        setquota -n|-b val path
        history 
        redo cmdno
        printwatches on|off
        delete path [version]  * 
        sync path
        listquota path
        rmr path  * 
        get path [watch]
        create [-s] [-e] path data acl  * 
        addauth scheme auth
        quit   * 
        getAcl path
        close   * 
        connect host:port

-------------------------------------------------------------------
創建節點  節點必須以/開頭  (路徑必須是絕對路徑)
--   create [-s] [-e] path data acl
create  /a  1 
create  /b  2 
create  /a/a1 11
create  /a/a2 22
create  /a/a3 33 
[-s] 在節點后面添加一個自增的id  防止節點名重復而創建失敗
[-e]  創建的節點是臨時節點  , 只在當前連接中有效
------節點分類-----
--臨時節點
     臨時有序   -e
     臨時無序   -e  -s
--永久節點
     永久有序    -s
     永久無序   默認
注意: 節點的創建不能層級創建
查看節點串列 
ls    /a 
ls2   /a  顯示資料版本 , 事務id 創建時間等資訊

修改資料 
set  /a  123
獲取資料 
get  /a 
洗掉節點  
delete   /a  只能洗掉空節點
rmr      /a  可以洗掉任意節點

監聽和通知

事件 : 觸發了某種事件通知

節點資料的變化
子節點個數的變化

ls path [watch] 監控指定路徑下資料的變化
get path [watch] 監控指定路徑資料的變化

注意: 監控和通知只能1次

打開兩個zk客戶端
1) 一個監控   2) 一個修改資料 

#  子節點個數變化
視窗01  :  ls  /   watch  
          ->(WatchedEvent state:SyncConnected type:NodeChildrenChangedpath:/)
視窗02  :  rmr  /a000000004 
---------------------------------------------------------------------------
#  節點資料變化
視窗01  :   get  /b   watch  
          ->(WatchedEvent state:SyncConnected type:NodeDataChanged path:/b)
視窗02  :  set /b   321

3.4 java操作

添加依賴
獲取zk客戶端
呼叫API
釋放資源

添加依賴

<dependency>
  <groupId>org.apache.zookeeper</groupId>
  <artifactId>zookeeper</artifactId>
  <version>3.4.6</version>
</dependency>

入門示例-[獲取連接\創建節點]

package com.doitedu.zk.cli;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import java.io.IOException;
/**
 * @Date 2022/2/23
 * @Created by HANGGE
 * @Description TODO
 */
public class Demo01 {
    public static void main(String[] args) throws Exception {
        // 獲取zk的客戶端
        /**
         * 引數一 zk的地址  host:port,host2:port
         * 引數二 連接的超時時間  毫秒
         * 引數三 連接成功后的監聽
         */
        ZooKeeper zk = new ZooKeeper("linux01:2181,linux02:2181,linux03:2181", 3000, null);
        // 呼叫API
        //1  創建節點
        /**
         * 引數1   路徑
         * 引數2   資料 值 位元組陣列
         * 引數3   權限
         * 引數4   節點型別  1 2 3 4
         */
        zk.create("/doitedu" , "JINGYINGDOIT30_".getBytes() , ZooDefs.Ids.OPEN_ACL_UNSAFE , CreateMode.EPHEMERAL_SEQUENTIAL);
      //  System.out.println(zk);
        //釋放資源
        Thread.sleep(8000);
        zk.close();

    }
}

zookeeper-Java-API使用

package com.doitedu.zk.cli;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import java.util.List;

/**
 * @Date 2022/2/23
 * @Created by HANGGE
 * @Description
 * 創建節點
 * 洗掉節點
 * 獲取值
 * 修改值
 * 遍歷子節點串列
 *   判斷節點是否存在
 */
public class ClientApiDemo {
    public static void main(String[] args) throws Exception {
        // 獲取客戶端物件
        ZooKeeper zk = ZookeeperUtil.getZookeeper();
        // testGetData(zk);
        //修改資料
        //testSetData(zk);
        //獲取子節點串列
        //testls(zk);
        // 只能洗掉空節點   遞回
       // zk.delete("/b" , -1);
       // 遞回洗掉節點
        ZookeeperUtil.rmr(zk,"/b") ;
        zk.close();
    }
    // 查看路徑下的子節點串列
    public static void testls(ZooKeeper zk) throws KeeperException, InterruptedException {
        List<String> ls = zk.getChildren("/", null);
        if(ls!=null && ls.size()>0){
            for (String name : ls) {
                System.out.println("節點的名字是:" + name);
                byte[] data = zk.getData("/" + name, null, null);
                System.out.println(new String(data));
            }
        }
    }
    // 更新資料
    public static void testSetData(ZooKeeper zk) throws KeeperException, InterruptedException {
        Stat stat = zk.setData("/b", "liulan".getBytes(), -1);
        // 如果更新成功Stat 不為null
        if(stat != null){
            testGetData(zk);
        }
    }

    // 獲取資料
    public static void testGetData(ZooKeeper zk) throws KeeperException, InterruptedException {

        byte[] data = zk.getData("/b", null, null);
        String str = new String(data);
        System.out.println(str);
    }
}

封裝工具類

package com.doitedu.zk.cli;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;
import java.io.IOException;
import java.util.List;
/**
 * @Date 2022/2/23
 * @Created by HANGGE
 * @Description TODO
 */
public class ZookeeperUtil {
    /**
     * 獲取指定zk集群的客戶端物件
     *   linux01  02  03
     * @return
     * @throws IOException
     */
    public  static ZooKeeper  getZookeeper() throws IOException {
        return  new ZooKeeper("linux01:2181,linux02:2181,linux03:2181", 3000, null);
    }

    /**
     * 洗掉任意節點
     * @param zk
     * @param path
     * @throws KeeperException
     * @throws InterruptedException
     */
    public static void  rmr(ZooKeeper zk , String path) throws KeeperException, InterruptedException {
        // 遍歷是否有子節點
        List<String> ls = zk.getChildren(path, null);
        // 有子節點
        if(ls!=null &&  ls.size() >0){
            // 洗掉子節點
            for (String name : ls) {
                //---遞回
                rmr(zk , path+"/"+name);
            }
        }
        //沒有子節點
        zk.delete(path , -1);
    }
}

監控和通知

連接成功監聽通知
資料變化
節點變化

例子

package com.doitedu.zk.cli;

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

import java.io.IOException;

/**
 * @Date 2022/2/23
 * @Created by HANGGE
 * @Description 連接物件獲取成功 事件通知
 */
public class TestWatcher01 {

    public static void main(String[] args) throws Exception {
        /**
         * 引數三  事件通知  監聽器
         *  在獲取物件的時候使用  當連接成功以后會  回呼一次process方法
         */
        ZooKeeper zk = new ZooKeeper("linux11:2181", 2000, new Watcher() {
            public void process(WatchedEvent event) {
                System.out.println("獲取連接成功......");
            }
        });
        System.out.println(zk);
        zk.close();


    }


}

----------------------------------------------------------------------
package com.doitedu.zk.cli;

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

/**
 * @Date 2022/2/23
 * @Created by HANGGE
 * @Description
 * 監聽節點資料的變化
 *   當監聽的節點的資料發生變化    通知客戶端  回呼process
 */
public class TestWatcher02 {

    public static void main(String[] args) throws Exception {
        /**
         * 引數三  事件通知  監聽器
         *  在獲取物件的時候使用  當連接成功以后會  回呼一次process方法
         */
        final ZooKeeper zk = new ZooKeeper("linux01:2181", 2000, null);
        // 獲取指定節點資料的時候  系結監聽器  監聽此節點資料的變化
        byte[] data = zk.getData("/b", new Watcher() {
            // /b節點資料變化就會執行這個方法   1次
            public void process(WatchedEvent event) {
                try {
                    System.out.println("/b的資料發生了變化....");
                    System.out.println(event.getType());
                    // 改變后的資料是
                    byte[] data1 = zk.getData("/b", this, null);
                    System.out.println("變化后的的資料是: "+new String(data1));
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }, null);
        System.out.println("變化以前的資料是: "+new String(data));

        Thread.sleep(Integer.MAX_VALUE);
        zk.close();
    }


}
-----------------------------------------------------------------------
package com.doitedu.zk.cli;

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

import java.util.List;

/**
 * @Date 2022/2/23
 * @Created by HANGGE
 * @Description
 * 監聽節點資料的變化
 *   當監聽的節點的資料發生變化    通知客戶端  回呼process
 */
public class TestWatcher03 {

    public static void main(String[] args) throws Exception {
        /**
         * 引數三  事件通知  監聽器
         *  在獲取物件的時候使用  當連接成功以后會  回呼一次process方法
         */
        final ZooKeeper zk = new ZooKeeper("linux01:2181", 2000, null);

        List<String> ls = zk.getChildren("/", new Watcher() {
            // 當 / 節點的子節點 被洗掉  添加 執行這個方法
            public void process(WatchedEvent event) {
                try {
                    System.out.println(event.getType());
                    List<String> ls2 = zk.getChildren("/", this);
                    System.out.println("變化后的節點有:"+ls2);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });

        System.out.println("當前的節點有: "+ls);


        Thread.sleep(Integer.MAX_VALUE);
        zk.close();
    }


}

4 原理基礎

4.1 選舉機制

在ZK中有三種狀態 ;

Leader: 優先負責寫資料

follower 存盤資料參與選舉

observe 觀察者不參與選舉存盤資料

每個服務器都有一個唯一的myid標記

初始啟動選舉

ZK服務啟動[linux01(myid=1)] , 發現集群中沒有Leader .進入到選舉狀態 ; 投自己一票
myid=1的機器獲取1票 , 票數沒有過半, 不能當選leader
linux01, 廣播投票結果
linux02(myid=2)啟動 ,發現集群中沒有leader , 進入到選舉狀態
收到linux01的1票資訊 , 發現linux02的myid > linux01的myid ;投自己一票 ,廣播
linux01收到廣播 ,發現自己的myid小 ; 改投lnux02 1票 , 廣播
linux02 收到linux01的廣播 ,自己的票數為2 , 過半 ,切換leader狀態
linux01 切換follower狀態
linux03啟動 ,發現集群中有leader ;切換follower狀態

運行程序中選舉

注意: 要求整個集群中的資料是一致的! 有事務的保證! zxid事務id

1 先比較zxid的大小 , zxid大的優先當選

2 如果事務id一致 , 再比較myid

整個集群中有機器宕機 ,屬于不正常的狀態! 修復.........

資料一致性

zk中存盤資料以KV的形式 ; K是以節點的組織的以/開頭

整個集群中所有的節點存盤一樣的資料

記錄的資料一般為狀態資料 ,配置資料 ,命令....資料量不大

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/431492.html

標籤：其他

上一篇：flink如何利用checkpoint保證資料狀態一致性

下一篇：入門ClickHouse和Elasticsearch

zookeeper隨堂筆記