zookeeper集群安裝與配置

    1、解壓zookeeper

    2、在$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選舉端口

    3、根據zoo.cfg在dataDir配置的路徑下創建myid文件，且輸入server.N的N內容（例如：當前機器是server.1，那么在    myid中輸入1）

    4、復制配置完畢的zookeepr到其他節點

scp -r $ZOOKEEPER_HOME root@hostname:/home

    5、復制完畢之后記得修改myid

    6、在各個節點上分別啟動腳本：

$ZOOKEEPER_HOME/bin/zkServer.sh start

    7、檢查命令

7.1、#zkServer.sh status
//如果出現Error contacting service. It is probably not running.錯誤，是因為節點只啟動了一個，無法啟動集群模式
7.2、#echo ruok | nc localhost 2181
//如果命令成功會返回imok（I'm ok）

    8、zk客戶端

$ZOOKEEPER_HOME/bin/zkCli.sh -server hostname:2181
//啟動一個zk客戶端
//最終要記住一個help命令即可掌握其他命令的用法
（操作之前需要了解類似文件系統的znode，
    簡單描述：zk中所有節點都是znode節點,znode節點不僅可以作為目錄還可以存儲數據.）
CREATE 創建znode
SET 設置znode的數據
GET 查詢znode
RMR 刪除znode
...等等自行查詢

    9、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 {

        //1、持久節點：創建的節點成功時,節點便會持久化在zk上
        //2、臨時節點：創建節點過程中一旦發生有意或無意的客戶端超時或者關閉,節點都被刪除
        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();
    }
}