ZooKeeper构建配置服务

rr322w

                      ZooKeeper构建配置服务
* 配置服务是分布式应用所需要的基本服务之一，它使集群中的机器可以共享配置信息中那些公共的部分。
* 简单的说，ZooKeeper可以作为一个具有高可用性的配置存储器，允许分布式应用的参与者检索和更新配置文件。
* 使用ZooKeeper中的观察机制，可以建立一个活跃的配置服务，使那些感兴趣的客户端能够获得配置信息修改的通知。
在每个znode上存储一个键值对，ActiveKeyValueStore 提供了从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

public class ActiveKeyValueStore extends ConnectionWatcher{     private static final Charset CHARSET =Charset.forName("GBk");     private static final int MAX_RETRIES = 5;     private static final long RETRY_PERIOD_SECONDS = 60;           public void write(String path, String value) throws Exception{         int retries = 0;         while(true){             try {                 Stat stat = zk.exists(path, false);                 if(stat == null){                     zk.create(path, value.getBytes(CHARSET), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);                 }else{                     zk.setData(path, value.getBytes(CHARSET), -1);                 }             } catch (KeeperException.SessionExpiredException e) {                 throw e;             }catch(KeeperException e){                 if(retries++ == MAX_RETRIES){                     throw e;                 }                 TimeUnit.SECONDS.sleep(RETRY_PERIOD_SECONDS);             }         }     }           public String read(String path, Watcher watcher) throws Exception{         byte[] data  = zk.getData(path, watcher, null);         return new String(data, CHARSET);     } }

与zookeeper服务创建连接

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23

public class ConnectionWatcher implements Watcher{     private static final int SESSION_TIMEOUT = 5000;     protected ZooKeeper zk;     private CountDownLatch connectedSignal = new CountDownLatch(1);           public void connect(String hosts) throws Exception{         //创建zookeeper实例的时候会启动一个线程连接到zookeeper服务。         zk = new ZooKeeper(hosts, SESSION_TIMEOUT, this);         connectedSignal.await();     }     //当客户端已经与zookeeper建立连接后，Watcher的process方法会被调用。     public void process(WatchedEvent event) {         if(event.getState() == KeeperState.SyncConnected){             connectedSignal.countDown();         }     }     public void close() throws Exception{         zk.close();     } }

ResilientConfigUpdater类提供了管理更新配置信息方法。

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

public class ResilientConfigUpdater {           public static final String PATH = "/config";           private ActiveKeyValueStore store;     private Random random = new Random();           public ResilientConfigUpdater(String hosts) throws Exception{         store = new ActiveKeyValueStore();         store.connect(hosts);     }           public void run() throws Exception{         while(true){             String value = random.nextInt(100)+"";             store.write(PATH, value);                          System.out.printf("Set %s to %s\n", PATH, value);             TimeUnit.SECONDS.sleep(random.nextInt(10));         }     }     public static void main(String[] args) throws Exception {         while(true){             try{                 ResilientConfigUpdater updater = new ResilientConfigUpdater("192.168.44.231");                 updater.run();             }catch(KeeperException.SessionExpiredException e){                 //start a new session             }catch(KeeperException e){                 e.printStackTrace();                 break;             }         }     } }

ConfigWatcher类提供了配置信息变更观察器，在信息修改后会触发显示方法被调用。

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

public class ConfigWatcher implements Watcher{     private ActiveKeyValueStore store;           public ConfigWatcher(String hosts) throws Exception{         store = new ActiveKeyValueStore();         store.connect(hosts);     }           public void displayConfig() throws Exception{         String value = store.read(ConfigUpdater.PATH, this);         System.out.printf("Read %s as %s\n", ConfigUpdater.PATH, value);     }           public void process(WatchedEvent event) {         // TODO Auto-generated method stub         if(event.getType() == EventType.NodeDataChanged){             try {                 displayConfig();             } catch (Exception e) {                 System.out.println("Interrupted. Exiting.");                 Thread.currentThread().interrupt();             }         }     }           public static void main(String[] args) throws Exception {         ConfigWatcher watcher = new ConfigWatcher("192.168.44.231");         watcher.displayConfig();                   Thread.sleep(Long.MAX_VALUE);     } }