环境搭建

1）创建一个Maven工程
2）添加pom文件

  1. <dependencies>
  2.        <dependency>
  3.            <groupId>junit</groupId>
  4.            <artifactId>junit</artifactId>
  5.            <version>RELEASE</version>
  6.        </dependency>
  7.        <dependency>
  8.            <groupId>org.apache.logging.log4j</groupId>
  9.            <artifactId>log4j-core</artifactId>
  10.            <version>2.8.2</version>
  11.        </dependency>
  12.        <!-- https://mvnrepository.com/artifact/org.apache.zookeeper/zookeeper -->
  13.        <dependency>
  14.            <groupId>org.apache.zookeeper</groupId>
  15.            <artifactId>zookeeper</artifactId>
  16.            <version>3.5.7</version>
  17.        </dependency>
  18. </dependencies>

3）拷贝log4j.properties文件到项目根目录
需要在项目的src/main/resources目录下，新建一个文件，命名为“log4j.properties”，在文件中填入。

  1. log4j.rootLogger=INFO, stdout  
  2. log4j.appender.stdout=org.apache.log4j.ConsoleAppender  
  3. log4j.appender.stdout.layout=org.apache.log4j.PatternLayout  
  4. log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
  5. log4j.appender.logfile=org.apache.log4j.FileAppender  
  6. log4j.appender.logfile.File=target/spring.log  
  7. log4j.appender.logfile.layout=org.apache.log4j.PatternLayout  
  8. log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

在API中使用指令

  1. import org.apache.zookeeper.*;
  2. import org.apache.zookeeper.data.Stat;
  3. import org.junit.After;
  4. import org.junit.Before;
  5. import org.junit.Test;
  7. import java.io.IOException;
  8. import java.util.List;
  10. public class ZKDemo {
  11.     private ZooKeeper zk;
  12.     /*
  13.         通过代码操作Zookeeper
  14.         1.创建Zookeeper的客户端
  15.         2.具体API操作
  16.         3.关闭资源
  17.      */
  18.     @Before
  19.     public void before() throws IOException {
  20.         //1.创建Zookeeper的客户端
  21.         //zookeeper集群中节点的地址
  22.         String hostName = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
  23.         //session超时时间
  24.         int sessionTimeout = 4000;
  25.         //Watcher对象是用来接收服务器端的响应事件（总响应事件--一般不用）
  26.         zk = new ZooKeeper(hostName, sessionTimeout, new Watcher() {
  27.             public void process(WatchedEvent event) {
  29.             }
  30.         });
  31.     }
  32.     @After
  33.     public void after(){
  34.         // 3.关闭资源
  35.         if (zk != null){
  36.             try {
  37.                 zk.close();
  38.             } catch (InterruptedException e) {
  39.                 e.printStackTrace();
  40.             }
  41.         }
  42.     }
  44.     //创建子节点
  45.     @Test
  46.     public void test() throws KeeperException, InterruptedException {
  47.         /*
  48.         create(final String path, byte data[], List<ACL> acl,
  49.             CreateMode createMode)
  50.          path : 目标节点的路径
  51.          data ： 节点存放的数据
  52.          acl : 节点的权限
  53.          createMode ： 节点类型（持久有序列号，持久无序列号，临时有序列号，临时无序列号）
  54.          */
  55.         zk.create("/sanguo", "caocaobushiren".getBytes(),
  56.                 ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
  57.     }
  59.     //获取子节点并监听节点变化
  60.     @Test
  61.     public void test2() throws KeeperException, InterruptedException {
  63.         try {
  64.             listener();
  65.         } catch (Exception e) {
  66.             e.printStackTrace();
  67.         }
  68.         //因为主线程一旦死掉，那么用来监听响应事件的线程也会死掉。
  69.         Thread.sleep(Long.MAX_VALUE);
  70.     }
  72.     public void listener() throws Exception {
  73.        /*
  74.         getChildren(final String path, Watcher watcher)
  75.         path : 目标节点的路径
  76.         watcher ： 监听器对象--接收响应事件
  77.          */
  78.         List<String> children = zk.getChildren("/xiyouji", new Watcher() {
  79.             //process方法是真正用来接收响应事件的方法
  80.             public void process(WatchedEvent event) {
  81.                 //1.处理响应事件
  82.                 System.out.println("节点发生了改变");
  83.                 //2.再次注册监听
  84.                 try {
  85.                     listener();
  86.                 } catch (Exception e) {
  87.                     e.printStackTrace();
  88.                 }
  89.             }
  90.         });
  92.         for (String child : children) {
  93.             System.out.println(child);
  94.         }
  95.     }
  97.     //判断节点是否存在
  98.     @Test
  99.     public void test3() throws KeeperException, InterruptedException {
  100.         /*
  101.             exists(String path, boolean watch)
  102.             path:目标节点的路径
  103.             watch : 是否使用总监听器对象
  105.             如果节点存在则返回Stat类型的对象，如果节点不存在则返回null
  106.          */
  107.         Stat exists = zk.exists("/sanguo2", false);
  108.         System.out.println((exists==null)?"false":"true");
  110.     }
  111. }

在API实现服务器动态上下线。

说明：
创建的ZKClient是ZKServer的客户端
ZKClient和ZKServer都是ZooKeeper的客户端

代码：
1）创建服务器，复制3台服务器副本，设置不同的参数

  2. import org.apache.zookeeper.*;
  3. import org.apache.zookeeper.data.Stat;
  5. import java.io.IOException;
  7. /*
  8.     具体步骤：
  9.     1.创建一个临时节点作为Zookeeper客户端
  10.     2.先判断父节点是否存在，如果不存在则创建
  11.     3.创建临时节点
  12.  */
  13. public class ZKServer {
  15.     public static void main(String[] args){
  16.         ZooKeeper zk = null;
  17.         try {
  18.             //1.创建Zookeeper客户端
  19.             String hostName = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
  20.             int sessionTimeout = 4000;
  21.             zk = new ZooKeeper(hostName, sessionTimeout, new Watcher() {
  22.                 public void process(WatchedEvent event) {
  24.                 }
  25.             });
  26.             //2.判断父节点是否存在
  27.             Stat exists = zk.exists("/server", false);
  28.             if (exists == null) {
  29.                 //父节点不存在，则创建--永久节点（父节点必须一直存在，client会监听此节点）
  30.                 zk.create("/server", "".getBytes(),
  31.                         ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
  32.             }
  34.             //3.创建临时节点---不同的服务器临时节点是不一样的所以不能写死。
  35.             //因为服务器断开节点就得消失所以节点必须是临时节点
  36.             zk.create("/server/" + args[0], args[1].getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
  37.                     CreateMode.EPHEMERAL);
  39.             Thread.sleep(Long.MAX_VALUE);//程序不能结束。
  40.         }catch (Exception e){
  41.             e.printStackTrace();
  42.         }finally {
  43.             //关闭资源
  44.             try {
  45.                 zk.close();
  46.             } catch (InterruptedException e) {
  47.                 e.printStackTrace();
  48.             }
  49.         }
  51.     }
  52. }

2）创建客户端

  2. import org.apache.zookeeper.*;
  3. import org.apache.zookeeper.data.Stat;
  5. import java.io.IOException;
  6. import java.util.List;
  8. /*
  9.     客户端具体步骤：
  10.     1.创建Zookeeper的客户端
  11.     2.判断父节点是否存在（因为是先启动服务器-服务器中已经判断过了所以该操作不用做了）
  12.         如果允许先启动客户端那么必须做该操作。
  13.     3.获取/server节点下的子节点的内容并监听此节点
  14.  */
  15. public class ZKClient {
  17.     private static ZooKeeper zk;
  19.     public static void main(String[] args){
  20.         try {
  21.             //1.创建Zookeeper客户端
  22.             String hostName = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
  23.             int sessionTimeout = 4000;
  24.             zk = new ZooKeeper(hostName, sessionTimeout, new Watcher() {
  25.                 public void process(WatchedEvent event) {
  27.                 }
  28.             });
  29.             //2.判断父节点是否存在（允许先起动客户端）
  30.             Stat exists = zk.exists("/server", false);
  31.             if (exists == null) {
  32.                 //父节点不存在，则创建--永久节点（父节点必须一直存在，client会监听此节点）
  33.                 zk.create("/server", "".getBytes(),
  34.                         ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
  35.             }
  36.             //3.获取子节点并监听（一直可以监听-递归）
  37.             listener();
  38.             //4.程序需要一直运行
  39.             Thread.sleep(Long.MAX_VALUE);
  40.         }catch (Exception e){
  41.             e.printStackTrace();
  42.         }finally {
  43.             if (zk != null){
  44.                 try {
  45.                     zk.close();
  46.                 } catch (InterruptedException e) {
  47.                     e.printStackTrace();
  48.                 }
  49.             }
  50.         }
  51.     }
  53.     public static void listener() throws Exception {
  54.         List<String> children = zk.getChildren("/server", new Watcher() {
  55.             //一旦/server节点的子节点发生变化那么就会触发该方法
  56.             public void process(WatchedEvent event) {
  57.                 //1.一旦子节点发生变化那么就重新获取该节点中的内容（那么就知道现在在线的有哪台服务器）
  58.                 //2.重新注册监听
  59.                 try {
  60.                     listener();
  61.                 } catch (Exception e) {
  62.                     e.printStackTrace();
  63.                 }
  64.             }
  65.         });
  67.         for (String child : children) {
  68.             System.out.println(child);
  69.         }
  71.         System.out.println("=====================================================");
  72.     }
  73. }