package cn.com.toto.zkonlineoroffline;

?

import java.util.ArrayList;

import java.util.List;

?

import org.apache.zookeeper.WatchedEvent;

import org.apache.zookeeper.Watcher;

import org.apache.zookeeper.ZooKeeper;

?

public class DistributedClient {

??? private static final String connectString = "192.168.106.80:2181,192.168.106.81:2181,192.168.106.82:2181";

??? private static final int sessionTimeout = 2000;

???? private static final String parentNode = "/servers";

???? // 注意:加volatile的意義何在？

???? private volatile List<String> serverList;

???? private ZooKeeper zk = null;

?

???? /**

???? ?* 創建到zk的客戶端連接

???? ?*

???? ?* @throws Exception

???? ?*/

???? public void getConnect() throws Exception {

???????? zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {

????????????? @Override

????????????? public void process(WatchedEvent event) {

?????????????????? // 收到事件通知后的回調函數（應該是我們自己的事件處理邏輯）

?????????????????? try {

?????????????????????? //重新更新服務器列表，并且注冊了監聽

?????????????????????? getServerList();

?

?????????????????? } catch (Exception e) {

?????????????????? }

????????????? }

???????? });

???? }

?

???? /**

???? ?* 獲取服務器信息列表

???? ?*

???? ?* @throws Exception

???? ?*/

???? public void getServerList() throws Exception {

???????? // 獲取服務器子節點信息，并且對父節點進行監聽

???????? List<String> children = zk.getChildren(parentNode, true);

?

???????? // 先創建一個局部的list來存服務器信息

???????? List<String> servers = new ArrayList<String>();

???????? for (String child : children) {

????????????? // child只是子節點的節點名

????????????? byte[] data = zk.getData(parentNode + "/" + child, false, null);

????????????? servers.add(new String(data));

???????? }

???????? // 把servers賦值給成員變量serverList，已提供給各業務線程使用

???????? serverList = servers;

????????

???????? //打印服務器列表

???????? System.out.println(serverList);

???? }

?

???? /**

???? ?* 業務功能

???? ?*

???? ?* @throws InterruptedException

???? ?*/

???? public void handleBussiness() throws InterruptedException {

???????? System.out.println("client start working.....");

???????? Thread.sleep(Long.MAX_VALUE);

???? }

????

???? public static void main(String[] args) throws Exception {

???????? // 獲取zk連接

???????? DistributedClient client = new DistributedClient();

???????? client.getConnect();

???????? // 獲取servers的子節點信息（并監聽），從中獲取服務器信息列表

???????? client.getServerList();

?

???????? // 業務線程啟動

???????? client.handleBussiness();

???? }

}

package cn.com.toto.zkonlineoroffline;

?

import java.util.concurrent.CountDownLatch;

?

import org.apache.zookeeper.CreateMode;

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.ZooDefs.Ids;

import org.apache.zookeeper.data.Stat;

?

public class DistributedServer {

???? private static final String CONNECT_STRING = "192.168.106.80:2181,192.168.106.81:2181,192.168.106.82:2181";

???? private static final int SESSION_TIME_OUT = 2000;

???? private static final String PARENT_NODE = "/servers";

?

???? private ZooKeeper zk = null;

????

???? CountDownLatch countDownLatch = new CountDownLatch(1);

?

???? /**

???? ?* 創建到zk的客戶端連接

???? ?*

???? ?* @throws Exception

???? ?*/

???? public void getConnect() throws Exception {

???????? zk = new ZooKeeper(CONNECT_STRING, SESSION_TIME_OUT, new Watcher() {

????????????? @Override

????????????? public void process(WatchedEvent event) {

?????????????????? if(event.getState()==KeeperState.SyncConnected){

?????????????????????? countDownLatch.countDown();

?????????????????? }

????????????? }

???????? });

???????? countDownLatch.await();

???? }

?

???? /**

???? ?* 向zk集群注冊服務器信息

???? ?*

???? ?* @param hostname

???? ?* @throws Exception

???? ?*/

???? public void registerServer(String hostname) throws Exception {

???????? Stat exists = zk.exists(PARENT_NODE, false);

???????? if(exists==null) zk.create(PARENT_NODE, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

???????? String path = zk.create(PARENT_NODE + "/server", hostname.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

???????? System.out.println(hostname + "is online.." + path);

???? }

?

???? /**

???? ?* 業務功能

???? ?*

???? ?* @throws InterruptedException

???? ?*/

???? public void handleBussiness(String hostname) throws InterruptedException {

???????? System.out.println(hostname + "start working.....");

???????? Thread.sleep(Long.MAX_VALUE);

???? }

?

???? public static void main(String[] args) throws Exception {

???????? DistributedServer server = new DistributedServer();

???????? // 獲取zk連接

???????? server.getConnect();

?

???????? // 利用zk連接注冊服務器信息(主機名)

???????? server.registerServer(args[0]);

?

???????? // 啟動業務功能

???????? server.handleBussiness(args[0]);

???? }

}