参考:从Paxos到Zookeeper分布式一致性原理和实践
使用的zk依赖是cdh5.16.2的3.4.5
1 2 3 4 5 6 7
| <!-- zookeeper --> <dependency> <groupId>org.apache.zookeeper</groupId> <artifactId>zookeeper</artifactId> <version>3.4.5-cdh5.16.2</version> </dependency>
|
代码,创建一个临时节点,并赋值,此时初始的节点version=0
使用version=-1对这个节点进行修改,然后查看这个stat,version=1,
使用version=1对这个节点进行修改,成功,然后查看这个stat,version=2,
再次使用version=1对这个节点进行修改,失败,抛出BadVersionException
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 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60
| package com.bigdata.zookeeper;
import org.apache.zookeeper.*; import org.apache.zookeeper.data.Stat;
import java.util.List; import java.util.concurrent.CountDownLatch;
public class ZkExample implements Watcher {
public static CountDownLatch connectedSemaphore = new CountDownLatch(1); private static ZooKeeper zk;
public static void main(String[] args) throws Exception { zk = new ZooKeeper("master:2181", 5000, new ZkExample()); System.out.println(zk.getState()); try { connectedSemaphore.await(); // 创建一个节点 String path = "/app6"; zk.create(path, "123".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); zk.getData(path, true, null);
// version=-1,修改节点 Stat stat = zk.setData(path, "555".getBytes(), -1); System.out.println(stat.getVersion());
// version=1,修改节点 Stat stat2 = zk.setData(path, "55555".getBytes(), stat.getVersion()); System.out.println(stat2.getVersion());
// version=1,再次修改节点 Stat stat3 = zk.setData(path, "555555555".getBytes(), stat.getVersion());
Thread.sleep(10000); // 10秒延时
} catch (InterruptedException e) { System.out.println("Zk session established" + e); } }
@Override public void process(WatchedEvent watchedEvent) { System.out.println(watchedEvent); if (Event.KeeperState.SyncConnected == watchedEvent.getState()) { if (Event.EventType.None == watchedEvent.getType() && null == watchedEvent.getPath()) { connectedSemaphore.countDown(); } else if (watchedEvent.getType() == Event.EventType.NodeChildrenChanged) { try { System.out.println(zk.getChildren(watchedEvent.getPath(), true)); } catch (Exception e) {
} }
} } }
|
输出
删除节点
和修改节点一样,version=-1的时候可以正常删除节点,不等于-1的时候需要判断version是不是相等还能正常删除
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 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64
| package com.bigdata.zookeeper;
import org.apache.zookeeper.*; import org.apache.zookeeper.data.Stat;
import java.util.List; import java.util.concurrent.CountDownLatch;
public class ZkExample implements Watcher {
public static CountDownLatch connectedSemaphore = new CountDownLatch(1); // private static Stat stat = new Stat(); private static ZooKeeper zk;
public static void main(String[] args) throws Exception { zk = new ZooKeeper("master:2181", 5000, new ZkExample()); System.out.println(zk.getState()); try { connectedSemaphore.await(); // 创建一个节点 String path = "/app6"; zk.create(path, "123".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); zk.getData(path, true, null);
// version=-1,修改节点 Stat stat = zk.setData(path, "555".getBytes(), -1); System.out.println(stat.getVersion());
// version=1,修改节点 Stat stat2 = zk.setData(path, "55555".getBytes(), stat.getVersion()); System.out.println(stat2.getVersion());
// version=-1,删除节点,抛异常 // zk.delete(path, stat.getVersion()); // version=-1,删除节点,正常 zk.delete(path, stat2.getVersion());
Thread.sleep(10000); // 10秒延时
} catch (InterruptedException e) { System.out.println("Zk session established" + e); } }
@Override public void process(WatchedEvent watchedEvent) { System.out.println(watchedEvent); if (Event.KeeperState.SyncConnected == watchedEvent.getState()) { if (Event.EventType.None == watchedEvent.getType() && null == watchedEvent.getPath()) { connectedSemaphore.countDown(); } else if (watchedEvent.getType() == Event.EventType.NodeChildrenChanged) { try { System.out.println(zk.getChildren(watchedEvent.getPath(), true)); } catch (Exception e) {
} }
} } }
|