Zookeeper詳細教程、分布式協調服務原理

Zookeeper詳細教程、分布式協調服務原理

Zookeeper分布式服務架構是Apache Hadoop的一個子項目，主要為分布式系統提供協調服務以及一些資料管理問題，如命名服務、叢集管理、分布式應用配置等。zookeeper可以将簡單易用的接口和高效穩定的系統提供給使用者。

在大型網站中，zookeeper一直占據着重要地位，主要功能如下：

zookeeper是為别的分布式程式服務的

Zookeeper本身就是一個分布式程式（隻要有半數以上節點存活，zk就能正常服務）

Zookeeper所提供的服務涵蓋：主從協調、伺服器節點動态上下線、統一配置管理、分布式共享鎖、統一名稱服務

雖然說可以提供各種服務，但是zookeeper在底層其實隻提供了兩個功能：

為使用者程式提供資料節點監聽服務；

管理(存儲，讀取)使用者程式送出的資料；

一、zookeeper叢集安裝

由于zookeeper的叢集投票選主機制（下面會介紹），超過半數的節點投票才能完成選主。并且必須超過半數的節點存活才能提供服務。是以叢集中節點數最好為奇數台，但不少于3台。我們将我們的zookeeper叢集安裝到三台虛拟機上。

1.1 環境準備

三台虛拟機

192.168.66.101

192.168.66.102

192.168.66.103

JDK安裝包（jdk-7u71-linux-i586.tar.gz）

zookeeper安裝包（zookeeper-3.4.5.tar.gz）

1.2 建立zk使用者

登入三台虛拟機，添加zookeeper的管理使用者，執行如下指令添加一個新使用者，注意必須使用root使用者權限來添加新使用者，需要在三台虛拟機上都要建立一個新使用者。

groupadd zkg #添加一個組

useradd zk -g zkg # 添加一個使用者，并制定該使用者屬于zkg組

passwd # 給使用者設定密碼，下面提示輸入密碼，以及确認密碼

1.3 安裝

三台虛拟機的使用者建立完成之後，在一台虛拟機上來安裝zookeeper，切換到剛建立的使用者下執行如下指令

su zk #切換到zk使用者

cd ~ # 進入zk使用者的home目錄

使用ftp工具将jdk安裝包和zookeeper安裝包上傳至任一台虛拟機中zk使用者home目錄下，并解壓至apps目錄下

mkdir apps

tar -zxvf jdk-7u71-linux-i586.tar.gz -C apps/

tar -zxvf zookeeper-3.4.5.tar.gz -C apps/

按照下面的步驟，配置環境變量，JDK安裝完畢，注意三台機器都需要配置

su # 切換到root使用者

vi /etc/profile # 編輯系統配置檔案，将下面三行内容粘貼至檔案末尾

########################################################

export JAVA_HOME=/home/zk/apps/jdk-7u71-linux-i586

export ZOOKEEPER_HOME=/home/zk/apps/zookeeper-3.4.5

export PATH=$PATH:$ZOOKEEPER_HOME/bin:$JAVA_HOME/bin

#########################################################

source /etc/profile # 使修改後配置檔案生效

su zk # 切換為zk使用者

檢查jdk的安裝是否成功，隻檢查已有加壓檔案的機器

java -version # 檢視jdk版本資訊

修改zookeeper的配置檔案

cd ~/apps/zookeeper-3.4.5/conf # 進入到zookeeper的配置檔案存放目錄

cp zoo_sample.cfg zoo.cfg # 将zoo_sample.cfg複制一份，并更名為zoo.cfg

vi zoo.cfg # 編輯zoo.cfg并将下面内容添加到檔案末尾

server.1=192.168.66.101:2888:3888 # (主機名, 心跳端口、資料端口)

server.2=192.168.66.102:2888:3888

server.3=192.168.66.103:2888:3888

修改dataDir和dataLogDir的值

dataDir=/home/zk/apps/zookeeper-3.4.5/data

dataLogDir=/home/zk/apps/zookeeper-3.4.5/log

:wq #儲存并退出

建立data和log目錄，用于存放資料和日志資訊

cd /home/zk/apps/zookeeper-3.4.5/

mkdir -m 755 data

mkdir -m 755 log

在data檔案夾下建立myid檔案，myid的檔案内容為1

cd data/

echo 1 > myid #建立myid檔案，并輸入内容為1

将配置好的檔案目錄發至其他機器上

scp -r /home/zk/apps [email protected]:/home/zk/ # 通過scp将apps目錄發至其他機器，需要輸入密碼

scp -r /home/zk/apps [email protected]:/home/zk/

修改其他機器上的myid

到192.168.66.102上：修改myid為：2

到192.168.66.103上：修改myid為：3

啟動每台機器上的zookeeper

zkServer.sh start

檢視叢集狀态

jps # 檢視程序

zkServer.sh status #檢視叢集狀态，主從資訊

二、zookeeper的結構和指令

2.1 zookeeper特性

Zookeeper：一個leader，多個follower組成的叢集

全局資料一緻：每個server儲存一份相同的資料副本，client無論連接配接到哪個server，資料都是一緻的

分布式讀寫，更新請求轉發，由leader實施

更新請求順序進行，來自同一個client的更新請求按其發送順序依次執行

資料更新原子性，一次資料更新要麼成功，要麼失敗

實時性，在一定時間範圍内，client能讀到最新資料

2.2 zookeeper資料結構

階層化的目錄結構，命名符合正常檔案系統規範(見下圖)

每個節點在zookeeper中叫做znode,并且其有一個唯一的路徑辨別

節點Znode可以包含資料和子節點（但是EPHEMERAL類型的節點不能有子節點，下一頁詳細講解）

用戶端應用可以在節點上設定螢幕（後續詳細講解）

2.3 節點類型

Znode有兩種類型

短暫（ephemeral）（斷開連接配接自己删除）

持久（persistent）（斷開連接配接不删除）

Znode有四種形式的目錄節點（預設是persistent ）

PERSISTENT

PERSISTENT_SEQUENTIAL（持久序列/test0000000019 ）

EPHEMERAL

EPHEMERAL_SEQUENTIAL

建立znode時設定順序辨別，znode名稱後會附加一個值，順序号是一個單調遞增的計數器，由父節點維護

在分布式系統中，順序号可以被用于為所有的事件進行全局排序，這樣用戶端可以通過順序号推斷事件的順序

2.4 zookeeper指令行操作

運作 zkCli.sh –server <ip>進入指令行工具

zkCli.sh –server 192.168.66.101

ls / #檢視目前 ZooKeeper 中所包含的内容

create /zk "myData" # 建立一個新的 znode ，使用 create /zk myData 。這個指令建立了一個新的 znode 節點“ zk ”以及與它關聯的字元串

get /zk # 我們運作 get 指令來确認 znode 是否包含我們所建立的字元串

get /zk watch

#監聽這個節點的變化,當另外一個用戶端改變/zk時,它會打出下面的

#WATCHER::

#WatchedEvent state:SyncConnected type:NodeDataChanged path:/zk

set /zk "zsl" #通過 set 指令來對 zk 所關聯的字元串進行設定

delete /zk # 删除一個節點

rmr /zk # 删除一個節點

2.5 zookeeper Java api的使用

2.5.1 基本使用

org.apache.zookeeper.Zookeeper是用戶端入口主類，負責建立與server的會話

它提供了表 1 所示幾類主要方法

功能

描述

create

在本地目錄樹中建立一個節點

delete

删除一個節點

exists

測試本地是否存在目标節點

get/set data

從目标節點上讀取 / 寫資料

get/set ACL

擷取 / 設定目标節點通路控制清單資訊

get children

檢索一個子節點上的清單

sync

等待要被傳送的資料

2.5.2 是用Java API實作簡單的增删改查

public class SimpleDemo {

// 會話逾時時間，設定為與系統預設時間一緻

private static final int SESSION_TIMEOUT = 30000;

// 建立 ZooKeeper 執行個體

ZooKeeper zk;

// 建立 Watcher 執行個體

Watcher wh = new Watcher() {

public void process(org.apache.zookeeper.WatchedEvent event)

{

System.out.println(event.toString());

}

};

// 初始化 ZooKeeper 執行個體

private void createZKInstance() throws IOException

zk = new ZooKeeper("weekend01:2181", SimpleDemo.SESSION_TIMEOUT, this.wh);

private void ZKOperations() throws IOException, InterruptedException, KeeperException

System.out.println("/n1. 建立 ZooKeeper 節點 (znode ： zoo2, 資料： myData2 ，權限： OPEN_ACL_UNSAFE ，節點類型： Persistent");

zk.create("/zoo2", "myData2".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

System.out.println("/n2. 檢視是否建立成功： ");

System.out.println(new String(zk.getData("/zoo2", false, null)));

System.out.println("/n3. 修改節點資料 ");

zk.setData("/zoo2", "shenlan211314".getBytes(), -1);

System.out.println("/n4. 檢視是否修改成功： ");

System.out.println("/n5. 删除節點 ");

zk.delete("/zoo2", -1);

System.out.println("/n6. 檢視節點是否被删除： ");

System.out.println(" 節點狀态： [" + zk.exists("/zoo2", false) + "]");

private void ZKClose() throws InterruptedException

zk.close();

public static void main(String[] args) throws IOException, InterruptedException, KeeperException {

SimpleDemo dm = new SimpleDemo();

dm.createZKInstance();

dm.ZKOperations();

dm.ZKClose();

2.5.3 Zookeeper的監聽器工作機制

監聽器是一個接口，我們的代碼中可以實作Wather這個接口，實作其中的process方法，方法中即我們自己的業務邏輯

監聽器的注冊是在擷取資料的操作中實作：

getData(path,watch?)監聽的事件是：節點資料變化事件

getChildren(path,watch?)監聽的事件是：節點下的子節點增減變化事件

三、zookeeper的應用案例（分布式應用HA||分布式鎖）

3.1 實作分布式應用的(主節點HA)及用戶端動态更新主節點狀态

某分布式系統中，主節點可以有多台，可以動态上下線

任意一台用戶端都能實時感覺到主節點伺服器的上下線

A、用戶端實作

public class AppClient {

private String groupNode = "sgroup";

private ZooKeeper zk;

private Stat stat = new Stat();

private volatile List<String> serverList;

/**

• 連接配接zookeeper

  */

  public void connectZookeeper() throws Exception {

  zk

  = new ZooKeeper("localhost:4180,localhost:4181,localhost:4182", 5000, new Watcher() {

  public void process(WatchedEvent event) {

  // 如果發生了"/sgroup"節點下的子節點變化事件, 更新server清單, 并重新注冊監聽

  if (event.getType() == EventType.NodeChildrenChanged

  && ("/" + groupNode).equals(event.getPath())) {

  try {

  updateServerList();

  } catch (Exception e) {

  e.printStackTrace();

  });

• 更新server清單

  private void updateServerList() throws Exception {

  List<String> newServerList = new ArrayList<String>();

  // 擷取并監聽groupNode的子節點變化

  // watch參數為true, 表示監聽子節點變化事件.

  // 每次都需要重新注冊監聽, 因為一次注冊, 隻能監聽一次事件, 如果還想繼續保持監聽, 必須重新注冊

  List<String> subList = zk.getChildren("/" + groupNode, true);

  for (String subNode : subList) {

  // 擷取每個子節點下關聯的server位址

  byte[] data = zk.getData("/" + groupNode + "/" + subNode, false, stat);

  newServerList.add(new String(data, "utf-8"));

  // 替換server清單

  serverList = newServerList;

  System.out.println("server list updated: " + serverList);

• client的工作邏輯寫在這個方法中

• 此處不做任何處理, 隻讓client sleep

  public void handle() throws InterruptedException {

  Thread.sleep(Long.MAX_VALUE);

  public static void main(String[] args) throws Exception {

  AppClient ac = new AppClient();

  ac.connectZookeeper();

  ac.handle();

  B、伺服器端實作

public class AppServer {

private String subNode = "sub";

• @param address server的位址

  public void connectZookeeper(String address) throws Exception {

  ZooKeeper zk = new ZooKeeper(

  "localhost:4180,localhost:4181,localhost:4182",

  5000, new Watcher() {

  // 不做處理

  // 在"/sgroup"下建立子節點

  // 子節點的類型設定為EPHEMERAL_SEQUENTIAL, 表明這是一個臨時節點, 且在子節點的名稱後面加上一串數字字尾

  // 将server的位址資料關聯到新建立的子節點上

  String createdPath = zk.create("/" + groupNode + "/" + subNode, address.getBytes("utf-8"),

  Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

  System.out.println("create: " + createdPath);

• server的工作邏輯寫在這個方法中

• 此處不做任何處理, 隻讓server sleep

  // 在參數中指定server的位址

  if (args.length == 0) {

  System.err.println("The first argument must be server address");

  System.exit(1);

  AppServer as = new AppServer();

  as.connectZookeeper(args[0]);

  as.handle();

  3.2 分布式共享鎖的簡單實作

用戶端A

public class DistributedClient {

// 逾時時間

private static final int SESSION_TIMEOUT = 5000;

// zookeeper server清單

private String hosts = "localhost:4180,localhost:4181,localhost:4182";

private String groupNode = "locks";

// 目前client建立的子節點

private String thisPath;

// 目前client等待的子節點

private String waitPath;

private CountDownLatch latch = new CountDownLatch(1);

• zk = new ZooKeeper(hosts, SESSION_TIMEOUT, new Watcher() {

  // 連接配接建立時, 打開latch, 喚醒wait在該latch上的線程

  if (event.getState() == KeeperState.SyncConnected) {

  latch.countDown();

  // 發生了waitPath的删除事件

  if (event.getType() == EventType.NodeDeleted && event.getPath().equals(waitPath)) {

  doSomething();

  // 等待連接配接建立

  latch.await();

  // 建立子節點

  thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE,

  CreateMode.EPHEMERAL_SEQUENTIAL);

  // wait一小會, 讓結果更清晰一些

  Thread.sleep(10);

  // 注意, 沒有必要監聽"/locks"的子節點的變化情況

  List<String> childrenNodes = zk.getChildren("/" + groupNode, false);

  // 清單中隻有一個子節點, 那肯定就是thisPath, 說明client獲得鎖

  if (childrenNodes.size() == 1) {

  } else {

  String thisNode = thisPath.substring(("/" + groupNode + "/").length());

  // 排序

  Collections.sort(childrenNodes);

  int index = childrenNodes.indexOf(thisNode);

  if (index == -1) {

  // never happened

  } else if (index == 0) {

  // inddx == 0, 說明thisNode在清單中最小, 目前client獲得鎖

  // 獲得排名比thisPath前1位的節點

  this.waitPath = "/" + groupNode + "/" + childrenNodes.get(index - 1);

  // 在waitPath上注冊監聽器, 當waitPath被删除時, zookeeper會回調監聽器的process方法

  zk.getData(waitPath, true, new Stat());

  private void doSomething() throws Exception {

  System.out.println("gain lock: " + thisPath);

  Thread.sleep(2000);

  // do something

  } finally {

  System.out.println("finished: " + thisPath);

  // 将thisPath删除, 監聽thisPath的client将獲得通知

  // 相當于釋放鎖

  zk.delete(this.thisPath, -1);

  for (int i = 0; i < 10; i++) {

  new Thread() {

  public void run() {

  DistributedClient dl = new DistributedClient();

  dl.connectZookeeper();

  }.start();

  分布式多程序模式實作

public class DistributedClientMy {

private String hosts = "spark01:2181,spark02:2181,spark03:2181";

private boolean haveLock = false;

private volatile String thisPath;

• zk = new ZooKeeper("spark01:2181", SESSION_TIMEOUT, new Watcher() {

  // 子節點發生變化

  if (event.getType() == EventType.NodeChildrenChanged && event.getPath().equals("/" + groupNode)) {

  // thisPath是否是清單中的最小節點

  List<String> childrenNodes = zk.getChildren("/" + groupNode, true);

  if (childrenNodes.indexOf(thisNode) == 0) {

  Thread.sleep(new Random().nextInt(1000));

  // 監聽子節點的變化

• 共享資源的通路邏輯寫在這個方法中

  DistributedClientMy dl = new DistributedClientMy();

  四、zookeeper的選舉機制（全新叢集paxos）

以一個簡單的例子來說明整個選舉的過程。

假設有五台伺服器組成的zookeeper叢集,它們的id從1-5,同時它們都是最新啟動的,也就是沒有曆史資料,在存放資料量這一點上,都是一樣的.假設這些伺服器依序啟動,來看看會發生什麼。

伺服器1啟動,此時隻有它一台伺服器啟動了,它發出去的報沒有任何響應,是以它的選舉狀态一直是LOOKING狀态

伺服器2啟動,它與最開始啟動的伺服器1進行通信,互相交換自己的選舉結果,由于兩者都沒有曆史資料,是以id值較大的伺服器2勝出,但是由于沒有達到超過半數以上的伺服器都同意選舉它(這個例子中的半數以上是3),是以伺服器1,2還是繼續保持LOOKING狀态.

伺服器3啟動,根據前面的理論分析,伺服器3成為伺服器1,2,3中的老大,而與上面不同的是,此時有三台伺服器選舉了它,是以它成為了這次選舉的leader.

伺服器4啟動,根據前面的分析,理論上伺服器4應該是伺服器1,2,3,4中最大的,但是由于前面已經有半數以上的伺服器選舉了伺服器3,是以它隻能接收當小弟的命了.

伺服器5啟動,同4一樣,當小弟

五、非全新叢集的選舉機制(資料恢複)

那麼，初始化的時候，是按照上述的說明進行選舉的，但是當zookeeper運作了一段時間之後，有機器down掉，重新選舉時，選舉過程就相對複雜了。

需要加入資料id、leader id和邏輯時鐘。

資料id：資料新的id就大，資料每次更新都會更新id。

Leader id：就是我們配置的myid中的值，每個機器一個。

邏輯時鐘：這個值從0開始遞增,每次選舉對應一個值,也就是說: 如果在同一次選舉中,那麼這個值應該是一緻的 ; 邏輯時鐘值越大,說明這一次選舉leader的程序更新.

選舉的标準就變成：

1、邏輯時鐘小的選舉結果被忽略，重新投票

2、統一邏輯時鐘後，資料id大的勝出