多線程訪問(wèn)同一個(gè)共享資源時(shí)，會(huì)出現(xiàn)并發(fā)問(wèn)題，synchronized或者lock 類的鎖只能控制單一進(jìn)程的資源訪問(wèn)，多進(jìn)程下就需要用到分布式鎖

利用zk 可以實(shí)現(xiàn)獨(dú)占鎖，(同級(jí)節(jié)點(diǎn)唯一性)多個(gè)進(jìn)程往zk指定節(jié)點(diǎn)下創(chuàng)建一個(gè)相同名稱的節(jié)點(diǎn)，只有一個(gè)能成功，創(chuàng)建失敗的通過(guò)zk的watcher機(jī)制監(jiān)聽(tīng)子節(jié)點(diǎn)變化，一個(gè)監(jiān)聽(tīng)到子節(jié)點(diǎn)刪除事件，會(huì)再次觸發(fā)所有進(jìn)程的寫鎖，但這里會(huì)有驚群效應(yīng)，會(huì)影響到性能

利用有序節(jié)點(diǎn)實(shí)現(xiàn)分布式鎖：每個(gè)客戶端都往一個(gè)指定節(jié)點(diǎn)(locks)注冊(cè)一個(gè)臨時(shí)有序節(jié)點(diǎn)，越早創(chuàng)建的節(jié)點(diǎn)編號(hào)越小，最小編號(hào)的節(jié)點(diǎn)獲得鎖，通過(guò)監(jiān)聽(tīng)比自己小的節(jié)點(diǎn)，當(dāng)比自己小的節(jié)點(diǎn)刪除后，客戶端會(huì)收到watcher，再次判斷自己的節(jié)點(diǎn)是不是所有節(jié)點(diǎn)最小的，是則獲得鎖；這種方式也會(huì)解決驚群?jiǎn)栴}

接下來(lái)我們來(lái)看實(shí)現(xiàn)：curator 分布式鎖的使用
curator對(duì)鎖封裝了一層，提供了InterProcessMutex；還提供了leader 選舉、分布式隊(duì)列 InterProcessMutex 分布式可重入排他鎖
InterProcessSemaphoreMutex 分布式排他鎖
InterProcessReadWriteLock 分布式讀寫鎖

public class LeaderSelectorClient extends LeaderSelectorListenerAdapter implements Closeable {private String name; //表示當(dāng)前的進(jìn)程private LeaderSelector leaderSelector; //leader選舉的APIprivate CountDownLatch countDownLatch=new CountDownLatch(1);public LeaderSelectorClient(){}public LeaderSelectorClient(String name) {this.name = name;}public LeaderSelector getLeaderSelector() {return leaderSelector;}public void setLeaderSelector(LeaderSelector leaderSelector) {this.leaderSelector = leaderSelector;}public void start(){leaderSelector.start(); //開(kāi)始競(jìng)爭(zhēng)leader}@Overridepublic void takeLeadership(CuratorFramework client) throws Exception {//如果進(jìn)入當(dāng)前的方法，意味著當(dāng)前的進(jìn)程獲得了鎖。獲得鎖以后，這個(gè)方法會(huì)被回調(diào)//這個(gè)方法執(zhí)行結(jié)束之后，表示釋放leader權(quán)限System.out.println(name+"->現(xiàn)在是leader了"); // countDownLatch.await(); //阻塞當(dāng)前的進(jìn)程防止leader丟失}@Overridepublic void close() throws IOException {leaderSelector.close();}private static String CONNECTION_STR="zk集群地址，至少三臺(tái)機(jī)器";public static void main(String[] args) throws IOException {CuratorFramework curatorFramework = CuratorFrameworkFactory.builder().connectString(CONNECTION_STR).sessionTimeoutMs(50000000).retryPolicy(new ExponentialBackoffRetry(1000, 3)).build();curatorFramework.start();LeaderSelectorClient leaderSelectorClient=new LeaderSelectorClient("ClientA");LeaderSelector leaderSelector=new LeaderSelector(curatorFramework,"/leader",leaderSelectorClient);leaderSelectorClient.setLeaderSelector(leaderSelector);leaderSelectorClient.start(); //開(kāi)始選舉System.in.read();} } 我們來(lái)看下curator 實(shí)現(xiàn)分布式鎖的原理，這里我把注釋寫在了代碼中，所以把代碼貼到一塊public InterProcessMutex(CuratorFramework client, String path) {// 實(shí)現(xiàn)公平鎖的核心：zookeeper利用path 創(chuàng)建臨時(shí)有序節(jié)點(diǎn)this(client, path, new StandardLockInternalsDriver());}public StandardLockInternalsDriver() {}public InterProcessMutex(CuratorFramework client, String path, LockInternalsDriver driver) {this(client, path, "lock-", 1, driver);}//maxLeases：互斥鎖InterProcessMutex(CuratorFramework client, String path, String lockName, int maxLeases, LockInternalsDriver driver) {this.threadData = Maps.newConcurrentMap();this.basePath = PathUtils.validatePath(path);//InterProcessMutex 把分布式鎖的申請(qǐng)和釋放委托給了 LockInternals internalsthis.internals = new LockInternals(client, driver, path, lockName, maxLeases);}// 無(wú)限等待public void acquire() throws Exception {if (!this.internalLock(-1L, (TimeUnit)null)) {throw new IOException("Lost connection while trying to acquire lock: " + this.basePath);}}// 限時(shí)等待public boolean acquire(long time, TimeUnit unit) throws Exception {return this.internalLock(time, unit);}private boolean internalLock(long time, TimeUnit unit) throws Exception {Thread currentThread = Thread.currentThread();//同一線程再次acquire，首先判斷threaData 是否有這個(gè)線程鎖信息，如果有則原子+1，然后返回InterProcessMutex.LockData lockData = (InterProcessMutex.LockData)this.threadData.get(currentThread);if (lockData != null) {// 實(shí)現(xiàn)可重入；lockData.lockCount.incrementAndGet();return true;// 映射表沒(méi)有對(duì)應(yīng)的鎖信息，嘗試通過(guò)LockInternals 獲取鎖} else {String lockPath = this.internals.attemptLock(time, unit, this.getLockNodeBytes());if (lockPath != null) {// 成功獲取鎖，存儲(chǔ)到映射表中InterProcessMutex.LockData newLockData = new InterProcessMutex.LockData(currentThread, lockPath);this.threadData.put(currentThread, newLockData);return true;} else {return false;}}}// 記錄線程和鎖信息的映射關(guān)系private final ConcurrentMap<Thread, InterProcessMutex.LockData> threadData;//zk 中一個(gè)臨時(shí)有序節(jié)點(diǎn)對(duì)應(yīng)一個(gè)鎖，但是讓鎖生效需要排隊(duì)private static class LockData {final Thread owningThread;final String lockPath;final AtomicInteger lockCount;private LockData(Thread owningThread, String lockPath) {this.lockCount = new AtomicInteger(1); //分布式鎖重入次數(shù)this.owningThread = owningThread;this.lockPath = lockPath;}}//嘗試獲取鎖，并返回鎖對(duì)應(yīng)的zk 臨時(shí)有序節(jié)點(diǎn)路徑String attemptLock(long time, TimeUnit unit, byte[] lockNodeBytes) throws Exception {long startMillis = System.currentTimeMillis();// millisToWait 是個(gè)nullLong millisToWait = unit != null ? unit.toMillis(time) : null;byte[] localLockNodeBytes = this.revocable.get() != null ? new byte[0] : lockNodeBytes;int retryCount = 0;String ourPath = null;//是否已經(jīng)有分布式鎖？boolean hasTheLock = false;//是否已經(jīng)完成嘗試獲取分布式鎖操作boolean isDone = false;while(!isDone) {isDone = true;try {//driver = StandardLockInternalsDriverourPath = this.driver.createsTheLock(this.client, this.path, localLockNodeBytes);//循環(huán)等待激活分布式鎖hasTheLock = this.internalLockLoop(startMillis, millisToWait, ourPath);} catch (NoNodeException var14) {if (!this.client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++, System.currentTimeMillis() - startMillis, RetryLoop.getDefaultRetrySleeper())) {throw var14;}isDone = false;}}//成功獲取分布式鎖，返回有序節(jié)點(diǎn)的路徑return hasTheLock ? ourPath : null;}//在zk 中創(chuàng)建臨時(shí)順序節(jié)點(diǎn)public String createsTheLock(CuratorFramework client, String path, byte[] lockNodeBytes) throws Exception {String ourPath;//默認(rèn)內(nèi)容是ip地址if (lockNodeBytes != null) {// creatingParentContainersIfNeeded：創(chuàng)建父節(jié)點(diǎn)，如果不支持CreateMode.CONTAINER,就采用CreateMode.PERSISTENT// withProtection：臨時(shí)節(jié)點(diǎn)添加GUIDourPath = (String)((ACLBackgroundPathAndBytesable)client.create().creatingParentContainersIfNeeded().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL)).forPath(path, lockNodeBytes);} else {ourPath = (String)((ACLBackgroundPathAndBytesable)client.create().creatingParentContainersIfNeeded().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL)).forPath(path);}return ourPath;}//循環(huán)等待激活分布式鎖private boolean internalLockLoop(long startMillis, Long millisToWait, String ourPath) throws Exception {boolean haveTheLock = false;boolean doDelete = false;try {if (this.revocable.get() != null) {((BackgroundPathable)this.client.getData().usingWatcher(this.revocableWatcher)).forPath(ourPath);}while(this.client.getState() == CuratorFrameworkState.STARTED && !haveTheLock) {List<String> children = this.getSortedChildren();String sequenceNodeName = ourPath.substring(this.basePath.length() + 1);PredicateResults predicateResults = this.driver.getsTheLock(this.client, children, sequenceNodeName, this.maxLeases);if (predicateResults.getsTheLock()) {haveTheLock = true;} else {String previousSequencePath = this.basePath + "/" + predicateResults.getPathToWatch();synchronized(this) {try {((BackgroundPathable)this.client.getData().usingWatcher(this.watcher)).forPath(previousSequencePath);if (millisToWait == null) {this.wait();} else {millisToWait = millisToWait - (System.currentTimeMillis() - startMillis);startMillis = System.currentTimeMillis();if (millisToWait > 0L) {this.wait(millisToWait);} else {doDelete = true;break;}}} catch (NoNodeException var19) {}}}}} catch (Exception var21) {ThreadUtils.checkInterrupted(var21);doDelete = true;throw var21;} finally {if (doDelete) {this.deleteOurPath(ourPath);}}return haveTheLock;} 《新程序員》：云原生和全面數(shù)字化實(shí)踐50位技術(shù)專家共同創(chuàng)作，文字、視頻、音頻交互閱讀

總結(jié)

以上是生活随笔為你收集整理的分布式锁-zk临时节点的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。