ZooKeeper构建配置服务
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);
}
}
页:
[1]