Zookeeper

zookeeper场景使用

##服务器注册与状态监控
zookeeper有一种类型为EPHEMERAL的节点，这种节点的特点是当创建它的server（或者应用）终止的时候，它会自动删除，利用这个特性，我们可以利用zookeeper的getChildren方法，监控它的父节点的子节点列表的变化，进而做进一步处理。

服务器注册的样例代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
package com.louz.zookeeper;
import java.io.IOException;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooKeeper;
public class ServerRegister {
    /**
     * @param args 该数组元素个数必须为1，用于表明注册的服务名称
     * @throws InterruptedException 
     * @throws IOException 
     * @throws KeeperException 
     */
    public static void main(String[] args) throws InterruptedException, IOException, KeeperException {
        if (args == null || args.length != 1) {
            System.err.println("It must has one and only one arg");
            System.exit(1);
        }
        
        ServerRegister s = new ServerRegister();
        s.connectZookeeper(args[0]);
        
        Thread.sleep(10000);
    }
    private void connectZookeeper(String serverName) throws IOException, KeeperException, InterruptedException {
        ZooKeeper zk = new ZooKeeper("hadoop2:2181", 5000, null);
        
        // /servers节点必须已经在zookeeper上存在
        String createdPath = zk.create("/servers/" + serverName, serverName.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
        
        System.out.println("Created Path: " + createdPath);
    }
}

监听程序的代码，需要注意的是，由于zookeeper的监听都是只监听一次，所以在监控到变化之后，需要重新再注册一次监听器：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
package com.louz.zookeeper;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
public class ServerListener2 {
    private static ZooKeeper zk;
    private final static String monitorNode = "/servers";
    
    /**
     * @param args
     * @throws IOException 
     * @throws InterruptedException 
     */
    public static void main(String[] args) throws IOException, InterruptedException {
        zk = new ZooKeeper("hadoop2:2181", 5000, null);
        
        updateServerList();      // 触发第一次监听
        Thread.sleep(30000);     // 延时30秒再结束进程，以便演示应用注册时，该监听进程的变化
    }
    protected static void updateServerList() {
        List<String> serverList = new ArrayList<String>();
        
        try {
            
            List<String> children = zk.getChildren(monitorNode, new Watcher(){
                @Override
                public void process(WatchedEvent event) {
                    System.out.println(event.getPath());
                    updateServerList();     // 由于zookeeper的监听都是只监听一次，所以在监控到变化之后，需要重新再注册一次监听器
                }
                
            });
            for (String subName : children) {
                byte[] data = zk.getData(monitorNode + "/" + subName, false, new Stat());
                serverList.add(new String(data));
            }
            
            System.out.println("Server list changed: " + serverList);
        } catch (KeeperException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        
    }
}

演示：

  1. 先将ServerRegister打包成jar包，可通过Eclipse的Export->Runnable JAR file导出，我本地的名字为zookeeper-server-register.jar，留待后面使用

  2. 运行ServerListener2的main函数，显示如下：

    1
    Server list changed: []

  3. 进入到zookeeper-server-register.jar所在目录，在命令行运行

    1
    2
    3
    4
    5
    6
    E:\tmp\zookeeper>java -jar zookeeper-server-register.jar server1
    SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
    SLF4J: Defaulting to no-operation (NOP) logger implementation
    SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further detail
    s.
    Created Path: /servers/server1

同时在ServerListener2的输出窗口可以看到:

1
2
3
Server list changed: []
/servers
Server list changed: [server1]

再新打开一个命令行窗口，运行

1
2
3
4
5
6
E:\tmp\zookeeper>java -jar zookeeper-server-register.jar server2
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further detail
s.
Created Path: /servers/server2

同时在ServerListener2的输出窗口可以看到:

1
2
3
4
5
Server list changed: []
/servers
Server list changed: [server1]
/servers
Server list changed: [server1, server2]

当ServerRegister的进程结束后，可以看到ServerListener2的输出窗口变化：

1
2
3
4
5
6
7
8
9
Server list changed: []
/servers
Server list changed: [server1]
/servers
Server list changed: [server1, server2]
/servers
Server list changed: [server2]
/servers
Server list changed: []

说明ServerListener2可以监听到/servers的子节点的变化