public class DistributedClientMy {

???? // 超時(shí)時(shí)間

???? private static final int SESSION_TIMEOUT = 5000;

???? // zookeeper server列表

???? private String hosts = "spark01:2181,spark02:2181,spark03:2181";

???? private String groupNode = "locks";

???? private String subNode = "sub";

???? private boolean haveLock = false;

?

???? private ZooKeeper zk;

???? // 當(dāng)前client創(chuàng)建的子節(jié)點(diǎn)

???? private volatile String thisPath;

?

???? /**

???? ?* 連接zookeeper

???? ?*/

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

???? ???? zk = new ZooKeeper("spark01:2181", SESSION_TIMEOUT, new Watcher() {

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

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

?

???? ???? ???? ???? ???? // 子節(jié)點(diǎn)發(fā)生變化

???? ???? ???? ???? ???? if (event.getType() == EventType.NodeChildrenChanged && event.getPath().equals("/" + groupNode)) {

???? ???? ???? ???? ???? ???? // thisPath是否是列表中的最小節(jié)點(diǎn)

???? ???? ???? ???? ???? ???? List<String> childrenNodes = zk.getChildren("/" + groupNode, true);

???? ???? ???? ???? ???? ???? String thisNode = thisPath.substring(("/" + groupNode + "/").length());

???? ???? ???? ???? ???? ???? // 排序

???? ???? ???? ???? ???? ???? Collections.sort(childrenNodes);

???? ???? ???? ???? ???? ???? if (childrenNodes.indexOf(thisNode) == 0) {

???? ???? ???? ???? ???? ???? ???? doSomething();

???? ???? ???? ???? ???? ???? ???? thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE,

???? ???? ???? ???? ???? ???? ???? ???? ???? CreateMode.EPHEMERAL_SEQUENTIAL);

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

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

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

???? ???? ???? ???? ???? e.printStackTrace();

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

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

???? ???? });

?

???? ???? // 創(chuàng)建子節(jié)點(diǎn)

???? ???? thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE,

???? ???? ???? ???? CreateMode.EPHEMERAL_SEQUENTIAL);

?

???? ???? // wait一小會(huì), 讓結(jié)果更清晰一些

???? ???? Thread.sleep(new Random().nextInt(1000));

?

???? ???? // 監(jiān)聽子節(jié)點(diǎn)的變化

???? ???? List<String> childrenNodes = zk.getChildren("/" + groupNode, true);

?

???? ???? // 列表中只有一個(gè)子節(jié)點(diǎn), 那肯定就是thisPath, 說明client獲得鎖

???? ???? if (childrenNodes.size() == 1) {

???? ???? ???? doSomething();

???? ???? ???? thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE,

???? ???? ???? ???? ???? CreateMode.EPHEMERAL_SEQUENTIAL);

???? ???? }

???? }

?

???? /**

???? ?* 共享資源的訪問邏輯寫在這個(gè)方法中

???? ?*/

???? private void doSomething() throws Exception {

???? ???? try {

???? ???? ???? System.out.println("gain lock: " + thisPath);

???? ???? ???? Thread.sleep(2000);

???? ???? ???? // do something

???? ???? } finally {

???? ???? ???? System.out.println("finished: " + thisPath);

???? ???? ???? // 將thisPath刪除, 監(jiān)聽thisPath的client將獲得通知

???? ???? ???? // 相當(dāng)于釋放鎖

???? ???? ???? zk.delete(this.thisPath, -1);

???? ???? }

???? }

?

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

???? ???? DistributedClientMy dl = new DistributedClientMy();

???? ???? dl.connectZookeeper();

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

???? }

?

????

}