zookeeper系列（五）實戰分佈式鎖
分佈式鎖
咱們常說的鎖是單進程多線程鎖，在多線程併發編程中，用於線程之間的數據同步，保護共享資源的訪問。而分佈式鎖，指在分佈式環境下，保護跨進程、跨主機、跨網絡的共享資源，實現互斥訪問，保證一致性。node
架構圖
左側是zookeeper集羣，locker是數據節點，node_1到node_n表明一系列的順序節點。編程
右側client_1至client_n表明客戶端，Service表明須要互斥訪問的服務。網絡
總實現思路，是在獲取鎖的時候在locker節點下建立順序節點，在釋放鎖的時候，把本身建立的節點刪除。session
流程圖
類圖
代碼實現
public interface DistributedLock {
/*
* 獲取鎖，若是沒有獲得就等待
*/
public void acquire() throws Exception;
/*
* 獲取鎖，直到超時
*/
public boolean acquire(long time, TimeUnit unit) throws Exception;
/*
* 釋放鎖
*/
public void release() throws Exception;
}
public class SimpleDistributedLockMutex extends BaseDistributedLock implements
DistributedLock {
//鎖名稱前綴,成功建立的順序節點如lock-0000000000,lock-0000000001,...
private static final String LOCK_NAME = "lock-";
// zookeeper中locker節點的路徑
private final String basePath;
// 獲取鎖之後本身建立的那個順序節點的路徑
private String ourLockPath;
private boolean internalLock(long time, TimeUnit unit) throws Exception {
ourLockPath = attemptLock(time, unit);
return ourLockPath != null;
}
public SimpleDistributedLockMutex(ZkClientExt client, String basePath){
super(client,basePath,LOCK_NAME);
this.basePath = basePath;
}
// 獲取鎖
public void acquire() throws Exception {
if ( !internalLock(-1, null) ) {
throw new IOException("鏈接丟失!在路徑:'"+basePath+"'下不能獲取鎖!");
}
}
// 獲取鎖，能夠超時
public boolean acquire(long time, TimeUnit unit) throws Exception {
return internalLock(time, unit);
}
// 釋放鎖
public void release() throws Exception {
releaseLock(ourLockPath);
}
}
public class BaseDistributedLock {
private final ZkClientExt client;
private final String path;
private final String basePath;
private final String lockName;
private static final Integer MAX_RETRY_COUNT = 10;
public BaseDistributedLock(ZkClientExt client, String path, String lockName){
this.client = client;
this.basePath = path;
this.path = path.concat("/").concat(lockName);
this.lockName = lockName;
}
// 刪除成功獲取鎖以後所建立的那個順序節點
private void deleteOurPath(String ourPath) throws Exception{
client.delete(ourPath);
}
// 建立臨時順序節點
private String createLockNode(ZkClient client, String path) throws Exception{
return client.createEphemeralSequential(path, null);
}
// 等待比本身次小的順序節點的刪除
private boolean waitToLock(long startMillis, Long millisToWait, String ourPath) throws Exception{
boolean haveTheLock = false;
boolean doDelete = false;
try {
while ( !haveTheLock ) {
// 獲取/locker下的通過排序的子節點列表
List<String> children = getSortedChildren();
// 獲取剛纔本身建立的那個順序節點名
String sequenceNodeName = ourPath.substring(basePath.length()+1);
// 判斷本身排第幾個
int ourIndex = children.indexOf(sequenceNodeName);
if (ourIndex < 0){ // 網絡抖動，獲取到的子節點列表裏可能已經沒有本身了
throw new ZkNoNodeException("節點沒有找到: " + sequenceNodeName);
}
// 若是是第一個，表明本身已經得到了鎖
boolean isGetTheLock = ourIndex == 0;
// 若是本身沒有得到鎖，則要watch比咱們次小的那個節點
String pathToWatch = isGetTheLock ? null : children.get(ourIndex - 1);
if ( isGetTheLock ){
haveTheLock = true;
} else {
// 訂閱比本身次小順序節點的刪除事件
String previousSequencePath = basePath .concat( "/" ) .concat( pathToWatch );
final CountDownLatch latch = new CountDownLatch(1);
final IZkDataListener previousListener = new IZkDataListener() {
public void handleDataDeleted(String dataPath) throws Exception {
latch.countDown(); // 刪除後結束latch上的await
}
public void handleDataChange(String dataPath, Object data) throws Exception {
// ignore
}
};
try {
//訂閱次小順序節點的刪除事件，若是節點不存在會出現異常
client.subscribeDataChanges(previousSequencePath, previousListener);
if ( millisToWait != null ) {
millisToWait -= (System.currentTimeMillis() - startMillis);
startMillis = System.currentTimeMillis();
if ( millisToWait <= 0 ) {
doDelete = true; // timed out - delete our node
break;
}
latch.await(millisToWait, TimeUnit.MICROSECONDS); // 在latch上await
} else {
latch.await(); // 在latch上await
}
// 結束latch上的等待後，繼續while從新來過判斷本身是否第一個順序節點
}
catch ( ZkNoNodeException e ) {
//ignore
} finally {
client.unsubscribeDataChanges(previousSequencePath, previousListener);
}
}
}
}
catch ( Exception e ) {
//發生異常須要刪除節點
doDelete = true;
throw e;
} finally {
//若是須要刪除節點
if ( doDelete ) {
deleteOurPath(ourPath);
}
}
return haveTheLock;
}
private String getLockNodeNumber(String str, String lockName) {
int index = str.lastIndexOf(lockName);
if ( index >= 0 ) {
index += lockName.length();
return index <= str.length() ? str.substring(index) : "";
}
return str;
}
// 獲取/locker下的通過排序的子節點列表
List<String> getSortedChildren() throws Exception {
try{
List<String> children = client.getChildren(basePath);
Collections.sort(
children, new Comparator<String>() {
public int compare(String lhs, String rhs) {
return getLockNodeNumber(lhs, lockName).compareTo(getLockNodeNumber(rhs, lockName));
}
}
);
return children;
} catch (ZkNoNodeException e){
client.createPersistent(basePath, true);
return getSortedChildren();
}
}
protected void releaseLock(String lockPath) throws Exception{
deleteOurPath(lockPath);
}
protected String attemptLock(long time, TimeUnit unit) throws Exception {
final long startMillis = System.currentTimeMillis();
final Long millisToWait = (unit != null) ? unit.toMillis(time) : null;
String ourPath = null;
boolean hasTheLock = false;
boolean isDone = false;
int retryCount = 0;
//網絡閃斷須要重試一試
while ( !isDone ) {
isDone = true;
try {
// 在/locker下建立臨時的順序節點
ourPath = createLockNode(client, path);
// 判斷本身是否得到了鎖，若是沒有得到那麼等待直到得到鎖或者超時
hasTheLock = waitToLock(startMillis, millisToWait, ourPath);
} catch ( ZkNoNodeException e ) { // 捕獲這個異常
if ( retryCount++ < MAX_RETRY_COUNT ) { // 重試指定次數
isDone = false;
} else {
throw e;
}
}
}
if ( hasTheLock ) {
return ourPath;
}
return null;
}
}
public class TestDistributedLock {
public static void main(String[] args) {
final ZkClientExt zkClientExt1 = new ZkClientExt("192.168.1.105:2181", 5000, 5000, new BytesPushThroughSerializer());
final SimpleDistributedLockMutex mutex1 = new SimpleDistributedLockMutex(zkClientExt1, "/Mutex");
final ZkClientExt zkClientExt2 = new ZkClientExt("192.168.1.105:2181", 5000, 5000, new BytesPushThroughSerializer());
final SimpleDistributedLockMutex mutex2 = new SimpleDistributedLockMutex(zkClientExt2, "/Mutex");
try {
mutex1.acquire();
System.out.println("Client1 locked");
Thread client2Thd = new Thread(new Runnable() {
public void run() {
try {
mutex2.acquire();
System.out.println("Client2 locked");
mutex2.release();
System.out.println("Client2 released lock");
} catch (Exception e) {
e.printStackTrace();
}
}
});
client2Thd.start();
Thread.sleep(5000);
mutex1.release();
System.out.println("Client1 released lock");
client2Thd.join();
} catch (Exception e) {
e.printStackTrace();
}
}
}
public class ZkClientExt extends ZkClient {
public ZkClientExt(String zkServers, int sessionTimeout, int connectionTimeout, ZkSerializer zkSerializer) {
super(zkServers, sessionTimeout, connectionTimeout, zkSerializer);
}
@Override
public void watchForData(final String path) {
retryUntilConnected(new Callable<Object>() {
public Object call() throws Exception {
Stat stat = new Stat();
_connection.readData(path, stat, true);
return null;
}
});
}
}
相關文章
- 1. zookeeper系列（五）實戰分佈式鎖
- 2. Zookeeper基礎+Zookeeper實戰分佈式鎖
- 3. zookeeper實戰：自定義分佈式鎖
- 4. 【Zookeeper系列四】ZooKeeper 分佈式鎖實現
- 5. Zookeeper實現分佈式鎖
- 6. zookeeper實現分佈式鎖
- 7. 分佈式鎖（Zookeeper實現）
- 8. zookeeper分佈式鎖實現
- 9. zookeeper 實現分佈式鎖
- 10. ZooKeeper實現分佈式鎖
- 更多相關文章...
- • Redis發佈訂閱模式 - Redis教程
- • ASP.NET MVC - 樣式和佈局 - ASP.NET 教程
- • Git五分鐘教程
- • 再有人問你分佈式事務，把這篇扔給他
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
- 1. 融合阿里雲，牛客助您找到心儀好工作
- 2. 解決jdbc(jdbctemplate)在測試類時不報錯在TomCatb部署後報錯
- 3. 解決PyCharm GoLand IntelliJ 等 JetBrains 系列 IDE無法輸入中文
- 4. vue+ant design中關於圖片請求不顯示的問題。
- 5. insufficient memory && Native memory allocation (malloc) failed
- 6. 解決IDEA用Maven創建的Web工程不能創建Java Class文件的問題
- 7. [已解決] Error: Cannot download ‘https://start.spring.io/starter.zip?
- 8. 在idea讓java文件夾正常使用
- 9. Eclipse啓動提示「subversive connector discovery」
- 10. 帥某-技巧-快速轉帖博主文章（article_content）
歡迎關注本站公眾號,獲取更多信息
