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 zk@192.168.66.102:/home/zk/ # 通過scp將apps目錄發至其他機器，需要輸入密碼
scp -r /home/zk/apps zk@192.168.66.103:/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(new String(zk.getData("/zoo2", false, null)));
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();
  }
  }
  }
  });
  updateServerList();
  }
  /**

• 更新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 groupNode = "sgroup";
private String subNode = "sub";
/**

• 連接zookeeper

• @param address server的地址
  */
  public void connectZookeeper(String address) throws Exception {
  ZooKeeper zk = new ZooKeeper(
  "localhost:4180,localhost:4181,localhost:4182",
  5000, new Watcher() {
  public void process(WatchedEvent event) {
  // 不做處理
  }
  });
  // 在"/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
  */
  public void handle() throws InterruptedException {
  Thread.sleep(Long.MAX_VALUE);
  }
  public static void main(String[] args) throws Exception {
  // 在參數中指定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";
private String subNode = "sub";
private ZooKeeper zk;
// 當前client創建的子節點
private String thisPath;
// 當前client等待的子節點
private String waitPath;
private CountDownLatch latch = new CountDownLatch(1);
/**

• 連接zookeeper
  */
  public void connectZookeeper() throws Exception {
  zk = new ZooKeeper(hosts, SESSION_TIMEOUT, new Watcher() {
  public void process(WatchedEvent event) {
  try {
  // 連接建立時, 打開latch, 喚醒wait在該latch上的線程
  if (event.getState() == KeeperState.SyncConnected) {
  latch.countDown();
  }
  // 發生了waitPath的刪除事件
  if (event.getType() == EventType.NodeDeleted && event.getPath().equals(waitPath)) {
  doSomething();
  }
  } catch (Exception e) {
  e.printStackTrace();
  }
  }
  });
  // 等待連接建立
  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) {
  doSomething();
  } 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獲得鎖
  doSomething();
  } else {
  // 獲得排名比thisPath前1位的節點
  this.waitPath = "/" + groupNode + "/" + childrenNodes.get(index - 1);
  // 在waitPath上注冊監聽器, 當waitPath被刪除時, zookeeper會回調監聽器的process方法
  zk.getData(waitPath, true, new Stat());
  }
  }
  }
  private void doSomething() throws Exception {
  try {
  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);
  }
  }
  public static void main(String[] args) throws Exception {
  for (int i = 0; i < 10; i++) {
  new Thread() {
  public void run() {
  try {
  DistributedClient dl = new DistributedClient();
  dl.connectZookeeper();
  } catch (Exception e) {
  e.printStackTrace();
  }
  }
  }.start();
  }
  Thread.sleep(Long.MAX_VALUE);
  }
  }
  分布式多進程模式實現

public class DistributedClientMy {
// 超時時間
private static final int SESSION_TIMEOUT = 5000;
// zookeeper server列表
private String hosts = "spark01:2181,spark02:2181,spark03:2181";
private String groupNode = "locks";
private String subNode = "sub";
private boolean haveLock = false;
private ZooKeeper zk;
// 當前client創建的子節點
private volatile String thisPath;
/**

• 連接zookeeper
  */
  public void connectZookeeper() throws Exception {
  zk = new ZooKeeper("spark01:2181", SESSION_TIMEOUT, new Watcher() {
  public void process(WatchedEvent event) {
  try {
  // 子節點發生變化
  if (event.getType() == EventType.NodeChildrenChanged && event.getPath().equals("/" + groupNode)) {
  // thisPath是否是列表中的最小節點
  List<String> childrenNodes = zk.getChildren("/" + groupNode, true);
  String thisNode = thisPath.substring(("/" + groupNode + "/").length());
  // 排序
  Collections.sort(childrenNodes);
  if (childrenNodes.indexOf(thisNode) == 0) {
  doSomething();
  thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE,
  CreateMode.EPHEMERAL_SEQUENTIAL);
  }
  }
  } catch (Exception e) {
  e.printStackTrace();
  }
  }
  });
  // 創建子節點
  thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE,
  CreateMode.EPHEMERAL_SEQUENTIAL);
  // wait一小會, 讓結果更清晰一些
  Thread.sleep(new Random().nextInt(1000));
  // 監聽子節點的變化
  List<String> childrenNodes = zk.getChildren("/" + groupNode, true);
  // 列表中只有一個子節點, 那肯定就是thisPath, 說明client獲得鎖
  if (childrenNodes.size() == 1) {
  doSomething();
  thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE,
  CreateMode.EPHEMERAL_SEQUENTIAL);
  }
  }
  /**

• 共享資源的訪問邏輯寫在這個方法中
  */
  private void doSomething() throws Exception {
  try {
  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);
  }
  }
  public static void main(String[] args) throws Exception {
  DistributedClientMy dl = new DistributedClientMy();
  dl.connectZookeeper();
  Thread.sleep(Long.MAX_VALUE);
  }
  }
  四、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大的勝出

3、數據id相同的情況下，leader id大的勝出

根據這個規則選出leader。