zookeeper集群安装与配置

    一、解压zookeeper

    二、在$ZOOKEEPER_HOME/conf下建立zoo.cfg文件（参考配置文件：zoo_sample.cfg），内容为：

# CS通讯心跳时间，Zookeeper 服务器之间或客户端与服务器之间维持心跳的时间间隔，也就是每一个 tickTime 时间就会发送一个心跳。tickTime以毫秒为单位
tickTime=2000

# LF初始化通讯时限，集群中的follower服务器(F)与leader服务器(L)之间初始链接时能容忍的最多心跳数（tickTime的数量）
initLimit=10

# LF同步通讯时限，集群中的follower服务器与leader服务器之间请求和应答之间能容忍的最多心跳数（tickTime的数量）。
syncLimit=5

#数据文件目录，Zookeeper保存数据的目录，默认状况下，Zookeeper将写数据的日志文件也保存在这个目录里。
dataDir=/root/app/zookeeper/data

#客户端链接端口，客户端链接 Zookeeper 服务器的端口，Zookeeper 会监听这个端口，接受客户端的访问请求。
clientPort=2181

#服务器名称与地址：集群信息（服务器编号，服务器地址，LF通讯端口，选举端口）
#这个配置项的书写格式比较特殊，规则以下：
#server.N=YYY:A:B 
server.1=hadoop.master:2888:3888
server.2=hadoop.slave:2888:3888
#端口
#2181：用于客户端的链接
#2888：follower与leader通讯
#3888：leader选举端口

    三、根据zoo.cfg在dataDir配置的路径下建立myid文件，且输入server.N的N内容（例如：当前机器是server.1，那么在    myid中输入1）

    四、复制配置完毕的zookeepr到其余节点

scp -r $ZOOKEEPER_HOME root@hostname:/home

    五、复制完毕以后记得修改myid

    六、在各个节点上分别启动脚本：

$ZOOKEEPER_HOME/bin/zkServer.sh start

    七、检查命令

7.一、#zkServer.sh status
//若是出现Error contacting service. It is probably not running.错误，是由于节点只启动了一个，没法启动集群模式
7.二、#echo ruok | nc localhost 2181
//若是命令成功会返回imok（I'm ok）

    八、zk客户端

$ZOOKEEPER_HOME/bin/zkCli.sh -server hostname:2181
//启动一个zk客户端
//最终要记住一个help命令便可掌握其余命令的用法
（操做以前须要了解相似文件系统的znode，
    简单描述：zk中全部节点都是znode节点,znode节点不只能够做为目录还能够存储数据.）
CREATE 建立znode
SET 设置znode的数据
GET 查询znode
RMR 删除znode
...等等自行查询

    九、Java客户端

    相关代码：

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
public class ZKClient {
    // 链接地址
    private final static String CONN = "192.168.1.200:2181";
    // client与zk超时时间(unit:ms)
    private final static int SESSION_TIMEOUT = 5000;

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

        //一、持久节点：建立的节点成功时,节点便会持久化在zk上
        //二、临时节点：建立节点过程当中一旦发生有意或无心的客户端超时或者关闭,节点都被删除
        ZooKeeper zk = new ZooKeeper(CONN, SESSION_TIMEOUT, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                //接收zk各个时间通知
                System.out.println("------------------------");
                System.out.println("Path:"+event.getPath());
                System.out.println("type:"+event.getType());
                System.out.println("state"+event.getState());
                System.out.println("wrapper:"+event.getWrapper());
                System.out.println("------------------------");
            }
        });
        String path = "/hello";
        byte[] data = "testhello!".getBytes();
        zk.create(path,data, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        zk.close();
    }
}