個(gè)人專題目錄

4.1 分布式安裝部署

1．集群規(guī)劃
在hadoop102、hadoop103和hadoop104三個(gè)節(jié)點(diǎn)上部署Zookeeper。
2．解壓安裝
（1）解壓Zookeeper安裝包到/opt/module/目錄下
$ tar -zxvf zookeeper-3.4.10.tar.gz -C /opt/module/
（2）同步/opt/module/zookeeper-3.4.10目錄內(nèi)容到hadoop103、hadoop104
$ xsync zookeeper-3.4.10/
3．配置服務(wù)器編號
（1）在/opt/module/zookeeper-3.4.10/這個(gè)目錄下創(chuàng)建zkData
$ mkdir -p zkData
（2）在/opt/module/zookeeper-3.4.10/zkData目錄下創(chuàng)建一個(gè)myid的文件
$ touch myid
添加myid文件，注意一定要在linux里面創(chuàng)建，在notepad++里面很可能亂碼
（3）編輯myid文件
$ vi myid
    在文件中添加與server對應(yīng)的編號：2
（4）拷貝配置好的zookeeper到其他機(jī)器上
$ xsync myid
并分別在hadoop102、hadoop103上修改myid文件中內(nèi)容為3、4
4．配置zoo.cfg文件
（1）重命名/opt/module/zookeeper-3.4.10/conf這個(gè)目錄下的zoo_sample.cfg為zoo.cfg
$ mv zoo_sample.cfg zoo.cfg
（2）打開zoo.cfg文件
$ vim zoo.cfg
修改數(shù)據(jù)存儲路徑配置
dataDir=/opt/module/zookeeper-3.4.10/zkData
增加如下配置
#######################cluster##########################
server.2=hadoop102:2888:3888
server.3=hadoop103:2888:3888
server.4=hadoop104:2888:3888
（3）同步zoo.cfg配置文件
$ xsync zoo.cfg
（4）配置參數(shù)解讀
server.A=B:C:D。
A是一個(gè)數(shù)字，表示這個(gè)是第幾號服務(wù)器；
集群模式下配置一個(gè)文件myid，這個(gè)文件在dataDir目錄下，這個(gè)文件里面有一個(gè)數(shù)據(jù)就是A的值，Zookeeper啟動時(shí)讀取此文件，拿到里面的數(shù)據(jù)與zoo.cfg里面的配置信息比較從而判斷到底是哪個(gè)server。
B是這個(gè)服務(wù)器的ip地址；
C是這個(gè)服務(wù)器與集群中的Leader服務(wù)器交換信息的端口；
D是萬一集群中的Leader服務(wù)器掛了，需要一個(gè)端口來重新進(jìn)行選舉，選出一個(gè)新的Leader，而這個(gè)端口就是用來執(zhí)行選舉時(shí)服務(wù)器相互通信的端口。
4．集群操作
（1）分別啟動Zookeeper
$ bin/zkServer.sh start
（2）查看狀態(tài)
bin/zkServer.sh status
JMX enabled by default
Using config: /opt/module/zookeeper-3.4.10/bin/../conf/zoo.cfg
Mode: follower
bin/zkServer.sh status
JMX enabled by default
Using config: /opt/module/zookeeper-3.4.10/bin/../conf/zoo.cfg
Mode: leader
bin/zkServer.sh status
JMX enabled by default
Using config: /opt/module/zookeeper-3.4.10/bin/../conf/zoo.cfg
Mode: follower

4.2 客戶端命令行操作

Zookeeper表現(xiàn)為一個(gè)分層的文件系統(tǒng)目錄樹結(jié)構(gòu)
不同于文件系統(tǒng)之處在于：zk節(jié)點(diǎn)可以有自己的數(shù)據(jù)，而unix文件系統(tǒng)中的目錄節(jié)點(diǎn)只有子節(jié)點(diǎn)

一個(gè)節(jié)點(diǎn)對應(yīng)一個(gè)應(yīng)用/服務(wù)，節(jié)點(diǎn)存儲的數(shù)據(jù)就是應(yīng)用需要的配置信息。

1．啟動客戶端
$ bin/zkCli.sh
2．顯示所有操作命令
[zk: localhost:2181(CONNECTED) 1] help
3．查看當(dāng)前znode中所包含的內(nèi)容
[zk: localhost:2181(CONNECTED) 0] ls /
[zookeeper]
4．查看當(dāng)前節(jié)點(diǎn)詳細(xì)數(shù)據(jù)
[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．分別創(chuàng)建2個(gè)普通節(jié)點(diǎn)
[zk: localhost:2181(CONNECTED) 3] create /sanguo "jinlian"
Created /sanguo
[zk: localhost:2181(CONNECTED) 4] create /sanguo/shuguo "liubei"
Created /sanguo/shuguo
6．獲得節(jié)點(diǎn)的值
[zk: localhost:2181(CONNECTED) 5] get /sanguo
jinlian
cZxid = 0x100000003
ctime = Wed Aug 29 00:03:23 CST 2018
mZxid = 0x100000003
mtime = Wed Aug 29 00:03:23 CST 2018
pZxid = 0x100000004
cversion = 1
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 7
numChildren = 1
[zk: localhost:2181(CONNECTED) 6]
[zk: localhost:2181(CONNECTED) 6] get /sanguo/shuguo
liubei
cZxid = 0x100000004
ctime = Wed Aug 29 00:04:35 CST 2018
mZxid = 0x100000004
mtime = Wed Aug 29 00:04:35 CST 2018
pZxid = 0x100000004
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 6
numChildren = 0
7．創(chuàng)建短暫節(jié)點(diǎn)
[zk: localhost:2181(CONNECTED) 7] create -e /sanguo/wuguo "zhouyu"
Created /sanguo/wuguo
（1）在當(dāng)前客戶端是能查看到的
[zk: localhost:2181(CONNECTED) 3] ls /sanguo 
[wuguo, shuguo]
（2）退出當(dāng)前客戶端然后再重啟客戶端
[zk: localhost:2181(CONNECTED) 12] quit
$ bin/zkCli.sh
（3）再次查看根目錄下短暫節(jié)點(diǎn)已經(jīng)刪除
[zk: localhost:2181(CONNECTED) 0] ls /sanguo
[shuguo]
8．創(chuàng)建帶序號的節(jié)點(diǎn)
    （1）先創(chuàng)建一個(gè)普通的根節(jié)點(diǎn)/sanguo/weiguo
[zk: localhost:2181(CONNECTED) 1] create /sanguo/weiguo "caocao"
Created /sanguo/weiguo
    （2）創(chuàng)建帶序號的節(jié)點(diǎn)
[zk: localhost:2181(CONNECTED) 2] create -s /sanguo/weiguo/xiaoqiao "jinlian"
Created /sanguo/weiguo/xiaoqiao0000000000
[zk: localhost:2181(CONNECTED) 3] create -s /sanguo/weiguo/daqiao "jinlian"
Created /sanguo/weiguo/daqiao0000000001
[zk: localhost:2181(CONNECTED) 4] create -s /sanguo/weiguo/diaocan "jinlian"
Created /sanguo/weiguo/diaocan0000000002
如果原來沒有序號節(jié)點(diǎn)，序號從0開始依次遞增。如果原節(jié)點(diǎn)下已有2個(gè)節(jié)點(diǎn)，則再排序時(shí)從2開始，以此類推。
9．修改節(jié)點(diǎn)數(shù)據(jù)值
[zk: localhost:2181(CONNECTED) 6] set /sanguo/weiguo "simayi"
10．節(jié)點(diǎn)的值變化監(jiān)聽
    （1）在hadoop104主機(jī)上注冊監(jiān)聽/sanguo節(jié)點(diǎn)數(shù)據(jù)變化
[zk: localhost:2181(CONNECTED) 26] [zk: localhost:2181(CONNECTED) 8] get /sanguo watch
    （2）在hadoop103主機(jī)上修改/sanguo節(jié)點(diǎn)的數(shù)據(jù)
[zk: localhost:2181(CONNECTED) 1] set /sanguo "xisi"
    （3）觀察hadoop104主機(jī)收到數(shù)據(jù)變化的監(jiān)聽
WATCHER::
WatchedEvent state:SyncConnected type:NodeDataChanged path:/sanguo
11．節(jié)點(diǎn)的子節(jié)點(diǎn)變化監(jiān)聽（路徑變化）
    （1）在hadoop104主機(jī)上注冊監(jiān)聽/sanguo節(jié)點(diǎn)的子節(jié)點(diǎn)變化
[zk: localhost:2181(CONNECTED) 1] ls /sanguo watch
[aa0000000001, server101]
    （2）在hadoop103主機(jī)/sanguo節(jié)點(diǎn)上創(chuàng)建子節(jié)點(diǎn)
[zk: localhost:2181(CONNECTED) 2] create /sanguo/jin "simayi"
Created /sanguo/jin
    （3）觀察hadoop104主機(jī)收到子節(jié)點(diǎn)變化的監(jiān)聽
WATCHER::
WatchedEvent state:SyncConnected type:NodeChildrenChanged path:/sanguo
12．刪除節(jié)點(diǎn)
[zk: localhost:2181(CONNECTED) 4] delete /sanguo/jin
13．遞歸刪除節(jié)點(diǎn)
[zk: localhost:2181(CONNECTED) 15] rmr /sanguo/shuguo
14．查看節(jié)點(diǎn)狀態(tài)
[zk: localhost:2181(CONNECTED) 17] stat /sanguo
cZxid = 0x100000003
ctime = Wed Aug 29 00:03:23 CST 2018
mZxid = 0x100000011
mtime = Wed Aug 29 00:21:23 CST 2018
pZxid = 0x100000014
cversion = 9
dataVersion = 1
aclVersion = 0
ephemeralOwner = 0x0
dataLength = 4
numChildren = 1

zookeeper支持某些特定的四字命令，他們大多是用來查詢ZK服務(wù)的當(dāng)前狀態(tài)及相關(guān)信息的，
通過telnet或nc向zookeeper提交相應(yīng)命令，如：echo ruok | nc 127.0.0.1 2181

運(yùn)行公式：echo 四字命令 | nc 主機(jī)IP zookeeper端口

ruok：測試服務(wù)是否處于正確狀態(tài)。如果確實(shí)如此，那么服務(wù)返回“imok ”，否則不做任何相應(yīng)
stat：輸出關(guān)于性能和連接的客戶端的列表
conf：輸出相關(guān)服務(wù)配置的詳細(xì)信息
cons：列出所有連接到服務(wù)器的客戶端的完全的連接 /會話的詳細(xì)信息。包括“接受 / 發(fā)送”的包數(shù)量、會話id 、操作延遲、最后的操作執(zhí)行等等信息
dump：列出未經(jīng)處理的會話和臨時(shí)節(jié)點(diǎn)
envi：輸出關(guān)于服務(wù)環(huán)境的詳細(xì)信息（區(qū)別于conf命令）
reqs：列出未經(jīng)處理的請求
wchs：列出服務(wù)器watch的詳細(xì)信息
wchc：通過session列出服務(wù)器watch的詳細(xì)信息，它的輸出是一個(gè)與watch相關(guān)的會話的列表
wchp：通過路徑列出服務(wù)器 watch的詳細(xì)信息。它輸出一個(gè)與 session相關(guān)的路徑

4.3 API應(yīng)用

4.3.1 環(huán)境搭建

1．創(chuàng)建一個(gè)Maven工程
2．添加pom文件
<dependencies>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
            <version>2.8.2</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/org.apache.zookeeper/zookeeper -->
        <dependency>
            <groupId>org.apache.zookeeper</groupId>
            <artifactId>zookeeper</artifactId>
            <version>3.4.10</version>
        </dependency>
</dependencies>
3．拷貝log4j.properties文件到項(xiàng)目根目錄
需要在項(xiàng)目的src/main/resources目錄下，新建一個(gè)文件，命名為“l(fā)og4j.properties”，在文件中填入。
log4j.rootLogger=INFO, stdout  
log4j.appender.stdout=org.apache.log4j.ConsoleAppender  
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout  
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n  
log4j.appender.logfile=org.apache.log4j.FileAppender  
log4j.appender.logfile.File=target/spring.log  
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout  
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n  

4.3.2 創(chuàng)建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) {

                // 收到事件通知后的回調(diào)函數(shù)（用戶的業(yè)務(wù)邏輯）
                System.out.println(event.getType() + "--" + event.getPath());

                // 再次啟動監(jiān)聽
                try {
                    zkClient.getChildren("/", true);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }

4.3.3 創(chuàng)建子節(jié)點(diǎn)

// 創(chuàng)建子節(jié)點(diǎn)
@Test
public void create() throws Exception {
        // 參數(shù)1：要?jiǎng)?chuàng)建的節(jié)點(diǎn)的路徑； 參數(shù)2：節(jié)點(diǎn)數(shù)據(jù) ； 參數(shù)3：節(jié)點(diǎn)權(quán)限 ；參數(shù)4：節(jié)點(diǎn)的類型
        String nodeCreated = zkClient.create("/test", "jinlian".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}

4.3.4 獲取子節(jié)點(diǎn)并監(jiān)聽節(jié)點(diǎn)變化

// 獲取子節(jié)點(diǎn)
@Test
public void getChildren() throws Exception {

     List<String> children = zkClient.getChildren("/", true);

     for (String child : children) {
         System.out.println(child);
     }

     // 延時(shí)阻塞
     Thread.sleep(Long.MAX_VALUE);
}

4.3.5 判斷Znode是否存在

// 判斷znode是否存在
@Test
public void exist() throws Exception {

  Stat stat = zkClient.exists("/eclipse", false);

  System.out.println(stat == null ? "not exist" : "exist");
}

4.4 監(jiān)聽服務(wù)器節(jié)點(diǎn)動態(tài)上下線案例

具體實(shí)現(xiàn)
（0）先在集群上創(chuàng)建/servers節(jié)點(diǎn)
[zk: localhost:2181(CONNECTED) 10] create /servers "servers"
Created /servers
（1）服務(wù)器端向Zookeeper注冊代碼
package com.test.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";
    
    // 創(chuàng)建到zk的客戶端連接
    public void getConnect() throws IOException{
        
        zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {

            @Override
            public void process(WatchedEvent event) {

            }
        });
    }
    
    // 注冊服務(wù)器
    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);
    }
    
    // 業(yè)務(wù)功能
    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 {
        
// 1獲取zk連接
        DistributeServer server = new DistributeServer();
        server.getConnect();
        
        // 2 利用zk連接注冊服務(wù)器信息
        server.registServer(args[0]);
        
        // 3 啟動業(yè)務(wù)功能
        server.business(args[0]);
    }
}
（2）客戶端代碼
package com.test.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";

    // 創(chuàng)建到zk的客戶端連接
    public void getConnect() throws IOException {
        zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {

            @Override
            public void process(WatchedEvent event) {

                // 再次啟動監(jiān)聽
                try {
                    getServerList();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }

    // 獲取服務(wù)器列表信息
    public void getServerList() throws Exception {
        
        // 1獲取服務(wù)器子節(jié)點(diǎn)信息，并且對父節(jié)點(diǎn)進(jìn)行監(jiān)聽
        List<String> children = zk.getChildren(parentNode, true);

        // 2存儲服務(wù)器信息列表
        ArrayList<String> servers = new ArrayList<>();
        
        // 3遍歷所有節(jié)點(diǎn)，獲取節(jié)點(diǎn)中的主機(jī)名稱信息
        for (String child : children) {
            byte[] data = zk.getData(parentNode + "/" + child, false, null);

            servers.add(new String(data));
        }

        // 4打印服務(wù)器列表信息
        System.out.println(servers);
    }

    // 業(yè)務(wù)功能
    public void business() throws Exception{

        System.out.println("client is working ...");
Thread.sleep(Long.MAX_VALUE);
    }

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

        // 1獲取zk連接
        DistributeClient client = new DistributeClient();
        client.getConnect();

        // 2獲取servers的子節(jié)點(diǎn)信息，從中獲取服務(wù)器信息列表
        client.getServerList();

        // 3業(yè)務(wù)進(jìn)程啟動
        client.business();
    }
}