快捷導(dǎo)航

Java 中通過 key 獲取鎖的方法

更新時(shí)間：2022年11月15日 09:33:08 作者：明明如月學(xué)長

這篇文章主要介紹了Java 中通過 key 獲取鎖,本文演示如何對某個(gè) key 加鎖，以保證對該 key 的并發(fā)操作限制，可以實(shí)現(xiàn)同一個(gè) key 一個(gè)或者多個(gè)線程同時(shí)執(zhí)行，需要的朋友可以參考下

一、概覽

本文我們將了解如何通過特定鍵獲取鎖，以保證該鍵上的操作的線程安全，并且不妨礙其他鍵。
一般來說，我們需要實(shí)現(xiàn)兩個(gè)方法：

void lock(String key)
void unlock(String key)

本文以字符串作為鍵為例，大家可以根據(jù)實(shí)際需要改造成任意類型的鍵，重寫 equas 和 hashCode 方法，保證唯一性即可。

二、簡單的互斥鎖

假設(shè)需要滿足當(dāng)前線程獲取鎖則需要執(zhí)行特定代碼，否則不執(zhí)行這個(gè)場景。
我們可以維護(hù)一系列 Key 的 Set, 在使用時(shí)添加到 Set 中，解鎖時(shí)移除對應(yīng)的 Key。
此時(shí)，需要考慮線程安全問題。因此需要使用線程安全的 Set 實(shí)現(xiàn)，如基于 ConcurrentHashMap 的線程安全 Set。

public class SimpleExclusiveLockByKey {

    private static Set<String> usedKeys= ConcurrentHashMap.newKeySet();
    
    public boolean tryLock(String key) {
        return usedKeys.add(key);
    }
    
    public void unlock(String key) {
        usedKeys.remove(key);
    }

}

使用案例：

String key = "key";
SimpleExclusiveLockByKey lockByKey = new SimpleExclusiveLockByKey();
try {
    lockByKey.tryLock(key);
    // 在這里添加對該 key 獲取鎖之后要執(zhí)行的代碼
} finally { // 非常關(guān)鍵
    lockByKey.unlock(key);
}

注意一定要在 finally 代碼塊中解鎖，以保證即便發(fā)生異常時(shí)，也可以正常解鎖。

三、按鍵來獲取和釋放鎖

以上代碼可以保證獲取鎖后才執(zhí)行，但無法實(shí)現(xiàn)未拿到鎖的線程等待的效果。
有時(shí)候，我們需要讓未獲取到對應(yīng)鎖的線程等待。
流程如下：

第一個(gè)線程獲取某個(gè) key 的鎖
第二個(gè)線程獲取同一個(gè) key 的鎖，第二個(gè)線程需要等待
第一個(gè)線程釋放某個(gè) key 的鎖
第二個(gè)線程獲取該 key 的鎖，然后執(zhí)行其代碼

3.1 使用線程計(jì)數(shù)器定義 Lock

我們可以使用 ReentrantLock 來實(shí)行線程阻塞。
我們通過內(nèi)部類來封裝 Lock。該類統(tǒng)計(jì)某個(gè) key 上執(zhí)行的線程數(shù)。暴露兩個(gè)方法，一個(gè)是線程數(shù)增加，一個(gè)是減少線程數(shù)。

private static class LockWrapper {
    private final Lock lock = new ReentrantLock();
    private final AtomicInteger numberOfThreadsInQueue = new AtomicInteger(1);

    private LockWrapper addThreadInQueue() {
        numberOfThreadsInQueue.incrementAndGet(); 
        return this;
    }

    private int removeThreadFromQueue() {
        return numberOfThreadsInQueue.decrementAndGet(); 
    }

}

3.2 處理排隊(duì)的線程

接下來繼續(xù)使用 ConcurrentHashMap ， key 作為鍵， LockWrapper 作為值。
保證同一個(gè) key 使用同一個(gè) LockWrapper 中的同一把鎖。

private static ConcurrentHashMap<String, LockWrapper> locks = new ConcurrentHashMap<String, LockWrapper>();

一個(gè)線程想要獲取某個(gè) key 的鎖時(shí)，需要看該 key 對應(yīng)的 LockWrapper 是否已經(jīng)存在。

如果不存在，創(chuàng)建一個(gè) LockWrapper ，計(jì)數(shù)器設(shè)置為1
如果存在，對應(yīng)的 LockWrapper 加1

public void lock(String key) {
    LockWrapper lockWrapper = locks.compute(key, (k, v) -> v == null ? new LockWrapper() : v.addThreadInQueue());
    lockWrapper.lock.lock();
}

3.3 解鎖和移除 Entry

解鎖時(shí)將等待的隊(duì)列減一。
當(dāng)前 key 對應(yīng)的線程數(shù)為 0 時(shí)，可以將其從 ConcurrentHashMap 中移除。

public void unlock(String key) {
    LockWrapper lockWrapper = locks.get(key);
    lockWrapper.lock.unlock();
    if (lockWrapper.removeThreadFromQueue() == 0) { 
        // NB : We pass in the specific value to remove to handle the case where another thread would queue right before the removal
        locks.remove(key, lockWrapper);
    }
}

3.4 總結(jié)

最終效果如下：

public class LockByKey {
    
    private static class LockWrapper {
        private final Lock lock = new ReentrantLock();
        private final AtomicInteger numberOfThreadsInQueue = new AtomicInteger(1);
        
        private LockWrapper addThreadInQueue() {
            numberOfThreadsInQueue.incrementAndGet(); 
            return this;
        }
        
        private int removeThreadFromQueue() {
            return numberOfThreadsInQueue.decrementAndGet(); 
        }
        
    }
    
    private static ConcurrentHashMap<String, LockWrapper> locks = new ConcurrentHashMap<String, LockWrapper>();
    
    public void lock(String key) {
        LockWrapper lockWrapper = locks.compute(key, (k, v) -> v == null ? new LockWrapper() : v.addThreadInQueue());
        lockWrapper.lock.lock();
    }
    
    public void unlock(String key) {
        LockWrapper lockWrapper = locks.get(key);
        lockWrapper.lock.unlock();
        if (lockWrapper.removeThreadFromQueue() == 0) { 
            // NB : We pass in the specific value to remove to handle the case where another thread would queue right before the removal
            locks.remove(key, lockWrapper);
        }
    }
}

使用示例：

String key = "key"; 
LockByKey lockByKey = new LockByKey(); 
try { 
    lockByKey.lock(key);
    // insert your code here 
} finally { // CRUCIAL 
    lockByKey.unlock(key); 
}

四、允許同一個(gè) key 同時(shí)多個(gè)線程運(yùn)行

我們還需要考慮另外一種場景：前面對于同一個(gè) key 同一時(shí)刻只允許一個(gè)線程執(zhí)行。如果我們想實(shí)現(xiàn)，對于同一個(gè) key ，允許同時(shí)運(yùn)行 n 個(gè)線程該怎么辦？
為了方便理解，我們假設(shè)同一個(gè) key 允許兩個(gè)線程。

第一個(gè)線程想要獲取某個(gè) key 的鎖，允許
第二個(gè)線程也想要獲取該 key 的鎖，允許
第三個(gè)線程也想獲取該 key 的鎖，該線程需要等待第一個(gè)或第二個(gè)線程釋放鎖之后才可以執(zhí)行

Semaphore 很適合這種場景。Semaphore 可以控制同時(shí)運(yùn)行的線程數(shù)。

public class SimultaneousEntriesLockByKey {

    private static final int ALLOWED_THREADS = 2;
    
    private static ConcurrentHashMap<String, Semaphore> semaphores = new ConcurrentHashMap<String, Semaphore>();
    
    public void lock(String key) {
        Semaphore semaphore = semaphores.compute(key, (k, v) -> v == null ? new Semaphore(ALLOWED_THREADS) : v);
        semaphore.acquireUninterruptibly();
    }
    
    public void unlock(String key) {
        Semaphore semaphore = semaphores.get(key);
        semaphore.release();
        if (semaphore.availablePermits() == ALLOWED_THREADS) { 
            semaphores.remove(key, semaphore);
        }  
    }
    
}

使用案例：

String key = "key"; 
SimultaneousEntriesLockByKey lockByKey = new SimultaneousEntriesLockByKey(); 
try { 
    lockByKey.lock(key); 
    // 在這里添加對該 key 獲取鎖之后要執(zhí)行的代碼
} finally { // 非常關(guān)鍵
    lockByKey.unlock(key); 
}

五、結(jié)論

本文演示如何對某個(gè) key 加鎖，以保證對該 key 的并發(fā)操作限制，可以實(shí)現(xiàn)同一個(gè) key 一個(gè)或者多個(gè)線程同時(shí)執(zhí)行。
相關(guān)代碼：https://github.com/eugenp/tutorials/tree/master/core-java-modules/core-java-concurrency-advanced-4

原文：https://www.baeldung.com/java-acquire-lock-by-key

到此這篇關(guān)于Java 中通過 key 獲取鎖的文章就介紹到這了,更多相關(guān)java 獲取鎖內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: