zookeeper

记录下工作的点点滴滴。

好文推荐：http://agapple.iteye.com/blog/1111377

一、zookeeper的使用

下载zookeeper:
http://zookeeper.apache.org/releases.html

1、单机安装

（1）设置Path路径，将java/bin目录加入到path路径
（2）解压zookeeper
（3）将zookeeper目录下config/zoo_sample.cfg改为zoo.cfg
（4）修改zoo.cfg文件，修改dataDir，新增dataLogDir，同时为data文件和log文件建立相应的目录
dataDir=D:/zookeeper/servers/server0/data
dataLogDir=D:/zookeeper/servers/server0/logs
（5）修改zoo.cfg，根据实际需要修改配置
（6）双击bin目录下zkServer.cmd文件即可启动zookeeper服务

2、集群安装

（1）完成单机安装的步骤

（2）按照同样的方式安装几台zookeeper服务，至少3台，依次为server0、server1、server2,，配置同单机安装。

（3）修改每台zk服务的zoo.cfg文件，增加3个配置。保证server0、server1、server2每台服务地址不一样。server.x表示是第x台服务。
server.0=localhost:2881:3881
server.1=localhost:2882:3882
server.2=localhost:2883:3883

（4）修改每台zk服务的zoo.cfg文件，修改clientPort=2181，该属性为该台zk服务对应客户端调用的端口

（5）在每台zk服务的data目录下建立文件，文件名为myid，文件内容与server的序号对应，server.0对应的文件内容为0

（6）启动每台zk服务，即可通过控制台日志看到集群的情况。


3、zk客户端使用

zk客户端程序使用zkclient.jar和zookeeper.jar

依赖包包括：
commons-logging-1.1.3.jar
slf4j相关jar包

实例程序

import java.util.concurrent.CountDownLatch;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

public class ZKConnection {
    /**
     * server列表, 以逗号分割
     */
    protected String hosts = "localhost:2181,localhost:2182,localhost:2183";
    /**
     * 连接的超时时间, 毫秒
     */
    private static final int SESSION_TIMEOUT = 5000;
    private CountDownLatch connectedSignal = new CountDownLatch(1);
    protected ZooKeeper zk;

    /**
     * 连接zookeeper server
     */
    public void connect() throws Exception {
        zk = new ZooKeeper(hosts, SESSION_TIMEOUT, new ConnWatcher());
        // 等待连接完成
        connectedSignal.await();
    }

    public class ConnWatcher implements Watcher {
        public void process(WatchedEvent event) {
            // 连接建立, 回调process接口时, 其event.getState()为KeeperState.SyncConnected
            if (event.getState() ==Event.KeeperState.SyncConnected ) {
                // 放开闸门, wait在connect方法上的线程将被唤醒
                connectedSignal.countDown();
            }
        }
    }
    /**
     * 创建临时节点
     */
    public void create(String nodePath, byte[] data) throws Exception {
        zk.create(nodePath, data, Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
    }

    public byte[] get(String nodePath,boolean wath,Stat stat) throws KeeperException, InterruptedException{
    	return zk.getData(nodePath, wath, stat);
    }
    
    public static void main(String[] args) throws Exception {
		
    	byte[] data=new String("helloworld").getBytes("utf-8");
    	ZKConnection zkConnection=new ZKConnection();
    	zkConnection.connect();
    	
    	zkConnection.create("/mydate", data);
    	
    	Thread.sleep(20000);
    	
    	Stat stat=new Stat();
    	byte[] data1=zkConnection.get("/mydate", true, stat);
    	
    	System.out.println(new String(data1, "utf-8"));
    	
    	
    	
	}
    
    
    
}