大數據技術之Zookeeper
1. 一 Zookeeper概述
1.1 概述
Zookeeper是一個開源的分布式的，為分布式應用提供協調服務的Apache項目。
1.2 特點
1）Zookeeper：一個領導者（leader），多個跟隨者（follower）組成的集群。
2）Leader負責進行投票的發起和決議，更新系統狀態
3）Follower用於接收客戶請求並向客戶端返回結果，在選舉Leader過程中參與投票
4）集群中只要有半數以上節點存活，Zookeeper集群就能正常服務。(奇數台)
5）全局數據一致：每個server保存一份相同的數據副本，client無論連接到哪個server，數據都是一致的。
6）更新請求順序進行，來自同一個client的更新請求按其發送順序依次執行。
7）數據更新原子性，一次數據更新要么成功，要么失敗。
8）實時性，在一定時間范圍內，client能讀到最新數據。
1.3 數據結構
ZooKeeper數據模型的結構與Unix文件系統很類似，整體上可以看作是一棵樹，每個節點稱做一個ZNode。每一個ZNode默認能夠存儲1MB的數據，每個ZNode都可以通過其路徑唯一標識
1.4 應用場景
提供的服務包括：統一命名服務、統一配置管理、統一集群管理、服務器節點動態上下線、軟負載均衡等。
1.5 下載地址
1）官網首頁：
https://zookeeper.apache.org/
二 Zookeeper安裝
2.1 本地模式安裝部署
1）安裝前准備：
（1）安裝jdk
（2）通過SecureCRT工具拷貝zookeeper到linux系統下
（3）修改tar包權限
chmod u+x zookeeper-3.4.10.tar.gz
（4）解壓到指定目錄
[root@hadoop102 software]$ tar -zxvf zookeeper-3.4.10.tar.gz -C /opt/
2）配置修改
將/opt/zookeeper-3.4.10/conf這個路徑下的zoo_sample.cfg修改為zoo.cfg；
進入zoo.cfg文件：vim zoo.cfg
修改dataDir路徑為
dataDir=/opt/zookeeper-3.4.10/zkData
在/opt/zookeeper-3.4.10/這個目錄上創建zkData文件夾
mkdir zkData
3）操作zookeeper
（1）啟動zookeeper
[root@hadoop102 zookeeper-3.4.10]$ bin/zkServer.sh start
（2）查看進程是否啟動
[root@hadoop102 zookeeper-3.4.10]$ jps
4020 Jps
4001 QuorumPeerMain
（3）查看狀態：
[root@hadoop102 zookeeper-3.4.10]$ bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /opt/zookeeper-3.4.10/bin/../conf/zoo.cfg
Mode: standalone
（4）啟動客戶端：
[root@hadoop102 zookeeper-3.4.10]$ bin/zkCli.sh
（5）退出客戶端：
[zk: localhost:2181(CONNECTED) 0] quit
（6）停止zookeeper
[root@hadoop102 zookeeper-3.4.10]$ bin/zkServer.sh stop
2.2 配置參數解讀
解讀zoo.cfg 文件中參數含義
1）tickTime：通信心跳數，Zookeeper服務器心跳時間，單位毫秒
Zookeeper使用的基本時間，服務器之間或客戶端與服務器之間維持心跳的時間間隔，也就是每個tickTime時間就會發送一個心跳，時間單位為毫秒。
它用於心跳機制，並且設置最小的session超時時間為兩倍心跳時間。(session的最小超時時間是2*tickTime)
2）initLimit：LF初始通信時限
集群中的follower跟隨者服務器(F)與leader領導者服務器(L)之間初始連接時能容忍的最多心跳數（tickTime的數量），用它來限定集群中的Zookeeper服務器連接到Leader的時限。
投票選舉新leader的初始化時間
Follower在啟動過程中，會從Leader同步所有最新數據，然后確定自己能夠對外服務的起始狀態。
Leader允許F在initLimit時間內完成這個工作。
3）syncLimit：LF同步通信時限
集群中Leader與Follower之間的最大響應時間單位，假如響應超過syncLimit * tickTime，
Leader認為Follwer死掉，從服務器列表中刪除Follwer。
在運行過程中，Leader負責與ZK集群中所有機器進行通信，例如通過一些心跳檢測機制，來檢測機器的存活狀態。
如果L發出心跳包在syncLimit之后，還沒有從F那收到響應，那么就認為這個F已經不在線了。
4）dataDir：數據文件目錄+數據持久化路徑
保存內存數據庫快照信息的位置，如果沒有其他說明，更新的事務日志也保存到數據庫。
5）clientPort=2181：客戶端連接端口
監聽客戶端連接的端口
三 Zookeeper內部原理
3.1 選舉機制
啟動新集群的選舉機制
當leader掛掉之后的選舉機制(邏輯時鍾等等)
1）半數機制：集群中半數以上機器存活，集群可用。所以zookeeper適合裝在奇數台機器上。
2）Zookeeper雖然在配置文件中並沒有指定master和slave。但是，zookeeper工作時，是有一個節點為leader，其他則為follower，Leader是通過內部的選舉機制臨時產生的
3）以一個簡單的例子來說明整個選舉的過程。
假設有五台服務器組成的zookeeper集群，它們的id從1-5，同時它們都是最新啟動的，也就是沒有歷史數據，在存放數據量這一點上，都是一樣的。假設這些服務器依序啟動，來看看會發生什么。
（1）服務器1啟動，此時只有它一台服務器啟動了，它發出去的報沒有任何響應，所以它的選舉狀態一直是LOOKING狀態。
（2）服務器2啟動，它與最開始啟動的服務器1進行通信，互相交換自己的選舉結果，由於兩者都沒有歷史數據，所以id值較大的服務器2勝出，但是由於沒有達到超過半數以上的服務器都同意選舉它(這個例子中的半數以上是3)，所以服務器1、2還是繼續保持LOOKING狀態。
（3）服務器3啟動，根據前面的理論分析，服務器3成為服務器1、2、3中的老大，而與上面不同的是，此時有三台服務器選舉了它，所以它成為了這次選舉的leader。
（4）服務器4啟動，根據前面的分析，理論上服務器4應該是服務器1、2、3、4中最大的，但是由於前面已經有半數以上的服務器選舉了服務器3，所以它只能接收當小弟的命了。
（5）服務器5啟動，同4一樣當小弟。
非全新集群的選舉機制(數據恢復)
那么，初始化的時候，是按照上述的說明進行選舉的，但是當zookeeper運行了一段時間之后，有機器down掉，重新選舉時，選舉過程就相對復雜了。
需要加入數據id、leader id和邏輯時鍾。
數據id：數據新的id就大，數據每次更新都會更新id。
Leader id：就是我們配置的myid中的值，每個機器一個。
邏輯時鍾：這個值從0開始遞增,每次選舉對應一個值,也就是說: 如果在同一次選舉中,那么這個值應該是一致的 ; 邏輯時鍾值越大,說明這一次選舉leader的進程更新.
選舉的標准就變成：
1、邏輯時鍾小的選舉結果被忽略，重新投票
2、統一邏輯時鍾后，數據id大的勝出
3、數據id相同的情況下，leader id大的勝出
根據這個規則選出leader。
3.2 節點類型
1）Znode有兩種類型：
短暫（ephemeral）：客戶端和服務器端斷開連接后，創建的節點自己刪除
持久（persistent）：客戶端和服務器端斷開連接后，創建的節點不刪除
2）Znode有四種形式的目錄節點（默認是persistent ）
（1）持久化目錄節點（PERSISTENT）
客戶端與zookeeper斷開連接后，該節點依舊存在
（2）持久化順序編號目錄節點（PERSISTENT_SEQUENTIAL）
客戶端與zookeeper斷開連接后，該節點依舊存在，只是Zookeeper給該節點名稱進行順序編號
（3）臨時目錄節點（EPHEMERAL）
客戶端與zookeeper斷開連接后，該節點被刪除
（4）臨時順序編號目錄節點（EPHEMERAL_SEQUENTIAL）
客戶端與zookeeper斷開連接后，該節點被刪除，只是Zookeeper給該節點名稱進行順序編號
3）創建znode時設置順序標識，znode名稱后會附加一個值，順序號是一個單調遞增的計數器，由父節點維護
4）在分布式系統中，順序號可以被用於為所有的事件進行全局排序，這樣客戶端可以通過順序號推斷事件的順序
3.3 stat結構體
1）czxid- 引起這個znode創建的zxid，創建節點的事務的zxid
每次修改ZooKeeper狀態都會收到一個zxid形式的時間戳，也就是ZooKeeper事務ID。
事務ID是ZooKeeper中所有修改總的次序。每個修改都有唯一的zxid，如果zxid1小於zxid2，那么zxid1在zxid2之前發生。
2）ctime - znode被創建的毫秒數(從1970年開始)
3）mzxid - znode最后更新的zxid
4）mtime - znode最后修改的毫秒數(從1970年開始)
5）pZxid-znode最后更新的子節點zxid
6）cversion - znode子節點變化號，znode子節點修改次數
7）dataversion - znode數據變化號
8）aclVersion - znode訪問控制列表的變化號
9）ephemeralOwner- 如果是臨時節點，這個是znode擁有者的session id。如果不是臨時節點則是0。
10）dataLength- znode的數據長度
11）numChildren - znode子節點數量
3.4 監聽器原理
3.5 寫數據流程
四 Zookeeper實戰
4.1 分布式安裝部署
0）集群規划
在hadoop102、hadoop103和hadoop104三個節點上部署Zookeeper。
1）解壓安裝
（1）解壓zookeeper安裝包到/opt/目錄下
[root@hadoop102 software]$ tar -zxvf zookeeper-3.4.10.tar.gz -C /opt/
（2）在/opt/zookeeper-3.4.10/這個目錄下創建zkData
mkdir zkData
（3）重命名/opt/zookeeper-3.4.10/conf這個目錄下的zoo_sample.cfg為zoo.cfg
mv zoo_sample.cfg zoo.cfg
2）配置zoo.cfg文件
（1）具體配置
dataDir=/opt/zookeeper-3.4.10/zkData
增加如下配置
#######################cluster##########################
server.1=houda01:2888:3888
server.2=houda02:2888:3888
server.3=houda03:2888:3888
（2）配置參數解讀
Server.A=B:C:D。
A是一個數字，表示這個是第幾號服務器；
B是這個服務器的ip地址；
C是這個服務器與集群中的Leader服務器交換信息的端口；
D是萬一集群中的Leader服務器掛了，需要一個端口來重新進行選舉，選出一個新的Leader，而這個端口就是用來執行選舉時服務器相互通信的端口。
集群模式下配置一個文件myid，這個文件在dataDir目錄下，這個文件里面有一個數據就是A的值，Zookeeper啟動時讀取此文件，拿到里面的數據與zoo.cfg里面的配置信息比較從而判斷到底是哪個server。
3）集群操作
（1）在/opt/zookeeper-3.4.10/zkData目錄下創建一個myid的文件
touch myid
添加myid文件，注意一定要在linux里面創建，在notepad++里面很可能亂碼
（2）編輯myid文件
vi myid
在文件中添加與server對應的編號：如2
（3）拷貝配置好的zookeeper到其他機器上
scp -r zookeeper-3.4.10/ root@hadoop103.root.com:/opt/app/
scp -r zookeeper-3.4.10/ root@hadoop104.root.com:/opt/app/
並分別修改myid文件中內容為3、4
（4）分別啟動zookeeper
[root@hadoop102 zookeeper-3.4.10]# bin/zkServer.sh start
[root@hadoop103 zookeeper-3.4.10]# bin/zkServer.sh start
[root@hadoop104 zookeeper-3.4.10]# bin/zkServer.sh start
（5）查看狀態
[root@hadoop102 zookeeper-3.4.10]# bin/zkServer.sh status
JMX enabled by default
Using config: /opt/zookeeper-3.4.10/bin/../conf/zoo.cfg
Mode: follower
[root@hadoop103 zookeeper-3.4.10]# bin/zkServer.sh status
JMX enabled by default
Using config: /opt/zookeeper-3.4.10/bin/../conf/zoo.cfg
Mode: leader
[root@hadoop104 zookeeper-3.4.5]# bin/zkServer.sh status
JMX enabled by default
Using config: /opt/zookeeper-3.4.10/bin/../conf/zoo.cfg
Mode: follower
4.2 客戶端命令行操作
1）啟動客戶端
[root@hadoop103 zookeeper-3.4.10]$ bin/zkCli.sh
2）顯示所有操作命令
[zk: localhost:2181(CONNECTED) 1] help
3）查看當前znode中所包含的內容
[zk: localhost:2181(CONNECTED) 0] ls /
[zookeeper]
4）查看當前節點數據並能看到更新次數等數據
[zk: localhost:2181(CONNECTED) 1] ls2 /
[zookeeper]
cZxid = 0x0
ctime = Thu Jan 01 08:00:00 CST 1970
mZxid = 0x0
mtime = Thu Jan 01 08:00:00 CST 1970
pZxid = 0x0
cversion = -1
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 0
numChildren = 1
5）創建普通節點
[zk: localhost:2181(CONNECTED) 2] create /app1 "hello app1"
Created /app1
[zk: localhost:2181(CONNECTED) 4] create /app1/server101 "192.168.1.101"
Created /app1/server101
6）獲得節點的值
[zk: localhost:2181(CONNECTED) 6] get /app1
hello app1
cZxid = 0x20000000a
ctime = Mon Jul 17 16:08:35 CST 2017
mZxid = 0x20000000a
mtime = Mon Jul 17 16:08:35 CST 2017
pZxid = 0x20000000b
cversion = 1
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 10
numChildren = 1
[zk: localhost:2181(CONNECTED) 8] get /app1/server101
192.168.1.101
cZxid = 0x20000000b
ctime = Mon Jul 17 16:11:04 CST 2017
mZxid = 0x20000000b
mtime = Mon Jul 17 16:11:04 CST 2017
pZxid = 0x20000000b
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 13
numChildren = 0
7）創建短暫節點
[zk: localhost:2181(CONNECTED) 9] create -e /app-emphemeral 8888
（1）在當前客戶端是能查看到的
[zk: localhost:2181(CONNECTED) 10] ls /
[app1, app-emphemeral, zookeeper]
（2）退出當前客戶端然后再重啟啟動客戶端
[zk: localhost:2181(CONNECTED) 12] quit
[root@hadoop104 zookeeper-3.4.10]$ bin/zkCli.sh
（3）再次查看根目錄下短暫節點已經刪除
[zk: localhost:2181(CONNECTED) 0] ls /
[app1, zookeeper]
8）創建帶序號的節點
（1）先創建一個普通的根節點app2
[zk: localhost:2181(CONNECTED) 11] create /app2 "app2"
（2）創建帶序號的節點
[zk: localhost:2181(CONNECTED) 13] create -s /app2/aa 888
Created /app2/aa0000000000
[zk: localhost:2181(CONNECTED) 14] create -s /app2/bb 888
Created /app2/bb0000000001
[zk: localhost:2181(CONNECTED) 15] create -s /app2/cc 888
Created /app2/cc0000000002
如果原節點下有1個節點，則再排序時從1開始，以此類推。
[zk: localhost:2181(CONNECTED) 16] create -s /app1/aa 888
Created /app1/aa0000000001
9）修改節點數據值
[zk: localhost:2181(CONNECTED) 2] set /app1 999
10）節點的值變化監聽
（1）在104主機上注冊監聽/app1節點數據變化
[zk: localhost:2181(CONNECTED) 26] get /app1 watch
（2）在103主機上修改/app1節點的數據
[zk: localhost:2181(CONNECTED) 5] set /app1 777
（3）觀察104主機收到數據變化的監聽
WATCHER::
WatchedEvent state:SyncConnected type:NodeDataChanged path:/app1
11）節點的子節點變化監聽（路徑變化）
（1）在104主機上注冊監聽/app1節點的子節點變化
[zk: localhost:2181(CONNECTED) 1] ls /app1 watch
[aa0000000001, server101]
（2）在103主機/app1節點上創建子節點
[zk: localhost:2181(CONNECTED) 6] create /app1/bb 666
Created /app1/bb
（3）觀察104主機收到子節點變化的監聽
WATCHER::
WatchedEvent state:SyncConnected type:NodeChildrenChanged path:/app1
12）刪除節點
[zk: localhost:2181(CONNECTED) 4] delete /app1/bb
13）遞歸刪除節點
[zk: localhost:2181(CONNECTED) 7] rmr /app2
14）查看節點狀態
[zk: localhost:2181(CONNECTED) 12] stat /app1
cZxid = 0x20000000a
ctime = Mon Jul 17 16:08:35 CST 2017
mZxid = 0x200000018
mtime = Mon Jul 17 16:54:38 CST 2017
pZxid = 0x20000001c
cversion = 4
dataVersion = 2
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 3
numChildren = 2
4.3 API應用
4.3.2 創建ZooKeeper客戶端：
private static String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
private static int sessionTimeout = 2000;
private ZooKeeper zkClient = null;
@Before
public void init() throws Exception {
zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
@Override
public void process(WatchedEvent event) {
// 收到事件通知后的回調函數（用戶的業務邏輯）
System.out.println(event.getType() + "--" + event.getPath());
}
});
}
4.3.3 創建子節點
// 創建子節點
@Test
public void create() throws Exception {
// 數據的增刪改查
// 參數1：要創建的節點的路徑； 參數2：節點數據 ； 參數3：節點權限 ；參數4：節點的類型
String nodeCreated = zkClient.create("/idea", "hello zk".getBytes(), Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
}
4.3.4 獲取子節點
// 獲取子節點
@Test
public void getChildren() throws Exception {
List<String> children = zkClient.getChildren("/", true);
for (String child : children) {
System.out.println(child);
}
// 延時阻塞
Thread.sleep(Long.MAX_VALUE);
}
4.3.5 判斷znode是否存在
// 判斷znode是否存在
@Test
public void exist() throws Exception {
Stat stat = zkClient.exists("/idea", false);
System.out.println(stat == null ? "not exist" : "exist");
}
4.4 案例實戰
4.4.1 監聽服務器節點動態上下線案例
1）需求：某分布式系統中，主節點可以有多台，可以動態上下線，任意一台客戶端都能實時感知到主節點服務器的上下線
2）需求分析
3）具體實現：
（0）先在集群上創建/servers節點
[zk: localhost:2181(CONNECTED) 10] create /servers "servers"
Created /servers
（1）服務器端代碼
package com.root.zkcase;
import java.io.IOException;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooDefs.Ids;
public class DistributeServer {
private static String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
private static int sessionTimeout = 2000;
private ZooKeeper zk = null;
private String parentNode = "/servers";
// 創建到zk的客戶端連接
public void getConnect() throws IOException{
zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
@Override
public void process(WatchedEvent event) {
}
});
}
// 注冊服務器
public void registServer(String hostname) throws Exception{
String create = zk.create(parentNode + "/server", hostname.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
System.out.println(hostname +" is online "+ create);
}
// 業務功能
public void business(String hostname) throws Exception{
System.out.println(hostname+" is working ...");
Thread.sleep(Long.MAX_VALUE);
}
public static void main(String[] args) throws Exception {
// 獲取zk連接
DistributeServer server = new DistributeServer();
server.getConnect();
// 利用zk連接注冊服務器信息
server.registServer(args[0]);
// 啟動業務功能
server.business(args[0]);
}
}
（2）客戶端代碼
package com.root.zkcase;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
public class DistributeClient {
private static String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
private static int sessionTimeout = 2000;
private ZooKeeper zk = null;
private String parentNode = "/servers";
// 創建到zk的客戶端連接
public void getConnect() throws IOException {
zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
@Override
public void process(WatchedEvent event) {
// 再次啟動監聽
try {
getServerList();
} catch (Exception e) {
e.printStackTrace();
}
}
});
}
//
public void getServerList() throws Exception {
// 獲取服務器子節點信息，並且對父節點進行監聽
List<String> children = zk.getChildren(parentNode, true);
ArrayList<String> servers = new ArrayList<>();
for (String child : children) {
byte[] data = zk.getData(parentNode + "/" + child, false, null);
servers.add(new String(data));
}
// 把servers賦值給成員serverList，已提供給各業務線程使用
serversList = servers;
System.out.println(serversList);
}
// 業務功能
public void business() throws Exception {
System.out.println("client is working ...");
}
public static void main(String[] args) throws Exception {
// 獲取zk連接
DistributeClient client = new DistributeClient();
client.getConnect();
// 獲取servers的子節點信息，從中獲取服務器信息列表
client.getServerList();
// 業務進程啟動
client.business();
}
}