ZooKeeper學習第四期—構建ZooKeeper應用

一、配置服務

配置服務是分散式應用所需要的基本服務之一，它使叢集中的機器可以共享配置資訊中那些公共的部分。簡單地說，ZooKeeper可以作為一個具有高可用性的配置儲存器，允許分散式應用的參與者檢索和更新配置檔案。使用ZooKeeper中的觀察機制，可以建立一個活躍的配置服務，使那些感興趣的客戶端能夠獲得配置資訊修改的通知。

下面來編寫一個這樣的服務。我們通過兩個假設來簡化所需實現的服務（稍加修改就可以取消這兩個假設）。

第一，我們唯一需要儲存的配置資料

是字串，關鍵字是znode的路徑，因此我們在每個znode上儲存了一個鍵／值對。
第二，在任何時候只有一個客戶端會執行更新操作。

除此之外，這個模型看起來就像是有一個主人(類似於HDFS中的namenode)在更新資訊，而他的工人則需要遵循這些資訊。

在名為ActiveKeyValueStore的類中編寫了如下程式碼：

package org.zk;

import java.nio.charset.Charset;

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

public class ActiveKeyValueStore extends ConnectionWatcher {
    private static final Charset CHARSET=Charset.forName("UTF-8");
    public void write(String path,String value) throws
 
 KeeperException, InterruptedException {
        Stat stat = zk.exists(path, false);
        if(stat==null){
            zk.create(path, value.getBytes(CHARSET),Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }else{
            zk.setData(path, value.getBytes(CHARSET),-1);
        }
    }
    public String read(String path,Watcher watch) throws KeeperException, InterruptedException{
        byte[] data = zk.getData(path, watch, null);
        return new String(data,CHARSET);

    }

}

View Code

write()方法的任務是將一個關鍵字及其值寫到ZooKeeper。它隱藏了建立一個新的znode和用一個新值更新現有znode之間的區別，而是使用exists操作來檢測znode是否存在，然後再執行相應的操作。其他值得一提的細節是需要將字串值轉換為位元組陣列，因為我們只用了UTF-8編碼的getBytes()方法。☆☆☆

read()方法的任務是讀取一個節點的配置屬性。ZooKeeper的getData()方法有三個引數：

（1）路徑
（2）一個觀察物件
（3）一個Stat物件

Stat物件由getData()方法返回的值填充，用來將資訊回傳給呼叫者。通過這個方法，呼叫者可以獲得一個znode的資料和元資料，但在這個例子中，由於我們對元資料不感興趣，因此將Stat引數設為null。

為了說明ActiveKeyValueStore的用法，我們編寫了一個用來更新配置屬性值的類ConfigUpdater，如程式碼1.1所示。

程式碼1.1 用於隨機更新ZooKeeper中的屬性

package org.zk;

import java.io.IOException;
import java.util.Random;
import java.util.concurrent.TimeUnit;

import org.apache.zookeeper.KeeperException;

public class ConfigUpdater {

    public static final String  PATH="/config";

    private ActiveKeyValueStore store;
    private Random random=new Random();

    public ConfigUpdater(String hosts) throws IOException, InterruptedException {
        store = new ActiveKeyValueStore();
        store.connect(hosts);
    }
    public void run() throws InterruptedException, KeeperException{
        while(true){
            String value=random.nextInt(100)+"";
            store.write(PATH, value);
            System.out.printf("Set %s to %s\n",PATH,value);
            TimeUnit.SECONDS.sleep(random.nextInt(100));

        }
    }
    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        ConfigUpdater configUpdater = new ConfigUpdater(args[0]);
        configUpdater.run();
    }
}

View Code

　　這個程式很簡單，ConfigUpdater中定義了一個ActiveKeyValueStore，它在ConfigUpdater的建構函式中連線到ZooKeeper。run()方法永遠在迴圈，在隨機時間以隨機值更新/config znode。

　　作為配置服務的使用者，ConfigWatcher建立了一個ActiveKeyValueStore物件store，並且在啟動之後通過displayConfig()呼叫了store的read()方法，顯示它所讀到的配置資訊的初始值，並將自身作為觀察傳遞給store。當節點狀態發生變化時，再次通過displayConfig()顯示配置資訊，並再次將自身作為觀察傳遞給store，參見程式碼1.2：

例1.2 該用應觀察ZooKeeper中屬性的更新情況，並將其列印到控制檯

package org.zk;

import java.io.IOException;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;

public class ConfigWatcher implements Watcher{
    private ActiveKeyValueStore store;

    @Override
    public void process(WatchedEvent event) {
        if(event.getType()==EventType.NodeDataChanged){
            try{
                dispalyConfig();
            }catch(InterruptedException e){
                System.err.println("Interrupted. exiting. ");
                Thread.currentThread().interrupt();
            }catch(KeeperException e){
                System.out.printf("KeeperException錛?s. Exiting.\n", e);
            }

        }

    }
    public ConfigWatcher(String hosts) throws IOException, InterruptedException {
        store=new ActiveKeyValueStore();
        store.connect(hosts);
    }
    public void dispalyConfig() throws KeeperException, InterruptedException{
        String value=store.read(ConfigUpdater.PATH, this);
        System.out.printf("Read %s as %s\n",ConfigUpdater.PATH,value);
    }

    public static void main(String[] args) throws IOException, InterruptedException, KeeperException {
        ConfigWatcher configWatcher = new ConfigWatcher(args[0]);
        configWatcher.dispalyConfig();
        //stay alive until process is killed or Thread is interrupted
        Thread.sleep(Long.MAX_VALUE);
    }
}

View Code

當ConfigUpdater更新znode時，ZooKeeper產生一個型別為EventType.NodeDataChanged的事件，從而觸發觀察。ConfigWatcher在它的process()方法中對這個事件做出反應，讀取並顯示配置的最新版本。由於觀察僅傳送單次訊號，因此每次我們呼叫ActiveKeyValueStore的read()方法時，都將一個新的觀察告知ZooKeeper來確保我們可以看到將來的更新。但是，我們還是不能保證接收到每一個更新，因為在收到觀察事件通知與下一次讀之間，znode可能已經被更新過，而且可能是很多次，由於客戶端在這段時間沒有註冊任何觀察，因此不會收到通知。對於示例中的配置服務，這不是問題，因為客戶端只關心屬性的最新值，最新值優先於之前的值。但是，一般情況下，這個潛在的問題是不容忽視的。

讓我們看看如何使用這個程式。在一個終端視窗中執行ConfigUpdater，然後在另一個客戶端執行ConfigWatcher，我們可以預先分別在兩個客戶端輸入命令，先不按回車，等兩個客戶端的命令輸入好後，先在執行ConfigUpdater的客戶端按回車，再在另一個客戶端按回車，執行結果如下：

相關推薦