Zookeeper 分布式配置管理

CHSHJ

原创
配置中心代码：
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();
}
}  
}