Zookeeper實現分布式集群監控

Zookeeepr實現分布式集群監控

Zookeeper中節點有兩種：臨時節點和永久節點

從類型上看節點又可以分為四種節點類型：PERSIST，PERSIST_SEQUENTIAL,EPHEMERAL,EPHEMERAL_SEQUENTIAL

 

臨時節點有一個特點：當創建臨時節點的程序停掉之后，這個臨時節點就會消失。

監視器的特點：可以給zk中的節點注冊監視器，見識這個節點的變化情況。

監視器注冊一次，只能使用一次，多次使用就要多次注冊。

我們利用這個Zookeeper的臨時節點特性+監視器(Watch)來實現分布式集群監控

我們在/monitor（永久節點）下創建臨時節點。

 

實際上，zookeeper的sdk不是特別好用，很多邊界情況需要用戶自己處理。curator是對Zookeeper的sdk進行了封裝，所以說使用curator操作Zookeeper更加方便

在maven官網找到curator的依賴

 

<!-- https://mvnrepository.com/artifact/org.apache.curator/curator-framework,支持zookeeper3.4.6-->
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-framework</artifactId>
    <version>2.10.0</version>
</dependency>

 

我們通過curator來使用zookeeper，那么我們就必須知道如何使用curator來連上zookeeper，下面代碼是官方文檔所給出

 

RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3)CuratorFramework client = CuratorFrameworkFactory.newClient(zookeeperConnectionString, retryPolicy);
client.start();

 

開始我們的代碼

TestCurator.java ,實現功能：創建Zookeeper臨時節點

package zkdemo;

import java.net.InetAddress;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;  

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.junit.Test;


public class TestCurator {
    
    @Test
    public void test1() throws Exception{
        //1000：表示curator鏈接zk的時候超時時間是多少 3：表示鏈接zk的最大重試次數
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        String connectString = "djt1:2181,djt2:2181,djt3:2181,djt4:2181,djt5:2181";
        int sessionTimeoutMs = 5000;//這個值只能在4000-40000ms之間表示鏈接斷開后多長時間臨時節點會小時
        int connectionTimeoutMs = 3000;//獲取鏈接的超時時間
        //創建一個zk連接
        CuratorFramework client = CuratorFrameworkFactory.newClient(connectString, sessionTimeoutMs
                ,connectionTimeoutMs,retryPolicy);
        
        client.start();
        
        InetAddress localHost = InetAddress.getLocalHost();
        String ip = localHost.getHostAddress();
        
        client.create().creatingParentsIfNeeded()
        .withMode(CreateMode.EPHEMERAL)//指定節點類型
        .withACL(Ids.OPEN_ACL_UNSAFE)//指定設置節點權限信息
        .forPath("/monitor/"+ip);//指定節點名稱
        
        while(true)
        {
            ;
        }
    }
}

 

ZkNodeWatcher.java 實現功能:注冊watcher，監視節點的變化

 

package zk;

import java.util.List;
import java.util.ArrayList;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;



public class ZkNodeWatcher implements Watcher{
    CuratorFramework client;
    List<String> childrenList = new ArrayList<String>();
    List<String> newChildrenList = new ArrayList<String>();

    public  ZkNodeWatcher(){
        //1000：表示curator鏈接zk的時候超時時間是多少 3：表示鏈接zk的最大重試次數
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        String connectString = "djt1:2181,djt2:2181,djt3:2181,djt4:2181,djt5:2181";
        int sessionTimeoutMs = 5000;//這個值只能在4000-40000ms之間表示鏈接斷開后多長時間臨時節點會消失
        int connectionTimeoutMs = 3000;//獲取鏈接的超時時間
        //創建一個zk連接
        client = CuratorFrameworkFactory.newClient(connectString, sessionTimeoutMs
                ,connectionTimeoutMs,retryPolicy);
        client.start();
        
        //監視monitor節點,獲取下面的所有子節點的變化情況
        try {
            childrenList = client.getChildren().usingWatcher(this).forPath("/monitor");
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
        
    

    /**
     * 實現一個zk監視器，監視某個節點的變化情況
     * 
     * 這個監視程序需要一直運行
     * @CPH
     */
    
    public void process(WatchedEvent event) {
        System.out.println("我被調用了");
        try {
            newChildrenList = client.getChildren().usingWatcher(this).forPath("/monitor");
            for(String ip : childrenList)
            {
                if(!newChildrenList.contains(ip)){
                    System.out.println("節點消失了"+ip);
                    //TODO 給管理員發送短信什么的
                    
                }
            }
            
        for(String ip : newChildrenList){
            if(!childrenList.contains(ip)){
                System.out.println("節點新增"+ip);
            }
        }
        //重要
        childrenList = newChildrenList;
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        
        
        
    }
    
    public void start(){
        while (true){;}
        }

    public static void main(String[] args) {
        ZkNodeWatcher watcher = new ZkNodeWatcher();
        watcher.start();
    }
}

 

 

 

 

 

我們先開啟Zookeeper集群，啟動/bin/zkCli.sh,然后啟動這2個集群，我們可以看到由對應的ip目錄，這個ip不是虛擬機的ip，而是本地的ip，同時我們console下看到

然后暫停TestCurator.java,不一會兒，就會看到

這樣，通過Zookeeper實現分布式集群監控的功能就完成了！