SpringBoot 使用 Curator 操作 ZooKeeper
本文主要介紹使用 Curator 訪問 ZooKeeper 的一些基本方法,而不僅僅限于指定的 Recipes,你可以使用 Curator API 任意的訪問 ZooKeeper。Curator 框架提供了一套高級的 API,簡化了 ZooKeeper 的操作。它增加了很多使用 ZooKeeper 開發(fā)的特性,可以處理 ZooKeeper 集群復雜的連接管理和重試機制。
三種 zk 客戶端對比:
| 原生api | zkclient | apache curator | |
|---|---|---|---|
| 優(yōu)點 | 1、session會話超時重連; 2、解決watcher反復注冊; 3、簡化api開發(fā); | 1、解決watch注冊一次就會失效的問題; 2、api更加簡單易用; 3、提供了更多解決方案并且實現(xiàn)簡單,比如分布式鎖; 4、提供了常用的zk工具類; | |
| 缺點 | 1、watch注冊一次后會失效; 2、session超時之后沒有實現(xiàn)重連機制; 3、異常處理繁瑣; 4、只提供了簡單的byte[]數(shù)組的接口,沒有提供針對對象級別的序列化; 5、創(chuàng)建節(jié)點時如果節(jié)點存在拋出異常,需要自行檢查節(jié)點是否存在; 6、刪除節(jié)點無法實現(xiàn)級聯(lián)刪除; | 1、異常處理簡化,拋出RuntimeException; 2、重試機制比較難用; 3、沒有提供各種使用場景的實現(xiàn); |
使用 Curator 需要添加 Maven 依賴(兩個模塊按需添加,ZooKeeper 版本需要與服務器實例中 ZooKeeper 版本一致):
<!-- zookeeper -->
<dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>3.4.14</version>
<exclusions>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
<exclusion>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
</exclusion>
</exclusions>
</dependency>
<!-- curator-framework -->
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-framework</artifactId>
<version>4.2.0</version>
<exclusions>
<exclusion>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
</exclusion>
</exclusions>
</dependency>
<!-- curator-recipes -->
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>4.2.0</version>
<exclusions>
<exclusion>
<groupId>org.apache.curator</groupId>
<artifactId>curator-framework</artifactId>
</exclusion>
</exclusions>
</dependency>
可以看出,Curator 功能分兩大類,一類是對 ZooKeeper 的一些基本命令的封裝,比如增刪改查,即 Framework 模塊;另一類是他的高級特性,即 Recipes 模塊。下面將分兩部分分別進行闡述。
1.Curator Framework的使用
Curator 框架通過 CuratorFrameworkFactory 以工廠模式和 builder 模式創(chuàng)建 CuratorFramework 實例。CuratorFramework 實例都是線程安全的,你應該在你的應用中共享同一個 CuratorFramework 實例。工廠方法 newClient() 提供了一個簡單方式創(chuàng)建實例。而 Builder 提供了更多的參數(shù)控制。一旦你創(chuàng)建了一個 CuratorFramework 實例,你必須調(diào)用它的 start() 啟動,在應用退出時調(diào)用 close() 方法關(guān)閉。
下面的例子演示了兩種創(chuàng)建Curator的方法:
創(chuàng)建方式一(以 builder 模式創(chuàng)建,F(xiàn)luent 風格,建議使用這種):
// 多個地址逗號隔開
CuratorFramework client = CuratorFrameworkFactory.builder().connectString("localhost:2181")
.sessionTimeoutMs(1000) // 連接超時時間
.connectionTimeoutMs(1000) // 會話超時時間
// 剛開始重試間隔為1秒,之后重試間隔逐漸增加,最多重試不超過三次
.retryPolicy(new ExponentialBackoffRetry(1000, 3))
.build();
client.start();
//...
CloseableUtils.closeQuietly(client);//建議放在finally塊中
創(chuàng)建方式二(以工廠模式創(chuàng)建):
// 多個地址逗號隔開
CuratorFramework client = CuratorFrameworkFactory.newClient("localhost:2181",
// 剛開始重試間隔為1秒,之后重試間隔逐漸增加,最多重試不超過三次
new ExponentialBackoffRetry(1000, 3));
client.start();
//...
CloseableUtils.closeQuietly(client); // 建議放在finally塊中
下面看下 CuratorFramework 提供的方法:
| 方法名 | 描述 |
|---|---|
| create() | 開始創(chuàng)建操作,可以調(diào)用額外的方法(比如方式mode或者后臺執(zhí)行background)并在最后調(diào)用forPath()指定要操作的ZNode |
| delete() | 開始刪除操作,可以調(diào)用額外的方法(版本或者后臺處理version or background)并在最后調(diào)用forPath()指定要操作的ZNode |
| checkExists() | 開始檢查ZNode是否存在的操作,可以調(diào)用額外的方法(監(jiān)控或者后臺處理)并在最后調(diào)用forPath()指定要操作的ZNode |
| getData() | 開始獲得ZNode節(jié)點數(shù)據(jù)的操作,可以調(diào)用額外的方法(監(jiān)控、后臺處理或者獲取狀態(tài)watch, background or get stat)并在最后調(diào)用forPath()指定要操作的ZNode |
| setData() | 開始設置ZNode節(jié)點數(shù)據(jù)的操作,可以調(diào)用額外的方法(版本或者后臺處理)并在最后調(diào)用forPath()指定要操作的ZNode |
| getChildren() | 開始獲得ZNode的子節(jié)點列表,以調(diào)用額外的方法(監(jiān)控、后臺處理或者獲取狀態(tài)watch, background or get stat)并在最后調(diào)用forPath()指定要操作的ZNode |
| inTransaction() | 開始是原子ZooKeeper事務,可以復合create, setData, check, and/or delete 等操作然后調(diào)用commit()作為一個原子操作提交 |
后臺操作的通知和監(jiān)控可以通過 ClientListener 接口發(fā)布,你可以在 CuratorFramework 實例上通過 addListener() 注冊 listener,Listener 實現(xiàn)了下面的方法:
- eventReceived() 一個后臺操作完成或者一個監(jiān)控被觸發(fā)
事件類型以及事件的方法如下:
| Event Type | Event Methods |
|---|---|
| CREATE | getResultCode()、getPath() |
| DELETE | getResultCode()、getPath() |
| EXISTS | getResultCode()、getPath()、getStat() |
| GETDATA | getResultCode()、getPath()、getStat()、getData() |
| SETDATA | getResultCode()、getPath()、getStat() |
| CHILDREN | getResultCode()、getPath()、getStat()、getChildren() |
| WATCHED | getWatchedEvent() |
還可以通過 ConnectionStateListener 接口監(jiān)控連接的狀態(tài),強烈推薦你增加這個監(jiān)控器。你可以使用命名空間 Namespace 避免多個應用的節(jié)點的名稱沖突。CuratorFramework 提供了命名空間的概念,這樣 CuratorFramework 會為它的 API 調(diào)用的 path 加上命名空間:
CuratorFramework client = CuratorFrameworkFactory.builder().connectString("localhost:2181")
.namespace("configCenter") // 命名空間
...
.build();
client.start();
client.create().forPath("/search/business/test", data);
//node was actually written to: "/configCenter/search/business/test"
...
CloseableUtils.closeQuietly(client); // 建議放在finally塊中
Curator 還提供了臨時的 CuratorFramework: CuratorTempFramework,3分鐘不活動連接就被關(guān)閉,你也可以指定不活動的時間。創(chuàng)建 builder 時不是調(diào)用 build() 而是調(diào)用 buildTemp()。CuratorTempFramework 只提供了 close()、inTransaction()、getData() 這三個方法。
下面我們看下 Curator 增刪改查 ZooKeeper 的具體寫法:
client.start();
// 判斷是否存在,Stat就是對znode所有屬性的一個映射,stat=null表示節(jié)點不存在
Stat stat = client.checkExists().forPath("/search/business/test");
// 查詢子節(jié)點
List<String> childNodes = client.getChildren()
.forPath("/search/business/test"));
// 創(chuàng)建節(jié)點
client.create().creatingParentsIfNeeded() // 若創(chuàng)建節(jié)點的父節(jié)點不存在則先創(chuàng)建父節(jié)點再創(chuàng)建子節(jié)點
.withMode(CreateMode.PERSISTENT) // 創(chuàng)建的是持久節(jié)點
.withACL(Ids.OPEN_ACL_UNSAFE) // 默認匿名權(quán)限,權(quán)限scheme id:'world,'anyone,:cdrwa
.forPath("/search/business/test","1".getBytes());
// 更新節(jié)點數(shù)據(jù)
client.setData()
.withVersion(2) // 樂觀鎖
.forPath("/search/business/test","0".getBytes());
// 讀取節(jié)點數(shù)據(jù)
Stat stat = new Stat(); // Stat就是對znode所有屬性的一個映射,stat=null表示節(jié)點不存在
String re = new String(client.getData()
.storingStatIn(stat) // 在獲取節(jié)點內(nèi)容的同時把狀態(tài)信息存入Stat對象,如果不寫的話只會讀取節(jié)點數(shù)據(jù)
.forPath("/search/business/test"));
// 刪除節(jié)點
client.delete()
.guaranteed() // 保障機制,若未刪除成功,只要會話有效會在后臺一直嘗試刪除
.deletingChildrenIfNeeded() // 若當前節(jié)點包含子節(jié)點,子節(jié)點也刪除
.withVersion(2) // 指定版本號
.forPath("/search/business/test");
// watcher事件,使用usingWatcher的時候,監(jiān)聽只會觸發(fā)一次,監(jiān)聽完畢后就銷毀
client.getData()
.usingWatcher(new CuratorWatcher() {
@Override
public void process(WatchedEvent event) throws Exception {
logger.info("觸發(fā)watcher, path:{}", event.getPath());
}
})
.forPath("/search/business/test");
// watcher事件, NodeCache一次注冊N次監(jiān), 缺點:沒法監(jiān)聽當前節(jié)點增刪改操作,所以引出了PathChildrenCache
final NodeCache nodeCache = new NodeCache(client, "/search/business/test");
nodeCache.start(true); // 當zkServer與客戶端鏈接的時候, NodeCache會把zk數(shù)據(jù)緩存到我們本地
if (nodeCache.getCurrentData() != null) {
logger.info("節(jié)點初始化數(shù)據(jù)為:{}", new String(nodeCache.getCurrentData().getData()));
} else {
logger.info("節(jié)點初始化數(shù)據(jù)為空");
}
nodeCache.getListenable().addListener(new NodeCacheListener() {
@Override
public void nodeChanged() throws Exception {
if (nodeCache.getCurrentData() != null) {
String re = new String(nodeCache.getCurrentData().getData());
logger.info("節(jié)點路徑:{}, 節(jié)點數(shù)據(jù):{}", nodeCache.getCurrentData().getPath(), re);
} else {
logger.info("當前節(jié)點被刪除了");
}
}
});
// 監(jiān)聽父節(jié)點以下所有的子節(jié)點, 當子節(jié)點發(fā)生變化的時候(增刪改)都會監(jiān)聽到
// 為子節(jié)點添加watcher事件
// PathChildrenCache監(jiān)聽數(shù)據(jù)節(jié)點的增刪改
final PathChildrenCache childrenCache = new PathChildrenCache(client, "/search", true);
// NORMAL:異步初始化, BUILD_INITIAL_CACHE:同步初始化, POST_INITIALIZED_EVENT:異步初始化,初始化之后會觸發(fā)事件
childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
List<ChildData> childDataList = childrenCache.getCurrentData(); // 當前數(shù)據(jù)節(jié)點的子節(jié)點數(shù)據(jù)列表
childrenCache.getListenable().addListener(new PathChildrenCacheListener() {
@Override
public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
if (event.getType().equals(PathChildrenCacheEvent.Type.INITIALIZED)) {
logger.info("子節(jié)點初始化ok..");
} else if (event.getType().equals(PathChildrenCacheEvent.Type.CHILD_ADDED)) {
logger.info("添加子節(jié)點, path:{}, data:{}", event.getData().getPath(), event.getData().getData());
} else if (event.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)) {
logger.info("刪除子節(jié)點, path:{}", event.getData().getPath());
} else if (event.getType().equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)) {
logger.info("修改子節(jié)點, path:{}, data:{}", event.getData().getPath(), event.getData().getData());
}
}
});
CloseableUtils.closeQuietly(client);//建議放在finally塊中
所有這些方法都以 forpath() 結(jié)尾,輔以 watch (監(jiān)聽),withMode (指定模式),和 inBackground (后臺運行) 等方法來使用。
此外,Curator 還支持事務,一組crud操作要么都完成,要么都不完成:
client.start();
// 開啟事務
CuratorTransaction transaction = client.inTransaction();
Collection<CuratorTransactionResult> results = transaction.create().forPath("/curator/1", "0".getBytes())
.and().setData().forPath("/curator/2", "-1".getBytes())
.and().delete().forPath("/curator/3")
.and().commit();
CloseableUtils.closeQuietly(client);//建議放在finally塊中
2.Curator Recipes的使用
Recipes 模塊主要有 Elections (選舉)、Locks (鎖)、Barriers (關(guān)卡)、Atomic (原子量)、Caches、Queues 等。
1.Elections
選舉主要依賴于 LeaderSelector 和 LeaderLatch 兩個類。前者是所有存活的客戶端不間斷的輪流做 Leader。后者是一旦選舉出 Leader,除非有客戶端掛掉重新觸發(fā)選舉,否則不會交出領導權(quán)。這兩者在實現(xiàn)上是可以切換的。
下面看一下 LeaderSelector 的選舉:
/**
* 本類基于leaderSelector實現(xiàn),所有存活的client會公平的輪流做leader
* 如果不想頻繁的變化Leader,需要在takeLeadership方法里阻塞leader的變更! 或者使用 {@link}
* LeaderLatchClient
*/
public class LeaderSelectorClient extends LeaderSelectorListenerAdapter implements Closeable {
private static final Log log = LogFactory.getLog(LeaderSelectorClient.class);
private final String name;
private final LeaderSelector leaderSelector;
private final String PATH = "/leaderselector";
public LeaderSelectorClient(CuratorFramework client, String name) {
this.name = name;
leaderSelector = new LeaderSelector(client, PATH, this);
leaderSelector.autoRequeue();
}
public void start() throws IOException {
leaderSelector.start();
}
@Override
public void close() throws IOException {
leaderSelector.close();
}
/**
* client成為leader后,會調(diào)用此方法
*/
@Override
public void takeLeadership(CuratorFramework client) throws Exception {
int waitSeconds = (int) (5 * Math.random()) + 1;
log.info(name + "是當前的leader");
try {
Thread.sleep(TimeUnit.SECONDS.toMillis(waitSeconds));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
log.info(name + "讓出領導權(quán)\n");
}
}
}
2.Locks
Curator Lock 相關(guān)的實現(xiàn)在 recipes.locks 包里。頂級接口都是 InterProcessLock。我們直接看最有代表性的 InterProcessReadWriteLock 進程內(nèi)部讀寫鎖(可重入讀寫鎖)。什么叫可重入,什么叫讀寫鎖。不清楚的先查好資料吧??傊x寫鎖一定是成對出現(xiàn)的。 我們先定義兩個任務,可并行的執(zhí)行的,和互斥執(zhí)行的。
到此這篇關(guān)于SpringBoot 使用 Curator 操作 ZooKeeper的文章就介紹到這了,更多相關(guān)SpringBoot Curator 操作ZooKeeper內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
Maven的porfile與SpringBoot的profile結(jié)合使用案例詳解
mybatisplus 多表關(guān)聯(lián)條件分頁查詢的實現(xiàn)
Java基礎之FileInputStream和FileOutputStream流詳解

