您的位置: 首页  >  文章  >  ZooKeeper之（六）应用实例

ZooKeeper之（六）应用实例

分类: 文章 • 2025-05-04 20:29:16

6.1 Java API

客户端要连接 Zookeeper服务器可以通过创建 org.apache.zookeeper.ZooKeeper 的一个实例对象，然后调用这个类提供的接口来和服务器交互。

ZooKeeper 主要是用来维护和监控一个目录节点树中存储的数据的状态，所有我们能够操作 ZooKeeper 和操作目录节点树大体一样，如创建一个目录节点，给某个目录节点设置数据，获取某个目录节点的所有子目录节点，给某个目录节点设置权限和监控这个目录节点的状态变化。

下面通过代码实例，来熟悉一下JavaAPI的常用方法。

[java] view plain copy
 print?
  1. import java.util.List;  
  2.   
  3. import org.apache.zookeeper.CreateMode;  
  4. import org.apache.zookeeper.WatchedEvent;  
  5. import org.apache.zookeeper.Watcher;  
  6. import org.apache.zookeeper.ZooDefs.Ids;  
  7. import org.apache.zookeeper.ZooKeeper;  
  8. import org.apache.zookeeper.data.Stat;  
  9.   
  10. public class ZkTest {  
  11.   
  12.     private static final String CONNECT_STRING = "127.0.0.1:2181";  
  13.     private static final int SESSION_TIMEOUT = 3000;  
  14.   
  15.     public static void main(String[] args) throws Exception {  
  16.   
  17.         // 定义一个监控所有节点变化的Watcher  
  18.         Watcher allChangeWatcher = new Watcher() {  
  19.             @Override  
  20.             public void process(WatchedEvent event) {  
  21.                 System.out.println("**watcher receive WatchedEvent** changed path: " + event.getPath()  
  22.                         + "; changed type: " + event.getType().name());  
  23.             }  
  24.         };  
  25.   
  26.         // 初始化一个与ZK连接。三个参数：  
  27.         // 1、要连接的服务器地址，"IP:port"格式；  
  28.         // 2、会话超时时间  
  29.         // 3、节点变化监视器  
  30.         ZooKeeper zk = new ZooKeeper(CONNECT_STRING, SESSION_TIMEOUT, allChangeWatcher);  
  31.   
  32.         // 新建节点。四个参数：1、节点路径；2、节点数据；3、节点权限；4、创建模式  
  33.         zk.create("/myName", "chenlongfei".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);  
  34.         System.out.println("create new node '/myName'");  
  35.   
  36.         // 判断某路径是否存在。两个参数：1、节点路径；2、是否监控（Watcher即初始化ZooKeeper时传入的Watcher）  
  37.         Stat beforSstat = zk.exists("/myName", true);  
  38.         System.out.println("Stat of '/myName' before change : " + beforSstat.toString());  
  39.   
  40.         // 修改节点数据。三个参数：1、节点路径；2、新数据；3、版本，如果为-1，则匹配任何版本  
  41.         Stat afterStat = zk.setData("/myName", "clf".getBytes(), -1);  
  42.         System.out.println("Stat of '/myName' after change: " + afterStat.toString());  
  43.   
  44.         // 获取所有子节点。两个参数：1、节点路径；2、是否监控该节点  
  45.         List<String> children = zk.getChildren("/", true);  
  46.         System.out.println("children of path '/': " + children.toString());  
  47.   
  48.         // 获取节点数据。三个参数：1、节点路径；2、书否监控该节点；3、版本等信息可以通过一个Stat对象来指定  
  49.         byte[] nameByte = zk.getData("/myName", true, null);  
  50.         String name = new String(nameByte, "UTF-8");  
  51.         System.out.println("get data from '/myName': " + name);  
  52.   
  53.         // 删除节点。两个参数：1、节点路径；2、 版本，-1可以匹配任何版本，会删除所有数据  
  54.         zk.delete("/myName", -1);  
  55.         System.out.println("delete '/myName'");  
  56.   
  57.         zk.close();  
  58.     }  

运行程序，打印结果如下：

ZooKeeper之（六）应用实例

更详细的API请参考官方网站。

 

Zookeeper 从设计模式角度来看，是一个基于观察者模式设计的分布式服务管理框架，它负责存储和管理大家都关心的数据，然后接受观察者的注册，一旦这些数据的状态发生变化，Zookeeper 将负责通知已经在 Zookeeper 上注册的那些观察者做出相应的反应。

下面通过两个ZooKeeper的典型用用场景来体会下ZooKeeper的特性与使用方法。


6.2 分布式锁

先来回顾一下多线程中的锁控制。

[java] view plain copy
 print?
  1. public class MultiThreadTest {  
  2.   
  3.     // 以一个静态变量来模拟公共资源  
  4.     private static int counter = 0;  
  5.   
  6.     // 多线程环境下，会出现并发问题  
  7.     public static void plus() {  
  8.           
  9.         // 计数器加一  
  10.         counter++;  
  11.           
  12.         // 线程随机休眠数毫秒，模拟现实中的耗时操作  
  13.         int sleepMillis = (int) (Math.random() * 100);  
  14.         try {  
  15.             Thread.sleep(sleepMillis);  
  16.         } catch (InterruptedException e) {  
  17.             e.printStackTrace();  
  18.         }  
  19.     }  
  20.   
  21.     // 线程实现类  
  22.     static class CountPlus extends Thread {  
  23.         @Override  
  24.         public void run() {  
  25.             for (int i = 0; i < 20; i++) {  
  26.                 plus();  
  27.             }  
  28.             System.out.println(Thread.currentThread().getName() + "执行完毕：" + counter);  
  29.         }  
  30.   
  31.         public CountPlus(String threadName) {  
  32.             super(threadName);  
  33.         }  
  34.   
  35.     }  
  36.   
  37.     public static void main(String[] args) throws Exception {  
  38.   
  39.         // 开启五个线程  
  40.         CountPlus threadA = new CountPlus("threadA");  
  41.         threadA.start();  
  42.   
  43.         CountPlus threadB = new CountPlus("threadB");  
  44.         threadB.start();  
  45.   
  46.         CountPlus threadC = new CountPlus("threadC");  
  47.         threadC.start();  
  48.   
  49.         CountPlus threadD = new CountPlus("threadD");  
  50.         threadD.start();  
  51.   
  52.         CountPlus threadE = new CountPlus("threadE");  
  53.         threadE.start();  
  54.     }  
  55. }  

 上例中，开启了五个线程，每个线程通过plus()方法对静态变量counter分别进行20次累加，预期counter最后会变成100。运行程序：

ZooKeeper之（六）应用实例

可以发现，五个线程执行完毕之后，counter并没有变成100。plus()方法涉及到对公共资源的改动，但是并没有对它进行同步控制，可能会造成多个线程同时对公共资源发起改动，进而出现并发问题。问题的根源在于，上例中没有保证同一时刻只能有一个线程可以改动公共资源。

给plus()方法加上synchronized关键字，重新运行程序：

ZooKeeper之（六）应用实例

可见，最终达到了预期结果。

synchronized关键字的作用是对plus()方法加入锁控制，一个线程想要执行该方法，首先需要获得锁（锁是唯一的），执行完毕后，再释放锁。如果得不到锁，该线程会进入等待池中等待，直到抢到锁才能继续执行。这样就保证了同一时刻只能有一个线程可以改动公共资源，避免了并发问题。

 

共享锁在同一个进程中很容易实现，可以靠Java本身提供的同步机制解决，但是在跨进程或者在不同 Server 之间就不好实现了，这时候就需要一个中间人来协调多个Server之间的各种问题，比如如何获得锁/释放锁、谁先获得锁、谁后获得锁等。

借助Zookeeper 可以实现这种分布式锁：需要获得锁的 Server 创建一个 EPHEMERAL_SEQUENTIAL 目录节点，然后调用 getChildren()方法获取列表中最小的目录节点，如果最小节点就是自己创建的目录节点，那么它就获得了这个锁，如果不是那么它就调用 exists() 方法并监控前一节点的变化，一直到自己创建的节点成为列表中最小编号的目录节点，从而获得锁。释放锁很简单，只要删除它自己所创建的目录节点就行了。

流程图如下：

ZooKeeper之（六）应用实例 

下面我们对刚才的代码进行改造，不用synchronize关键字而是使用ZooKeeper达到锁控制的目的，模拟分布式锁的实现。 

[java] view plain copy
 print?
  1. import java.util.Collections;  
  2. import java.util.List;  
  3.   
  4. import org.apache.zookeeper.CreateMode;  
  5. import org.apache.zookeeper.KeeperException;  
  6. import org.apache.zookeeper.WatchedEvent;  
  7. import org.apache.zookeeper.Watcher;  
  8. import org.apache.zookeeper.ZooDefs.Ids;  
  9. import org.apache.zookeeper.ZooKeeper;  
  10. import org.apache.zookeeper.data.Stat;  
  11.   
  12. public class ZkDistributedLock {  
  13.   
  14.     // 以一个静态变量来模拟公共资源  
  15.     private static int counter = 0;  
  16.   
  17.     public static void plus() {  
  18.   
  19.         // 计数器加一  
  20.         counter++;  
  21.   
  22.         // 线程随机休眠数毫秒，模拟现实中的费时操作  
  23.         int sleepMillis = (int) (Math.random() * 100);  
  24.         try {  
  25.             Thread.sleep(sleepMillis);  
  26.         } catch (InterruptedException e) {  
  27.             e.printStackTrace();  
  28.         }  
  29.     }  
  30.   
  31.     // 线程实现类  
  32.     static class CountPlus extends Thread {  
  33.   
  34.         private static final String LOCK_ROOT_PATH = "/Locks";  
  35.         private static final String LOCK_NODE_NAME = "Lock_";  
  36.   
  37.         // 每个线程持有一个zk客户端，负责获取锁与释放锁  
  38.         ZooKeeper zkClient;  
  39.   
  40.         @Override  
  41.         public void run() {  
  42.   
  43.             for (int i = 0; i < 20; i++) {  
  44.   
  45.                 // 访问计数器之前需要先获取锁  
  46.                 String path = getLock();  
  47.   
  48.                 // 执行任务  
  49.                 plus();  
  50.   
  51.                 // 执行完任务后释放锁  
  52.                 releaseLock(path);  
  53.             }  
  54.               
  55.             closeZkClient();  
  56.             System.out.println(Thread.currentThread().getName() + "执行完毕：" + counter);  
  57.         }  
  58.   
  59.         /** 
  60.          * 获取锁，即创建子节点，当该节点成为序号最小的节点时则获取锁 
  61.          */  
  62.         private String getLock() {  
  63.             try {  
  64.                 // 创建EPHEMERAL_SEQUENTIAL类型节点  
  65.                 String lockPath = zkClient.create(LOCK_ROOT_PATH + "/" + LOCK_NODE_NAME,  
  66.                         Thread.currentThread().getName().getBytes(), Ids.OPEN_ACL_UNSAFE,  
  67.                         CreateMode.EPHEMERAL_SEQUENTIAL);  
  68.                 System.out.println(Thread.currentThread().getName() + " create path : " + lockPath);  
  69.   
  70.                 // 尝试获取锁  
  71.                 tryLock(lockPath);  
  72.   
  73.                 return lockPath;  
  74.             } catch (Exception e) {  
  75.                 e.printStackTrace();  
  76.             }  
  77.             return null;  
  78.         }  
  79.   
  80.         /** 
  81.          * 该函数是一个递归函数 如果获得锁，直接返回；否则，阻塞线程，等待上一个节点释放锁的消息，然后重新tryLock 
  82.          */  
  83.         private boolean tryLock(String lockPath) throws KeeperException, InterruptedException {  
  84.   
  85.             // 获取LOCK_ROOT_PATH下所有的子节点，并按照节点序号排序  
  86.             List<String> lockPaths = zkClient.getChildren(LOCK_ROOT_PATH, false);  
  87.             Collections.sort(lockPaths);  
  88.   
  89.             int index = lockPaths.indexOf(lockPath.substring(LOCK_ROOT_PATH.length() + 1));  
  90.             if (index == 0) { // lockPath是序号最小的节点，则获取锁  
  91.                 System.out.println(Thread.currentThread().getName() + " get lock, lockPath: " + lockPath);  
  92.                 return true;  
  93.             } else { // lockPath不是序号最小的节点  
  94.   
  95.                 // 创建Watcher，监控lockPath的前一个节点  
  96.                 Watcher watcher = new Watcher() {  
  97.                     @Override  
  98.                     public void process(WatchedEvent event) {  
  99.                         System.out.println(event.getPath() + " has been deleted");  
  100.                         synchronized (this) {  
  101.                             notifyAll();  
  102.                         }  
  103.                     }  
  104.                 };  
  105.                 String preLockPath = lockPaths.get(index - 1);  
  106.                 Stat stat = zkClient.exists(LOCK_ROOT_PATH + "/" + preLockPath, watcher);  
  107.   
  108.                 if (stat == null) { // 由于某种原因，前一个节点不存在了（比如连接断开），重新tryLock  
  109.                     return tryLock(lockPath);  
  110.                 } else { // 阻塞当前进程，直到preLockPath释放锁，重新tryLock  
  111.                     System.out.println(Thread.currentThread().getName() + " wait for " + preLockPath);  
  112.                     synchronized (watcher) {  
  113.                         watcher.wait();  
  114.                     }  
  115.                     return tryLock(lockPath);  
  116.                 }  
  117.             }  
  118.   
  119.         }  
  120.   
  121.         /** 
  122.          * 释放锁，即删除lockPath节点 
  123.          */  
  124.         private void releaseLock(String lockPath) {  
  125.             try {  
  126.                 zkClient.delete(lockPath, -1);  
  127.             } catch (InterruptedException | KeeperException e) {  
  128.                 e.printStackTrace();  
  129.             }  
  130.         }  
  131.   
  132.         public void setZkClient(ZooKeeper zkClient) {  
  133.             this.zkClient = zkClient;  
  134.         }  
  135.           
  136.         public void closeZkClient(){  
  137.             try {  
  138.                 zkClient.close();  
  139.             } catch (InterruptedException e) {  
  140.                 e.printStackTrace();  
  141.             }  
  142.         }  
  143.   
  144.         public CountPlus(String threadName) {  
  145.             super(threadName);  
  146.         }  
  147.     }  
  148.   
  149.     public static void main(String[] args) throws Exception {  
  150.   
  151.         // 开启五个线程  
  152.         CountPlus threadA = new CountPlus("threadA");  
  153.         setZkClient(threadA);  
  154.         threadA.start();  
  155.   
  156.         CountPlus threadB = new CountPlus("threadB");  
  157.         setZkClient(threadB);  
  158.         threadB.start();  
  159.   
  160.         CountPlus threadC = new CountPlus("threadC");  
  161.         setZkClient(threadC);  
  162.         threadC.start();  
  163.   
  164.         CountPlus threadD = new CountPlus("threadD");  
  165.         setZkClient(threadD);  
  166.         threadD.start();  
  167.   
  168.         CountPlus threadE = new CountPlus("threadE");  
  169.         setZkClient(threadE);  
  170.         threadE.start();  
  171.     }  
  172.   
  173.     public static void setZkClient(CountPlus thread) throws Exception {  
  174.         ZooKeeper zkClient = new ZooKeeper("127.0.0.1:2181", 3000, null);  
  175.         thread.setZkClient(zkClient);  
  176.     }  
  177.   
  178. }  

注意：运行程序之前需要创建“/Locks”作为存放锁信息的根节点。

一旦某个Server想要获得锁，就会在/Locks”下创建一个EPHEMERAL_SEQUENTIAL类型的名为“Lock_”子节点，ZooKeeper会自动为每个子节点附加一个递增的编号，该编号为int类型，长度为10，左端以0补全。“/Locks”下会维持着这样一系列的节点：

Lock_0000000001,Lock_0000000002, Lock_0000000003, Lock_0000000004…

一旦这些创建这些节点的Server断开连接，该节点就会被清除（当然也可以主动清除）。

由于节点的编号是递增的，创建越晚排名越靠后。遵循先到先得的原则，Server创建完节点之后会检查自己的节点是不是最小的，如果是，那就获得锁，如果不是，排队等待。执行完任务之后，Server清除自己创建的节点，这样后面的节点会依次获得锁。

程序的运行结果如下：

ZooKeeper之（六）应用实例


6.3 分布式队列

很多单机上很平常的事情，放在集群环境中都会发生质的变化。

以一个常见的生产者-消费者模型举例：有一个容量有限的邮筒，寄信者（即生产者）不断地将信件塞入邮筒，邮递员（即消费者）不断地从邮筒取出信件发往目的地。运行期间需要保证：

（1）邮筒已达上限时，寄信者停止活动，等带邮筒恢复到非满状态

（2）邮筒已空时，邮递员停止活动，等带邮筒恢复到非空状态

该邮筒用有序队列实现，保证FIFO（先进先出）特性。

在一台机器上，可以用有序队列来实现邮筒，保证FIFO（先进先出）特性，开启两个线程，一个充当寄信者，一个充当邮递员，通过wait()/notify()很容易实现上述功能。

但是，如果在跨进程或者分布式环境下呢？比如，一台机器运行生产者程序，另一台机器运行消费者程序，代表邮筒的有序队列无法跨机器共享，但是两者需要随时了解邮筒的状态（是否已满、是否已空）以及保证信件的有序（先到达的先发送）。

这种情况下，可以借助ZooKeeper实现一个分布式队列。新建一个“/mailBox”节点代表邮筒。一旦有信件到达，就在该节点下创建PERSISTENT_SEQUENTIAL类型的子节点，当子节点总数达到上限时，阻塞生产者，然后使用getChildren(String path, Watcher watcher)方法监控子节点的变化，子节点总数减少后再回复生产；而消费者每次选取序号最小的子节点进行处理，然后删除该节点，当子节点总数为0时，阻塞消费者，同样设置监控，子节点总数增加后再回复消费。

代码如下：

[java] view plain copy
 print?
  1. import java.io.IOException;  
  2. import java.util.Collections;  
  3. import java.util.List;  
  4.   
  5. import org.apache.zookeeper.CreateMode;  
  6. import org.apache.zookeeper.KeeperException;  
  7. import org.apache.zookeeper.WatchedEvent;  
  8. import org.apache.zookeeper.Watcher;  
  9. import org.apache.zookeeper.ZooDefs.Ids;  
  10. import org.apache.zookeeper.ZooKeeper;  
  11. import org.apache.zookeeper.data.Stat;  
  12.   
  13. public class ZkDistributedQueue {  
  14.   
  15.     // 邮箱上限为10封信  
  16.     private static final int MAILBOX_MAX_SIZE = 10;  
  17.   
  18.     // 邮箱路径  
  19.     private static final String MAILBOX_ROOT_PATH = "/mailBox";  
  20.   
  21.     // 信件节点  
  22.     private static final String LETTER_NODE_NAME = "letter_";  
  23.   
  24.     // 生产者线程，负责接受信件  
  25.     static class Producer extends Thread {  
  26.   
  27.         ZooKeeper zkClient;  
  28.   
  29.         @Override  
  30.         public void run() {  
  31.             while (true) {  
  32.                 try {  
  33.                     if (getLetterNum() == MAILBOX_MAX_SIZE) { // 信箱已满  
  34.                         System.out.println("mailBox has been full");  
  35.                         // 创建Watcher，监控子节点的变化  
  36.                         Watcher watcher = new Watcher() {  
  37.                             @Override  
  38.                             public void process(WatchedEvent event) {  
  39.                                 // 生产者已停止，只有消费者在活动，所以只可能出现发送信件的动作  
  40.                                 System.out.println("mailBox has been not full");  
  41.                                 synchronized (this) {  
  42.                                     notify(); // 唤醒生产者  
  43.                                 }  
  44.                             }  
  45.                         };  
  46.                         zkClient.getChildren(MAILBOX_ROOT_PATH, watcher);  
  47.   
  48.                         synchronized (watcher) {  
  49.                             watcher.wait(); // 阻塞生产者  
  50.                         }  
  51.                     } else {  
  52.                         // 线程随机休眠数毫秒，模拟现实中的费时操作  
  53.                         int sleepMillis = (int) (Math.random() * 1000);  
  54.                         Thread.sleep(sleepMillis);  
  55.                           
  56.                         // 接收信件，创建新的子节点  
  57.                         String newLetterPath = zkClient.create(  
  58.                                 MAILBOX_ROOT_PATH + "/" + LETTER_NODE_NAME,  
  59.                                 "letter".getBytes(),   
  60.                                 Ids.OPEN_ACL_UNSAFE,   
  61.                                 CreateMode.PERSISTENT_SEQUENTIAL);  
  62.                         System.out.println("a new letter has been received: "   
  63.                                 + newLetterPath.substring(MAILBOX_ROOT_PATH.length()+1)   
  64.                                 + ", letter num: " + getLetterNum());  
  65.                     }  
  66.                 } catch (Exception e) {  
  67.                     System.out.println("producer equit task becouse of exception !");  
  68.                     e.printStackTrace();  
  69.                     break;  
  70.                 }  
  71.             }  
  72.         }  
  73.   
  74.         private int getLetterNum() throws KeeperException, InterruptedException {  
  75.             Stat stat = zkClient.exists(MAILBOX_ROOT_PATH, null);  
  76.             int letterNum = stat.getNumChildren();  
  77.             return letterNum;  
  78.         }  
  79.   
  80.         public void setZkClient(ZooKeeper zkClient) {  
  81.             this.zkClient = zkClient;  
  82.         }  
  83.     }  
  84.   
  85.     // 消费者线程，负责发送信件  
  86.     static class Consumer extends Thread {  
  87.   
  88.         ZooKeeper zkClient;  
  89.   
  90.         @Override  
  91.         public void run() {  
  92.             while (true) {  
  93.                 try {  
  94.                     if (getLetterNum() == 0) { // 信箱已空  
  95.                         System.out.println("mailBox has been empty");  
  96.                         // 创建Watcher，监控子节点的变化  
  97.                         Watcher watcher = new Watcher() {  
  98.                             @Override  
  99.                             public void process(WatchedEvent event) {  
  100.                                 // 消费者已停止，只有生产者在活动，所以只可能出现收取信件的动作  
  101.                                 System.out.println("mailBox has been not empty");  
  102.                                 synchronized (this) {  
  103.                                     notify(); // 唤醒消费者  
  104.                                 }  
  105.                             }  
  106.                         };  
  107.                         zkClient.getChildren(MAILBOX_ROOT_PATH, watcher);  
  108.   
  109.                         synchronized (watcher) {  
  110.                             watcher.wait(); // 阻塞消费者  
  111.                         }  
  112.                     } else {  
  113.                         // 线程随机休眠数毫秒，模拟现实中的费时操作  
  114.                         int sleepMillis = (int) (Math.random() * 1000);  
  115.                         Thread.sleep(sleepMillis);  
  116.   
  117.                         // 发送信件，删除序号最小的子节点  
  118.                         String firstLetter = getFirstLetter();  
  119.                         zkClient.delete(MAILBOX_ROOT_PATH + "/" +firstLetter, -1);  
  120.                         System.out.println("a letter has been delivered: " + firstLetter   
  121.                                 + ", letter num: " + getLetterNum());  
  122.                     }  
  123.                 } catch (Exception e) {  
  124.                     System.out.println("consumer equit task becouse of exception !");  
  125.                     e.printStackTrace();  
  126.                     break;  
  127.                 }  
  128.             }  
  129.         }  
  130.   
  131.         private int getLetterNum() throws KeeperException, InterruptedException {  
  132.             Stat stat = zkClient.exists(MAILBOX_ROOT_PATH, false);  
  133.             int letterNum = stat.getNumChildren();  
  134.             return letterNum;  
  135.         }  
  136.   
  137.         private String getFirstLetter() throws KeeperException, InterruptedException {  
  138.             List<String> letterPaths = zkClient.getChildren(MAILBOX_ROOT_PATH, false);  
  139.             Collections.sort(letterPaths);  
  140.             return letterPaths.get(0);  
  141.         }  
  142.   
  143.         public void setZkClient(ZooKeeper zkClient) {  
  144.             this.zkClient = zkClient;  
  145.         }  
  146.     }  
  147.   
  148.     public static void main(String[] args) throws IOException {  
  149.         // 开启生产者线程  
  150.         Producer producer = new Producer();  
  151.         ZooKeeper zkClientA = new ZooKeeper("127.0.0.1:2181", 3000, null);  
  152.         producer.setZkClient(zkClientA);  
  153.         producer.start();  
  154.   
  155.         // 开启消费者线程  
  156.         Consumer consumer = new Consumer();  
  157.         ZooKeeper zkClientB = new ZooKeeper("127.0.0.1:2181", 3000, null);  
  158.         consumer.setZkClient(zkClientB);  
  159.         consumer.start();  
  160.     }  
  161. }  

打印结果如下：

ZooKeeper之（六）应用实例

上例中还有一个可以改进的地方，在分布式环境下，像MAILBOX_MAX_SIZE这类常量是被多台机器共用的，而且运行期间可能发生改变，比如邮筒上限需要从10改为20，只能停掉机器，然后改动每台机器上的参数，再重新部署。可是，如果该服务不允许停机，而且部署在数十台机器上，让参数在运行时生效且保持一致，怎么办？

这就涉及到了ZooKeeper另一个典型的应用场景——配置中心。被多台机器共享的参数可以托管在ZNode上，对该参数关心的机器在Znode上注册Watcher，一旦该参数发生变化，注册者会收到消息，然后做出相应的调整。

 

ZooKeeper的作用当然不止于此，更多的应用场景就需要使用者在实际项目中发掘跟探索了，毕竟，纸上得来终觉浅，实践出真知。

相关推荐