Curator簡介
Apache Curator是一個(gè)比較完善的ZooKeeper客戶端框架，通過封裝的一套高級(jí)API 簡化了ZooKeeper的操作。通過查看官方文檔，可以發(fā)現(xiàn)Curator主要解決了三類問題：
- 封裝ZooKeeper client與ZooKeeper server之間的連接處理
- 提供了一套Fluent風(fēng)格的操作API
- 提供ZooKeeper各種應(yīng)用場景(recipe， 比如：分布式鎖服務(wù)、集群領(lǐng)導(dǎo)選舉、共享計(jì)數(shù)器、緩存機(jī)制、分布式隊(duì)列等)的抽象封裝
Curator主要從以下幾個(gè)方面降低了zk使用的復(fù)雜性：
- 重試機(jī)制:提供可插拔的重試機(jī)制, 它將給捕獲所有可恢復(fù)的異常配置一個(gè)重試策略，并且內(nèi)部也提供了幾種標(biāo)準(zhǔn)的重試策略(比如指數(shù)補(bǔ)償)
- 連接狀態(tài)監(jiān)控: Curator初始化之后會(huì)一直對(duì)zk連接進(jìn)行監(jiān)聽，一旦發(fā)現(xiàn)連接狀態(tài)發(fā)生變化將會(huì)作出相應(yīng)的處理
- zk客戶端實(shí)例管理:Curator會(huì)對(duì)zk客戶端到server集群的連接進(jìn)行管理，并在需要的時(shí)候重建zk實(shí)例，保證與zk集群連接的可靠性
- 各種使用場景支持:Curator實(shí)現(xiàn)了zk支持的大部分使用場景（甚至包括zk自身不支持的場景），這些實(shí)現(xiàn)都遵循了zk的最佳實(shí)踐，并考慮了各種極端情況
采用Curator去是使用zookeeper往往操作簡單，代碼量少，可以很簡單地實(shí)現(xiàn)斷線重連，監(jiān)聽器等使用起來也比較簡單，還支持分布式鎖等一些zookeeper常用的功能的代碼實(shí)現(xiàn)。
使用Curator監(jiān)聽目錄的數(shù)據(jù)變化
集群中的機(jī)器去監(jiān)聽zookeeper某一個(gè)目錄的數(shù)據(jù)變化，就可以動(dòng)態(tài)地、實(shí)時(shí)地將新的配置信息修改至集群的每一臺(tái)機(jī)器，省去了手工配置地麻煩，還可以在程序運(yùn)行過程中動(dòng)態(tài)地更新一些配置。采用Curator的NodeCache可以完成一次注冊(cè)n次監(jiān)聽，這個(gè)對(duì)象可以緩存節(jié)點(diǎn)數(shù)據(jù)，在節(jié)點(diǎn)數(shù)據(jù)發(fā)生改變的時(shí)候，就會(huì)觸發(fā)這個(gè)事件。
下面分別是修改配置的客戶端和監(jiān)聽器程序代碼：
修改的客戶端代碼：
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.data.Stat;
public class myTest {
/** zookeeper地址 */
static final String CONNECT_ADDR = "localhost:2181";
/** session超時(shí)時(shí)間 */
static final int SESSION_OUTTIME = 5000;//ms
public static void main(String[] args) throws Exception {
String path = "/myApplication/myIPs";
//1 重試策略：初試時(shí)間為1s 重試10次
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 10);
//2 通過工廠創(chuàng)建連接
CuratorFramework cf = CuratorFrameworkFactory.builder()
.connectString(CONNECT_ADDR)
.sessionTimeoutMs(SESSION_OUTTIME)
.retryPolicy(retryPolicy)
.build();
//3 開啟連接
cf.start();
Stat stat = cf.checkExists().forPath(path);
if (stat == null)
{
System.out.println("節(jié)點(diǎn)尚不存在");
cf.create().forPath(path, "192.168.1.102".getBytes());
}
cf.setData().forPath(path, "192.168.1.102,192.168.2.11,192.168.2.11".getBytes());
cf.delete().forPath(path);
if (cf != null) {
cf.close();
}
}
}
監(jiān)聽器代碼：
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
public class CuratorWatcher1 {
private static List<String> ips = new ArrayList<String>();
/** zookeeper地址 */
static final String CONNECT_ADDR = "localhost:2181";
/** session超時(shí)時(shí)間 */
static final int SESSION_OUTTIME = 5000;//ms
public static void main(String[] args) throws Exception {
//1 重試策略：初試時(shí)間為1s 重試10次
RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 10);
//2 通過工廠創(chuàng)建連接
CuratorFramework cf = CuratorFrameworkFactory.builder()
.connectString(CONNECT_ADDR)
.sessionTimeoutMs(SESSION_OUTTIME)
.retryPolicy(retryPolicy)
.build();
//3 建立連接
cf.start();
/**
* 想要實(shí)現(xiàn)watch一次注冊(cè)n次監(jiān)聽的話，我們需要使用到curator里的一個(gè)NodeCache對(duì)象。
* 這個(gè)對(duì)象可以用來緩存節(jié)點(diǎn)數(shù)據(jù)，并且可以給節(jié)點(diǎn)添加nodeChange事件，當(dāng)節(jié)點(diǎn)的數(shù)據(jù)發(fā)生變化就會(huì)觸發(fā)這個(gè)事件
*/
//4 建立一個(gè)cache緩存 Curator之nodeCache一次注冊(cè)，N次監(jiān)聽
final NodeCache cache = new NodeCache(cf, "/myApplication/myIPs", false);
cache.start(true);
cache.getListenable().addListener(new NodeCacheListener() {
/**
* <B>方法名稱：</B>nodeChanged<BR>
* @see org.apache.curator.framework.recipes.cache.NodeCacheListener#nodeChanged()
*/
@Override
public void nodeChanged() throws Exception {
// 防止節(jié)點(diǎn)被刪除時(shí)發(fā)生錯(cuò)誤
if (cache.getCurrentData() == null) {
System.out.println("獲取節(jié)點(diǎn)數(shù)據(jù)異常，無法獲取當(dāng)前緩存的節(jié)點(diǎn)數(shù)據(jù)，可能該節(jié)點(diǎn)已被刪除");
return;
}
// 獲取節(jié)點(diǎn)最新的數(shù)據(jù)
String data = new String(cache.getCurrentData().getData());
System.out.println(cache.getCurrentData().getPath() + " 節(jié)點(diǎn)的數(shù)據(jù)發(fā)生變化，最新的數(shù)據(jù)為：" + data);
if(data!=null){
String[] ipsArray = data.split(",");
System.out.println("change the Parameter ips !!!");
synchronized(ips){
ips.clear();
if(ipsArray!=null&&ipsArray.length>0){
for(int i=0;i<ipsArray.length;i++){
ips.add(ipsArray[i]);
}
}
}
}
}
});
Thread.sleep(Integer.MAX_VALUE);
// 獲取當(dāng)前客戶端的狀態(tài)
boolean isZkCuratorStarted = cf.isStarted();
System.out.println("當(dāng)前客戶端的狀態(tài)：" + (isZkCuratorStarted ? "連接中..." : "已關(guān)閉..."));
}
}
運(yùn)行結(jié)果：
curator監(jiān)聽器.png
程序持續(xù)運(yùn)行，就可以將全局的靜態(tài)參數(shù)ips 動(dòng)態(tài)修改?？梢钥吹剑?jié)點(diǎn)創(chuàng)建并賦值的過程，更新數(shù)據(jù)的時(shí)候以及刪除的時(shí)候都觸發(fā)了監(jiān)聽器。
當(dāng)然，這只是一個(gè)簡單的demo,很多實(shí)際的場景中，我們還得有參數(shù)修改后的重新加載過程，這里要根據(jù)需求來進(jìn)行編碼。
