SharedCount
SharedCount類的原始碼注釋：
Manages a shared integer. All clients watching the same path will have the up-to-date value of the shared integer (considering ZK’s normal consistency guarantees).
管理一個共享整數，查看同一路徑的所有客戶端都將擁有共享整數的最新值（考慮到
ZK的正常一致性保證），
SharedCount類的原始碼：
public class SharedCount implements Closeable, SharedCountReader, Listenable<SharedCountListener>
{
// 存盤監聽器
private final Map<SharedCountListener, SharedValueListener> listeners = Maps.newConcurrentMap();
// SharedValue用于管理一個共享值，SharedCount將共享整數的維護委托給該SharedValue實體
private final SharedValue sharedValue;
/**
* client: CuratorFramework實體，用于與Zookeeper進行互動
* path: 共享計數器的路徑，即存盤共享整數的位置
* seedValue: 如果共享計數器的路徑尚未創建，則設定該值為共享整數的初始值
*/
public SharedCount(CuratorFramework client, String path, int seedValue)
{
// 初始化SharedValue實體
sharedValue = new SharedValue(client, path, toBytes(seedValue));
}
protected SharedCount(CuratorFramework client, String path, SharedValue sv)
{
sharedValue = sv;
}
// 獲取共享整數的值
@Override
public int getCount()
{
return fromBytes(sharedValue.getValue());
}
// 獲取共享整數的版本和值，封裝在VersionedValue實體中
@Override
public VersionedValue<Integer> getVersionedValue()
{
VersionedValue<byte[]> localValue = sharedValue.getVersionedValue();
return new VersionedValue<Integer>(localValue.getVersion(), fromBytes(localValue.getValue()));
}
/**
* 更改共享整數的值，而不管其先前的狀態如何
*/
public void setCount(int newCount) throws Exception
{
sharedValue.setValue(toBytes(newCount));
}
@Deprecated
public boolean trySetCount(int newCount) throws Exception
{
return sharedValue.trySetValue(toBytes(newCount));
}
/**
* 僅當共享整數的版本和指定的版本一致時，才更改共享整數的值
*/
public boolean trySetCount(VersionedValue<Integer> previous, int newCount) throws Exception
{
VersionedValue<byte[]> previousCopy = new VersionedValue<byte[]>(previous.getVersion(), toBytes(previous.getValue()));
return sharedValue.trySetValue(previousCopy, toBytes(newCount));
}
// 添加監聽器
@Override
public void addListener(SharedCountListener listener)
{
addListener(listener, MoreExecutors.directExecutor());
}
// 添加監聽器與執行該監聽器的Executor實體
@Override
public void addListener(final SharedCountListener listener, Executor executor)
{
// 使用匿名內部類創建SharedValueListener實體
SharedValueListener valueListener = new SharedValueListener()
{
@Override
public void valueHasChanged(SharedValueReader sharedValue, byte[] newValue) throws Exception
{
// 通過SharedValueListener實體來通知listener共享整數的值改變
listener.countHasChanged(SharedCount.this, fromBytes(newValue));
}
@Override
public void stateChanged(CuratorFramework client, ConnectionState newState)
{
// 通過SharedValueListener實體來通知listener共享整數的狀態改變
listener.stateChanged(client, newState);
}
};
// 將SharedValueListener實體和Executor實體添加到sharedValue實體中
sharedValue.getListenable().addListener(valueListener, executor);
// 將listener和SharedValueListener實體的映射存盤到listeners屬性中
listeners.put(listener, valueListener);
}
// 移除監聽器
@Override
public void removeListener(SharedCountListener listener)
{
// 洗掉listeners屬性中以該監聽器為key的鍵值對
// 如果listeners屬性存在以該監聽器為key的鍵值對
// 則會獲得對應的SharedValueListener實體
SharedValueListener valueListener = listeners.remove(listener);
if(valueListener != null) {
// 如果listeners屬性存在以該監聽器為key的鍵值對
// 則洗掉sharedValue實體中對應的SharedValueListener實體
sharedValue.getListenable().removeListener(valueListener);
}
}
/**
* 必須先啟動共享計數器，然后才能使用它
*/
public void start() throws Exception
{
sharedValue.start();
}
// 使用完共享計數器后需要呼叫close()進行關閉
@Override
public void close() throws IOException
{
sharedValue.close();
}
// 將整數轉換成位元組陣列
@VisibleForTesting
static byte[] toBytes(int value)
{
byte[] bytes = new byte[4];
ByteBuffer.wrap(bytes).putInt(value);
return bytes;
}
// 將位元組陣列轉換成整數
private static int fromBytes(byte[] bytes)
{
return ByteBuffer.wrap(bytes).getInt();
}
}
SharedCount將共享整數的維護委托給了SharedValue，SharedValue的原始碼以后再介紹，
測驗
pom.xml：
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.kaven</groupId>
<artifactId>zookeeper</artifactId>
<version>1.0-SNAPSHOT</version>
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>5.2.0</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.22</version>
</dependency>
</dependencies>
</project>
CuratorFrameworkProperties類（提供CuratorFramework需要的一些配置資訊，以及創建CuratorFramework實體的方法）：
package com.kaven.zookeeper;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.retry.ExponentialBackoffRetry;
public class CuratorFrameworkProperties {
// 連接地址
public static final String CONNECT_ADDRESS = "192.168.1.3:9000";
// 連接超時時間
public static final int CONNECTION_TIMEOUT_MS = 40000;
// Session超時時間
public static final int SESSION_TIMEOUT_MS = 10000;
// 命名空間
public static final String NAMESPACE = "MyNamespace";
// 重試策略
public static final RetryPolicy RETRY_POLICY = new ExponentialBackoffRetry(1000, 3);
public static CuratorFramework getCuratorFramework() {
// 創建CuratorFramework實體
CuratorFramework curator = CuratorFrameworkFactory.builder()
.connectString(CuratorFrameworkProperties.CONNECT_ADDRESS)
.retryPolicy(CuratorFrameworkProperties.RETRY_POLICY)
.connectionTimeoutMs(CuratorFrameworkProperties.CONNECTION_TIMEOUT_MS)
.sessionTimeoutMs(CuratorFrameworkProperties.SESSION_TIMEOUT_MS)
.namespace(CuratorFrameworkProperties.NAMESPACE)
.build();
curator.start();
assert curator.getState().equals(CuratorFrameworkState.STARTED);
return curator;
}
}
SharedCountRunnable類（實作了Runnable介面，模擬分布式節點使用共享計數器）：
package com.kaven.zookeeper;
import lombok.SneakyThrows;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.shared.SharedCount;
import java.util.Random;
public class SharedCountRunnable implements Runnable{
@SneakyThrows
@Override
public void run() {
// 使用不同的CuratorFramework實體，表示不同的分布式節點
CuratorFramework curator = CuratorFrameworkProperties.getCuratorFramework();
// 模擬隨機加入的分布式節點
int randomSleep = new Random().nextInt(10000);
Thread.sleep(randomSleep);
// 共享計數器的路徑
String path = "/kaven";
int seedValue = 0;
// 創建SharedCount實體，用于提供共享計數器功能
// 如果共享計數器的路徑尚未創建，則使用seedValue的值作為共享整數的初始值
SharedCount sharedCount = new SharedCount(curator, path, seedValue);
// 必須先啟動共享計數器，然后才能使用它
sharedCount.start();
// 使用亂數來設定共享整數的值
int randomValue = new Random().nextInt(1000);
sharedCount.setCount(randomValue);
System.out.println(Thread.currentThread().getName() + " 修改共享計數器的值為 " + randomValue);
Thread.sleep(randomSleep);
System.out.println(
"*** " + Thread.currentThread().getName() +
" 獲取共享計數器的值為 " + sharedCount.getCount());
// 使用完共享計數器后需要呼叫close()進行關閉
sharedCount.close();
}
}
啟動類：
package com.kaven.zookeeper;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class Application {
private static final ExecutorService EXECUTOR_SERVICE = Executors.newCachedThreadPool();
public static void main(String[] args) throws Exception {
// 分布式節點處理業務
for (int i = 0; i < 15; i++) {
EXECUTOR_SERVICE.execute(new SharedCountRunnable());
}
}
}
模擬15個分布式節點使用共享計數器，輸出如下所示：
pool-1-thread-6 修改共享計數器的值為 525
pool-1-thread-2 修改共享計數器的值為 379
pool-1-thread-12 修改共享計數器的值為 60
pool-1-thread-4 修改共享計數器的值為 387
pool-1-thread-5 修改共享計數器的值為 821
pool-1-thread-15 修改共享計數器的值為 273
pool-1-thread-9 修改共享計數器的值為 410
*** pool-1-thread-2 獲取共享計數器的值為 410
pool-1-thread-8 修改共享計數器的值為 897
pool-1-thread-1 修改共享計數器的值為 871
pool-1-thread-10 修改共享計數器的值為 803
*** pool-1-thread-12 獲取共享計數器的值為 803
pool-1-thread-13 修改共享計數器的值為 210
*** pool-1-thread-6 獲取共享計數器的值為 210
*** pool-1-thread-4 獲取共享計數器的值為 210
pool-1-thread-11 修改共享計數器的值為 227
pool-1-thread-7 修改共享計數器的值為 445
pool-1-thread-14 修改共享計數器的值為 871
pool-1-thread-3 修改共享計數器的值為 705
*** pool-1-thread-5 獲取共享計數器的值為 705
*** pool-1-thread-15 獲取共享計數器的值為 705
*** pool-1-thread-9 獲取共享計數器的值為 705
*** pool-1-thread-8 獲取共享計數器的值為 705
*** pool-1-thread-1 獲取共享計數器的值為 705
*** pool-1-thread-10 獲取共享計數器的值為 705
*** pool-1-thread-13 獲取共享計數器的值為 705
*** pool-1-thread-11 獲取共享計數器的值為 705
*** pool-1-thread-7 獲取共享計數器的值為 705
*** pool-1-thread-14 獲取共享計數器的值為 705
*** pool-1-thread-3 獲取共享計數器的值為 705
輸出符合預期，
public class SharedCountRunnable implements Runnable{
@SneakyThrows
@Override
public void run() {
// 使用不同的CuratorFramework實體，表示不同的分布式節點
CuratorFramework curator = CuratorFrameworkProperties.getCuratorFramework();
// 模擬隨機加入的分布式節點
int randomSleep = new Random().nextInt(1000);
Thread.sleep(randomSleep);
// 共享計數器的路徑
String path = "/kaven";
int seedValue = 0;
// 創建SharedCount實體，用于提供共享計數器功能
// 如果共享計數器的路徑尚未創建，則使用seedValue的值作為共享整數的初始值
SharedCount sharedCount = new SharedCount(curator, path, seedValue);
// 必須先啟動共享計數器，然后才能使用它
sharedCount.start();
// 使用亂數來設定共享整數的值
int randomValue = new Random().nextInt(1000);
// 獲取VersionedValue實體用于修改共享整數的值
sharedCount.trySetCount(sharedCount.getVersionedValue(), randomValue);
System.out.println(Thread.currentThread().getName() + " 修改共享計數器的值為 " + randomValue);
Thread.sleep(randomSleep);
System.out.println(
"*** " + Thread.currentThread().getName() +
" 獲取共享計數器的值為 " + sharedCount.getCount());
// 使用完共享計數器后需要呼叫close()進行關閉
sharedCount.close();
}
}
輸出如下所示：
pool-1-thread-7 修改共享計數器的值為 797
pool-1-thread-1 修改共享計數器的值為 291
pool-1-thread-2 修改共享計數器的值為 85
*** pool-1-thread-2 獲取共享計數器的值為 291
pool-1-thread-12 修改共享計數器的值為 696
pool-1-thread-8 修改共享計數器的值為 258
*** pool-1-thread-7 獲取共享計數器的值為 696
*** pool-1-thread-1 獲取共享計數器的值為 258
*** pool-1-thread-12 獲取共享計數器的值為 258
pool-1-thread-14 修改共享計數器的值為 426
*** pool-1-thread-8 獲取共享計數器的值為 426
pool-1-thread-9 修改共享計數器的值為 704
pool-1-thread-5 修改共享計數器的值為 187
pool-1-thread-13 修改共享計數器的值為 662
pool-1-thread-4 修改共享計數器的值為 390
pool-1-thread-15 修改共享計數器的值為 702
pool-1-thread-11 修改共享計數器的值為 65
pool-1-thread-6 修改共享計數器的值為 84
pool-1-thread-10 修改共享計數器的值為 93
*** pool-1-thread-14 獲取共享計數器的值為 93
*** pool-1-thread-9 獲取共享計數器的值為 93
*** pool-1-thread-5 獲取共享計數器的值為 93
*** pool-1-thread-13 獲取共享計數器的值為 93
*** pool-1-thread-4 獲取共享計數器的值為 93
*** pool-1-thread-15 獲取共享計數器的值為 93
*** pool-1-thread-11 獲取共享計數器的值為 93
pool-1-thread-3 修改共享計數器的值為 276
*** pool-1-thread-6 獲取共享計數器的值為 276
*** pool-1-thread-10 獲取共享計數器的值為 276
*** pool-1-thread-3 獲取共享計數器的值為 276
輸出也是符合預期的，也許有人會認為下面這段輸出是不符合預期的，但博主認為是符合預期的，因為博主這里的ZooKeeper服務端是單機模式，因此不用考慮集群的一致性問題，如果值在服務端修改成功，讀取肯定是最新的值，而造成這種情況的原因，可能是CPU時間片輪詢調度的原因（15個執行緒全部在一臺計算機上運行），也有可能是列印字串的相關函式獲取I/O設備存在快慢的原因，博主覺得是此類原因導致的，如果ZooKeeper服務端是集群模式，倒是有可能是集群的一致性問題導致的，
pool-1-thread-12 修改共享計數器的值為 696
pool-1-thread-8 修改共享計數器的值為 258
*** pool-1-thread-7 獲取共享計數器的值為 696
*** pool-1-thread-1 獲取共享計數器的值為 258
*** pool-1-thread-12 獲取共享計數器的值為 258
分布式環境是非常復雜的，獲取VersionedValue實體然后立即用于修改共享整數的值也是有可能不成功的，因為獲取的VersionedValue實體可能瞬間就失效了，比如有其他的節點在這個時間間隙中修改了共享整數的值，
// 獲取VersionedValue實體用于修改共享整數的值
sharedCount.trySetCount(sharedCount.getVersionedValue(), randomValue);
監聽器
public class SharedCountRunnable implements Runnable{
// 是否是監聽節點
private final boolean isListenerNode;
public SharedCountRunnable(boolean isListenerNode) {
this.isListenerNode = isListenerNode;
}
@SneakyThrows
@Override
public void run() {
// 使用不同的CuratorFramework實體，表示不同的分布式節點
CuratorFramework curator = CuratorFrameworkProperties.getCuratorFramework();
// 模擬隨機加入的分布式節點
int randomSleep = new Random().nextInt(100);
// 讓監聽節點先添加監聽器
if(!isListenerNode) {
Thread.sleep(3000 + randomSleep);
}
// 共享計數器的路徑
String path = "/kaven";
int seedValue = 0;
// 創建SharedCount實體，用于提供共享計數器功能
// 如果共享計數器的路徑尚未創建，則使用seedValue的值作為共享整數的初始值
SharedCount sharedCount = new SharedCount(curator, path, seedValue);
// 必須先啟動共享計數器，然后才能使用它
sharedCount.start();
if(isListenerNode) {
// 監聽節點添加監聽器
sharedCount.addListener(new SharedCountListener(){
@Override
public void stateChanged(CuratorFramework client, ConnectionState newState) {}
@Override
public void countHasChanged(SharedCountReader sharedCount, int newCount) throws Exception {
System.out.println("[Listener] : 共享整數被修改成 " + newCount);
}
});
System.out.println("監聽器已添加");
Thread.sleep(1000000);
}
else {
update(sharedCount);
}
// 使用完共享計數器后需要呼叫close()進行關閉
sharedCount.close();
}
private void update(SharedCount sharedCount) throws Exception {
// 使用亂數來設定共享整數的值
int randomValue = new Random().nextInt(1000);
// 獲取VersionedValue實體用于修改共享整數的值
sharedCount.trySetCount(sharedCount.getVersionedValue(), randomValue);
System.out.println(Thread.currentThread().getName() + " 修改共享計數器的值為 " + randomValue);
}
}
public class Application {
private static final ExecutorService EXECUTOR_SERVICE = Executors.newCachedThreadPool();
public static void main(String[] args) throws Exception {
// 分布式節點處理業務
for (int i = 0; i < 30; i++) {
// 設定一個監聽節點，其他節點用于修改共享整數的值
if(i > 0) EXECUTOR_SERVICE.execute(new SharedCountRunnable(false));
else EXECUTOR_SERVICE.execute(new SharedCountRunnable(true));
}
}
}
輸出如下所示：
監聽器已添加
pool-1-thread-29 修改共享計數器的值為 178
pool-1-thread-22 修改共享計數器的值為 526
[Listener] : 共享整數被修改成 526
pool-1-thread-26 修改共享計數器的值為 76
pool-1-thread-14 修改共享計數器的值為 27
pool-1-thread-4 修改共享計數器的值為 907
[Listener] : 共享整數被修改成 907
pool-1-thread-16 修改共享計數器的值為 616
pool-1-thread-18 修改共享計數器的值為 21
pool-1-thread-13 修改共享計數器的值為 851
pool-1-thread-15 修改共享計數器的值為 569
[Listener] : 共享整數被修改成 569
pool-1-thread-20 修改共享計數器的值為 806
pool-1-thread-21 修改共享計數器的值為 961
pool-1-thread-25 修改共享計數器的值為 562
pool-1-thread-19 修改共享計數器的值為 99
pool-1-thread-6 修改共享計數器的值為 784
pool-1-thread-11 修改共享計數器的值為 202
pool-1-thread-24 修改共享計數器的值為 97
[Listener] : 共享整數被修改成 202
pool-1-thread-17 修改共享計數器的值為 139
pool-1-thread-12 修改共享計數器的值為 437
[Listener] : 共享整數被修改成 437
pool-1-thread-7 修改共享計數器的值為 156
pool-1-thread-8 修改共享計數器的值為 416
pool-1-thread-28 修改共享計數器的值為 930
pool-1-thread-10 修改共享計數器的值為 857
pool-1-thread-23 修改共享計數器的值為 945
pool-1-thread-5 修改共享計數器的值為 38
pool-1-thread-27 修改共享計數器的值為 422
[Listener] : 共享整數被修改成 422
pool-1-thread-9 修改共享計數器的值為 428
pool-1-thread-2 修改共享計數器的值為 413
pool-1-thread-30 修改共享計數器的值為 59
pool-1-thread-3 修改共享計數器的值為 539
可見有些修改操作是不成功的，因為沒有被監聽到，
Curator框架的共享計數器SharedCount就介紹到這里，如果博主有說錯的地方或者大家有不同的見解，歡迎大家評論補充，
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/415132.html
標籤：其他