zookeeper的簡單搭建，java使用zk的例子和一些坑

一 整合

由於本人的碼雲太多太亂了，於是決定一個一個的整合到一個springboot項目里面。

附上自己的github項目地址 https://github.com/247292980/spring-boot

附上匯總博文地址 https://www.cnblogs.com/ydymz/p/9391653.html

以整合功能

spring-boot，FusionChart，thymeleaf，vue，ShardingJdbc，mybatis-generator，微信分享授權，drools，spring-security，spring-jpa，webjars，Aspect，drools-drt，rabbitmq

 

這次就來整合下簡單的zookeeper搭建

 

二 安裝

這玩意最坑的地方在於，和mq一樣，百度第一的安裝教程缺了一部分。

 

1.先安裝java並配置環境變量。直接百度，進官網下載，還是程序員最喜歡的解壓安裝，贊！

2.將安裝目錄下的conf文件夾下的zoo_sample.cfg復制一份並命名為zoo.cfg，如下圖

 

3.修改一下配置，主要是dataDir和dataDirLog，並且要創建相應的文件夾

配置文件簡單解析

1、tickTime：這個時間是作為Zookeeper 服務器之間或客戶端與服務器之間維持心跳的時間間隔，也就是每個 tickTime 時間就會發送一個心跳。

2、dataDir：顧名思義就是Zookeeper 保存數據的目錄，默認情況下，Zookeeper 將寫數據的日志文件也保存在這個目錄里。

3、dataLogDir：顧名思義就是Zookeeper 保存日志文件的目錄

4、clientPort：這個端口就是客戶端連接Zookeeper 服務器的端口，Zookeeper 會監聽這個端口，接受客戶端的訪問請求。

 

4.進入到bin目錄，並且啟動zkServer.cmd

zkServer.cmd
可以通過以下代碼檢驗有沒有成功啟動，成功后有一個QuorumPeerMain的進程
jps -l –v

5.啟動一個zkCli

zkCli 127.0.0.1:2181  

6.具體的cli可以去w3c里面了解，建議都試一次 https://www.w3cschool.cn/zookeeper/zookeeper_api.html

建議zkCli里面，跑一下這些命令，否則普通demo的java代碼里面就什么都沒有了

create /FirstZnode FirstZnode1

create /SecondZnode SecondZnode2

create /ThirdZnode ThirdZnode3

 

三 代碼

代碼有點多，我只貼集群監聽demo和普通demo，基本上實際使用的都是集群監聽demo，而普通demo是給大家一個過渡的階段。

要是看完普通還看不懂集群監聽，可以github把我的代碼拷下來，里面有個中間態，監聽demo。

 

普通demo，很簡單獲取根目錄下的全部children

public class ZkDemo {
    public static void main(String[] args) throws IOException {
        String hostPort = "localhost:2181";
        List<String> zooChildren = new ArrayList<String>();
        ZooKeeper zk = new ZooKeeper(hostPort, 2000, null);
        if (zk != null) {
            try {
                String zpath = "/";
                zooChildren = zk.getChildren(zpath, false);
                System.out.println("Znodes of '/': ");
                for (String child : zooChildren) {
                    System.out.println(child);
                }
            }  catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

集群監聽demo.ClusterMonitor，寫成線程是讓他當個服務器，小白如果看到idea提示紅色報警有點慌的話，可以看看文件名字有沒有紅色波浪線，沒有即說明代碼正常

public class ClusterMonitor implements Runnable {
    private static String membershipRoot = "/Members";
    private final Watcher connectionWatcher;
    private final Watcher childrenWatcher;
    private ZooKeeper zk;
    boolean alive = true;

    public ClusterMonitor(String HostPort) throws IOException, InterruptedException, KeeperException {
        connectionWatcher = new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                if (event.getType() == Watcher.Event.EventType.None && event.getState() == Watcher.Event.KeeperState.SyncConnected) {
                    System.out.printf("\nEvent Received: %s", event.toString());
                }
            }
        };

        childrenWatcher = new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                System.out.printf("\nEvent Received: %s", event.toString());
                if (event.getType() == Event.EventType.NodeChildrenChanged) {
                    try {
                        //Get current list of child znode,
                        //reset the watch
                        List<String> children = zk.getChildren(membershipRoot, this);
                        wall("!!!Cluster Membership Change!!!");
                        wall("Members: " + children);
                    } catch (KeeperException e) {
                        throw new RuntimeException(e);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        alive = false;
                        throw new RuntimeException(e);
                    }
                }
            }
        };

        zk = new ZooKeeper(HostPort, 2000, connectionWatcher);

        // Ensure the parent znode exists
        if (zk.exists(membershipRoot, false) == null) {
            zk.create(membershipRoot, "ClusterMonitorRoot".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }

        // Set a watch on the parent znode
        List<String> children = zk.getChildren(membershipRoot, childrenWatcher);
        System.err.println("Members: " + children);
    }

    public synchronized void close() {
        try {
            zk.close();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void wall(String message) {
        System.out.printf("\nMESSAGE: %s", message);
    }

    @Override
    public void run() {
        try {
            synchronized (this) {
                while (alive) {
                    wait();
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
            Thread.currentThread().interrupt();
        } finally {
            this.close();
        }
    }

    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        String hostPort = "localhost:2181";
        new ClusterMonitor(hostPort).run();
    }
}

集群監聽demo.ClusterClient，這個可能需要你多開幾個java線程跑，idea改一下啟動參數就好了，然后在服務器里面對比才比較明顯

命令行啟動java線程的代碼

java -cp $CLASSPATH ClusterMonitor

 

public class ClusterClient implements Watcher, Runnable {
    private static String membershipRoot = "/Members";
    ZooKeeper zk;

    public ClusterClient(String hostPort, Long pid) {
        String processId = pid.toString();
        try {
            zk = new ZooKeeper(hostPort, 2000, this);
            if (zk != null) {
                zk.create(membershipRoot + '/' + processId, processId.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }



    public synchronized void close() {
        try {
            zk.close();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void process(WatchedEvent event) {
        System.out.printf("\nEvent Received: %s", event.toString());
    }

    @Override
    public void run() {
        try {
            synchronized (this) {
                while (true) {
                    wait();
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
            Thread.currentThread().interrupt();
        } finally {
            this.close();
        }
    }

    public static void main(String[] args) {
        String hostPort = "localhost:2181";
        //Get the process id
        String name = ManagementFactory.getRuntimeMXBean().getName();
        int index = name.indexOf('@');
        Long processId = Long.parseLong(name.substring(0, index));
        new ClusterClient(hostPort, processId).run();
    }
}

 

四 總結

1.因為基本是和rabbitmq一起寫的博文，我難免會把他們做一番對比，zk和mq的安裝比起來，明顯zk簡單多了。或許原因是mq是用erlang寫的？

2.zk的生態圈很好，畢竟是Apache大廠出品，以上的代碼，都是直接看Apache zookeeper的官網java版的api寫的，比rabbitmq不是好的一點半點，例如mq包的報錯日志問題...

建議去zk官網看一下api或者英語不好的去w3c的zk專欄 https://www.w3cschool.cn/zookeeper/

3.坑也就是要生成幾個文件，看下日志就知道怎么做，頂多是其他人的博文不太完整。

4.zk的例子比mq的多，是因為zk可以直接創建znode，而mq的隊列，通道，廣播必須cli或者gui里面創建，工具包遠沒有zk方便。而我兩篇博文的目的就是java調用而不是設計，所以並沒有細說

5.最后每個項目都建議，諸位自己寫一遍跑一遍，或者直接github我的代碼跑一遍。