zookeeper-server-client 简单例子

高峰之巅

客户端将初始值加1，然后写入server。
2个客户端输出的值的顺序是一致的。 说明客户端可已感知其他客户端的处理结果。

github 地址
https://github.com/noobthinker/zookeeper-simple-demo
贴代码区域
server

public class Server {
NIOServerCnxnFactory serverCnxnFactory;
public void zkStart(Conf conf) throws IOException, InterruptedException {
ZooKeeperServer zkServer = new ZooKeeperServer();
File dataDirectory = new File(conf.DIR);
FileTxnSnapLog ftxn = new FileTxnSnapLog(dataDirectory, dataDirectory);
zkServer.setTxnLogFactory(ftxn);
zkServer.setTickTime(conf.TICK_TIME);
serverCnxnFactory = new NIOServerCnxnFactory();
serverCnxnFactory.configure(new InetSocketAddress(conf.PORT), conf.MAX_CLIENT_CONNECTIONS);
serverCnxnFactory.startup(zkServer);
}
}


client

public class Client implements Watcher, AsyncCallback.StatCallback {
ZooKeeper zk;
boolean dead;
String znode;
Stat stat;
int index;
public void initZkClient(String ipport,int timetout,String znode) {
try {
this.znode = znode;
zk = new ZooKeeper(ipport, timetout, this);
stat = zk.exists(znode, false);
if(null == stat){
zk.create(znode,new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,
CreateMode.PERSISTENT);
}
zk.getData(znode,this,stat);
} catch (IOException e) {
e.printStackTrace();
}catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public void processResult(int rc, String s, Object o, Stat stat) {
boolean exists;
switch (rc) {
case KeeperException.Code.Ok:
exists = true;
break;
case KeeperException.Code.NoNode:
exists = false;
break;
case KeeperException.Code.SessionExpired:
case KeeperException.Code.NoAuth:
dead = true;
return;
default:
zk.exists(znode, true, this, null);
return;
}
byte b[] = null;
if (exists) {
try {
b = zk.getData(znode, false, null);
System.out.println(new String(b));
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
return;
}
}
}
public void process(WatchedEvent event) {
String path = event.getPath();
if (event.getType() == Event.EventType.None) {
switch (event.getState()) {
case SyncConnected:
break;
case Expired:
dead = true;
break;
}
} else {
if (path != null && path.equals(znode)) {
zk.exists(znode, true, this, null);
}
}
}
public void dataMaker(int i){
index=i;
int max=index+10;
boolean run=true;
while(run){
try {
zk.setData(znode, ((index++) + "").getBytes(), -1);
Thread.sleep(50);
if(index>max){
run=false;
zk.close();
}
} catch (KeeperException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}catch (Exception e){
e.printStackTrace();
}
}
}
}


conf

public class Conf {
public String DIR;
public int TICK_TIME;
public int PORT;
public int MAX_CLIENT_CONNECTIONS;
}


test 类
server

public class ZkServer {
public static void main(String[]args){
Conf conf = new Conf();
conf.DIR="/zookeeper";
conf.PORT=22801;
conf.TICK_TIME=2000;
conf.MAX_CLIENT_CONNECTIONS=60;
Server srv = new Server();
try {
srv.zkStart(conf);
} catch (IOException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

client1

public class ZkClient1 {
public static void main(String[]args){
Client client = new Client();
client.initZkClient("localhost:22801",3000,"/zka");
client.dataMaker(0);
}
}

client1 输出

0 20 1 21 2 22 3 23 4 24 5 25 6 26 7 27 8 28 9 29 10 30


client2

public class ZkClient2 {
public static void main(String[]args){
Client client = new Client();
client.initZkClient("localhost:22801",3000,"/zka");
client.dataMaker(20);
}
}


client2输出

20 1 21 2 22 3 23 4 24 5 25 6 26 7 27 8 28 9 29 10 30


2个端输出一致。