zookeeper 用什么开发的

　　由于zookeeper的client只有zookeeper一个对象，使用也比较简单，所以就不许要文字说明了，在代码中注释下就ok 了。

1、测试用的main方法
package ClientExample;

public class TestMain {
public static void main(String[] args) {
/*
* 测试流程
* 1、创建sever1的连接client1，并且创建一个永久性的/test节点
* 2、创建一个针对server1的临时节点
* 3、创建server2的连接client21，并创建一个针对server2的临时节点
* 4、创建server3的连接client3，并创建一个针对server3的临时节点
* 5、分别查看client1、client2、client3的三个节点的字节点数量，确定是否同步成功
* 6、修改client1的临时节点内容，然后在在client2和client3中查看
* 7、kill掉client3的线程，然后检查是watcher是否有通知给client1和client2
*/

Thread t1= new ClientThread("127.0.0.1:2181","server1",false);
Thread t2= new ClientThread("127.0.0.1:2182","server2",false);
Thread t3= new ClientThread("127.0.0.1:2183","server3",false);
Thread t4= new ClientThread("127.0.0.1:2181","server4",false);

t1.start();
t2.start();
t3.start();
t4.start();
ControlThread c = new ControlThread(t1, t2, t3, t4);
c.start();
int i=0;
while(true)
{
i++;
i--;

}

/*
* 测试控制台输出：
* connectIP:server4,path:null,state:SyncConnected,type:None
* connectIP:server3,path:/test,state:SyncConnected,type:NodeChildrenChanged
* connectIP:server4,path:/test/server4,state:SyncConnected,type:NodeCreated
* 。。。。。。。。。。。
*
* connectIP:server2,path:null,state:Disconnected,type:None
server2exception,KeeperErrorCode = ConnectionLoss for /test
connectIP:newServer1,path:null,state:SyncConnected,type:None
connectIP:server1,path:/test,state:SyncConnected,type:NodeChildrenChanged
connectIP:server4,path:/test/server2,state:SyncConnected,type:NodeDeleted
connectIP:server4,path:/test,state:SyncConnected,type:NodeChildrenChanged
connectIP:newServer1,path:/test,state:SyncConnected,type:NodeChildrenChanged
connectIP:server3,path:/test/server2,state:SyncConnected,type:NodeDeleted
connectIP:server3,path:/test,state:SyncConnected,type:NodeChildrenChanged
*/
}
}

2、zookeeper封装的接口：

package ClientExample;

import java.io.IOException;
import java.util.List;
import org.apache.zookeeper.KeeperException;
/**
* zookeeper的操作封装接口，实现了常用的操作
* 创建、销毁、写入、修改、查询等。
* @author ransom
*
*/
public interface ServerOperation {
void init(String address,String serverName) throws IOException;
void destroy() throws InterruptedException;
List getChilds(String path) throws KeeperException,
InterruptedException;
String getData(String path) throws KeeperException, InterruptedException;
void changeData(String path, String data) throws KeeperException,
InterruptedException;
void delData(String path) throws KeeperException, InterruptedException;
void apendTempNode(String path, String data) throws KeeperException,
InterruptedException;
void apendPresistentNode(String path, String data) throws KeeperException,
InterruptedException;

void delNode(String path) throws KeeperException, InterruptedException;
boolean exist(String path) throws KeeperException, InterruptedException;
}

3、接口的实现：
package ClientExample;

import java.io.IOException;
import java.util.List;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooDefs.Ids;
public class ServerConnector implements ServerOperation {
// 创建一个Zookeeper实例，第一个参数为目标服务器地址和端口，第二个参数为Session超时时间，第三个为节点变化时的回调方法
private ZooKeeper zk = null;
public void init(String address,String serverName) throws IOException {
zk = new ZooKeeper(address, 500000,
new MultiWatcher(serverName));
}
@Override
public void destroy() throws InterruptedException {
// TODO Auto-generated method stub
if (zk != null) {
zk.close();
}
}
@Override
public List getChilds(String path) throws KeeperException, InterruptedException {
// TODO Auto-generated method stub
if (zk != null) {
return zk.getChildren(path, true);
}
return null;
}
@Override
public String getData(String path) throws KeeperException, InterruptedException {
// TODO Auto-generated method stub
if (zk != null) {
// 取得/root/childone节点下的数据,返回byte[]
byte[] b = zk.getData(path, true, null);
return new String(b);
}
return null;
}
@Override
public void changeData(String path,String data) throws KeeperException, InterruptedException {
// TODO Auto-generated method stub
if (zk != null) {
// 修改节点/root/childone下的数据，第三个参数为版本，如果是-1，那会无视被修改的数据版本，直接改掉
zk.setData(path, data.getBytes(),-1);
}
}
@Override
public void delData(String path) throws InterruptedException, KeeperException {
// TODO Auto-generated method stub
if (zk != null) {
// 删除/root/childone这个节点，第二个参数为版本，－1的话直接删除，无视版本
zk.delete(path, -1);
}
}
@Override
public void delNode(String path) throws InterruptedException, KeeperException {
// TODO Auto-generated method stub
if (zk != null) {
zk.delete(path, -1);
}
}
@Override
public boolean exist(String path) throws KeeperException,
InterruptedException {
// TODO Auto-generated method stub
if (zk != null) {
return zk.exists(path, true)!=null;
}
return false;
}
@Override
public void apendTempNode(String path, String data) throws KeeperException,
InterruptedException {
// TODO Auto-generated method stub
// TODO Auto-generated method stub
if (zk != null)
{
// 创建一个节点root，数据是mydata,不进行ACL权限控制，节点为永久性的(即客户端shutdown了也不会消失)
/*
* 创建一个给定的目录节点 path, 并给它设置数据，
* CreateMode 标识有四种形式的目录节点,分别是
* PERSISTENT：持久化目录节点，这个目录节点存储的数据不会丢失；
* PERSISTENT_SEQUENTIAL：顺序自动编号的目录节点，这种目录节点会根据当前已近存在的节点数自动加 1，然后返回给客户端已经成功创建的目录节点名；
* EPHEMERAL：临时目录节点，一旦创建这个节点的客户端与服务器端口也就是 session 超时，这种节点会被自动删除；
* EPHEMERAL_SEQUENTIAL：临时自动编号节点
*/
zk.create(path, data.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
}
}
@Override
public void apendPresistentNode(String path, String data)
throws KeeperException, InterruptedException {
// TODO Auto-generated method stub
if (zk != null)
{
// 创建一个节点root，数据是mydata,不进行ACL权限控制，节点为永久性的(即客户端shutdown了也不会消失)
/*
* 创建一个给定的目录节点 path, 并给它设置数据，
* CreateMode 标识有四种形式的目录节点,分别是
* PERSISTENT：持久化目录节点，这个目录节点存储的数据不会丢失；
* PERSISTENT_SEQUENTIAL：顺序自动编号的目录节点，这种目录节点会根据当前已近存在的节点数自动加 1，然后返回给客户端已经成功创建的目录节点名；
* EPHEMERAL：临时目录节点，一旦创建这个节点的客户端与服务器端口也就是 session 超时，这种节点会被自动删除；
* EPHEMERAL_SEQUENTIAL：临时自动编号节点
*/
zk.create(path, data.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
}
}
} 4、一个控制的线程，主要用来强制kill掉连接的线程

package ClientExample;

public class ControlThread extends Thread{
public ControlThread(Thread t1,Thread t2,Thread t3,Thread t4)
{
list[0]=t1;
list[1]=t2;
list[2]=t4;
list[3]=t4;
}

private Thread[] list = new Thread[4];
private int num=0;
public void run()
{
while(true)
{
if(num==7)
{
list[2].stop();
System.out.println("kill server3");
}
if(num==15)
{
list[3].stop();
System.out.println("kill server4");
}
try {
sleep(1000);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
}

5、watcher
的实现：

package ClientExample;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
/**
* 提供给多个client使用的watcher
* @author ransom
*
*/
public class MultiWatcher implements Watcher{
public MultiWatcher(String address)
{
connectAddress=address;
}

private String connectAddress=null;

@Override
public void process(WatchedEvent event) {
// TODO Auto-generated method stub
String outputStr="";
if(connectAddress!=null){
outputStr+="connectIP:"+connectAddress;
}
outputStr+=",path:"+event.getPath();
outputStr+=",state:"+event.getState();
outputStr+=",type:"+event.getType();

System.out.println(outputStr);
}
}

转载

多重语言的混合体