Zookeeper 分布式配置管理

原创

配置中心代码：

import java.io.IOException;
import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.data.Stat;

public class ConfigCenter implements Watcher{
	    private ZooKeeper zk = null; 
	    private CountDownLatch connectedSemaphore = new CountDownLatch( 1 ); 
	
	 public void createConnection( String connectString, int sessionTimeout ) { 
	        try { 
	            zk = new ZooKeeper( connectString, sessionTimeout, this ); 
	            connectedSemaphore.await(); 
	        } catch ( InterruptedException e ) { 
	            System.out.println( "连接创建失败，发生 InterruptedException" ); 
	            e.printStackTrace(); 
	        } catch ( IOException e ) { 
	            System.out.println( "连接创建失败，发生 IOException" ); 
	            e.printStackTrace(); 
	        } 
	    } 
      
    public void process(WatchedEvent event) {  
    	 System.out.println( "configcenter 收到事件通知：" + event.getState() +"\n"  ); 
	        if ( KeeperState.SyncConnected == event.getState() ) { 
	            connectedSemaphore.countDown(); 
	        } 
    }  
  
    void updateConfig(String znode,String str) {  
        try {  
            Stat s = this.zk.exists(znode, true);  
            this.zk.setData(znode, str.getBytes(), s.getVersion());  
            byte[] value = this.zk.getData(znode, null, new Stat());
            String res = new String(value);
            System.out.println("res ="+res);
        } catch (Exception e) {  
            e.printStackTrace();  
        }  
    }     
}

监听客户端代码：

import java.io.IOException;
import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event.KeeperState;

public class ZooKeeperWatcher implements Watcher ,Runnable{
	    private ZooKeeper zk = null; 
	    private CountDownLatch connectedSemaphore = new CountDownLatch( 1 ); 
	
	 public void createConnection( String connectString, int sessionTimeout ) { 
//	        this.releaseConnection(); 
	        try { 
	            zk = new ZooKeeper( connectString, sessionTimeout, this ); 
	            connectedSemaphore.await(); 
	        } catch ( InterruptedException e ) { 
	            System.out.println( "连接创建失败，发生 InterruptedException" ); 
	            e.printStackTrace(); 
	        } catch ( IOException e ) { 
	            System.out.println( "连接创建失败，发生 IOException" ); 
	            e.printStackTrace(); 
	        } 
	    } 
   
	 	public void process(WatchedEvent event) {  
			System.out.println( "zookeeperwatcher 收到事件通知：" + event.getState() +"\n"  ); 
				try {
					zk.exists("/conf2", true);////不知道为什么一定要加上这句话，下次事件到来时，才会触发process事件
				} catch (KeeperException e) {
					e.printStackTrace();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
		        if ( KeeperState.SyncConnected == event.getState() ) { 
		            connectedSemaphore.countDown(); 
		        } 
	 	}  
	 
	    public void run() {  
	        try {  
	            synchronized (this) {  
	                           while (true) {  
	                              wait();  
	                            }  
	                        }  
	        } catch (InterruptedException e) {  
	            e.printStackTrace();  
	        }  
	    }  
}

测试代码：

import java.io.IOException;

import org.apache.zookeeper.KeeperException;



public class test {
	 
	public static void main(String[] args) throws IOException, InterruptedException, KeeperException {  
		try{
        ZooKeeperWatcher zw1 = new ZooKeeperWatcher();  
        zw1.createConnection("192.168.26.141:2181", 1000);  
        ZooKeeperWatcher zw2 = new ZooKeeperWatcher();  
        zw2.createConnection("192.168.26.141:2182", 1000);  
        new Thread(zw1).start();  
        new Thread(zw2).start();  
        ConfigCenter cc = new ConfigCenter();  
        cc.createConnection("192.168.26.141:2181", 1000);
        cc.updateConfig("/conf2","a");  
        cc.updateConfig("/conf2","b");  
        cc.updateConfig("/conf2","c");  
        cc.updateConfig("/conf2","d");  
			
		}catch (Exception e) {
			e.printStackTrace();
		}
    }  
	
}