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 所示幾類主要方法

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。