摘要：隊(duì)列鎖就是將線程組織成一個(gè)隊(duì)列，讓每個(gè)線程在不同的存儲(chǔ)單元上旋轉(zhuǎn)，從而降低一致性流量。隊(duì)列鎖隊(duì)列鎖表示為對(duì)象的鏈表，每個(gè)線程通過一個(gè)線程局部變量指向其前驅(qū)。

編寫高效的并發(fā)程序，需要對(duì)互斥問題重新研究，設(shè)計(jì)出適用于多線程的互斥協(xié)議。那么問題來了，如果不能獲得鎖，應(yīng)該怎么做？

旋轉(zhuǎn)：繼續(xù)進(jìn)行嘗試，如自旋鎖，延遲較短；

阻塞：掛起自己，請(qǐng)求調(diào)度器切換到另一個(gè)線程，代價(jià)較大。

綜合來看，先旋轉(zhuǎn)一小段時(shí)間再阻塞，是種不錯(cuò)的選擇。

java.util.concurrent.locks.Lock接口提供了lock()和unlock()兩個(gè)重要的方法，用于解決實(shí)際互斥問題。

Lock mutex = new MyLock();
mutex.lock();
try {
    do something
} finally {
    mutex.unlock();
}

測試-設(shè)置來源于getAndSet()操作，通過一個(gè)原子布爾型狀態(tài)變量的值判斷當(dāng)前鎖的狀態(tài)。若為true表示鎖忙，若為false表示鎖空閑。

public class TASLock {
    AtomicBoolean state = new AtomicBoolean(false);

    public void lock() {
        while (state.getAndSet(true)) {
            ;
        }
    }

    public void unlock() {
        state.set(false);
    }
}

TTASLock是升級(jí)版的TASLock算法，沒有直接調(diào)用getAndSet()方法，而是在鎖看起來空閑（state.get()返回false）時(shí)才調(diào)用。

public class TTASLock {
    AtomicBoolean state = new AtomicBoolean(false);

    public void lock() {
        while (true) {
            while (state.get()) {
                ;
            }
            if (! state.getAndSet(true)) {
                return;
            }
        }
    }

    public void unlock() {
        state.set(false);
    }
}

這兩個(gè)算法都能保證無死鎖的互斥，但是TTASLock的性能會(huì)比TASLock高許多。

可以從計(jì)算機(jī)系統(tǒng)結(jié)構(gòu)的高速緩存和局部性來解釋這個(gè)問題，每個(gè)處理器都有一個(gè)cache，cache的訪問速度比內(nèi)存快好幾個(gè)數(shù)量級(jí)。當(dāng)cache命中時(shí)，會(huì)立即加載這個(gè)值；當(dāng)cache缺失時(shí)，會(huì)在內(nèi)存或兩一個(gè)處理器的cache中尋找這個(gè)數(shù)據(jù)。尋找的過程比較漫長，處理器在總線上廣播這個(gè)地址，其他處理器監(jiān)聽總線。若其他處理器在自己的cache中發(fā)現(xiàn)這個(gè)地址，則廣播該地址及其值來做出響應(yīng)。若所有處理器都沒發(fā)現(xiàn)這個(gè)地址，則以內(nèi)存地址及其所對(duì)應(yīng)的值進(jìn)行響應(yīng)。

getAndSet()的直接調(diào)用讓TASLock性能損失許多：

getAndSet()的調(diào)用實(shí)質(zhì)是總線上的一個(gè)廣播，這個(gè)調(diào)用將會(huì)延遲所有的線程，因?yàn)樗芯€程都要通過監(jiān)聽總線通信。

getAndSet()的調(diào)用會(huì)更新state的值，即使值仍為true，但是其他處理器cache中的鎖副本將會(huì)被丟棄，從而導(dǎo)致cache缺失。

當(dāng)持有鎖的線程試圖釋放鎖時(shí)可能被延遲，因?yàn)榭偩€被正在自旋的線程獨(dú)占。

與此相反，對(duì)于TTASLock算法采用的是本地旋轉(zhuǎn)（線程反復(fù)地重讀被緩存的值而不是反復(fù)地使用總線），線程A持有鎖時(shí)，線程B嘗試獲得鎖，但線程B只會(huì)在第一次讀鎖是cache缺失，之后每次cache命中不產(chǎn)生總線流量。

那么缺點(diǎn)來了，TTASLock釋放鎖時(shí)，會(huì)使各自旋線程處理器中的cache副本立即失效，這些線程會(huì)重新讀取這個(gè)值，造成總線流量風(fēng)暴。

對(duì)于TTASLock算法，當(dāng)鎖看似空閑（state.get()返回false）時(shí)，存在高爭用（多個(gè)線程試圖同時(shí)獲取一個(gè)鎖）的可能。高爭用意味著獲取鎖的可能性小，并且會(huì)造成總線流量增加。線程在重試之前回退一段時(shí)間是種不錯(cuò)的選擇。

這里實(shí)現(xiàn)的指數(shù)回退算法的回退準(zhǔn)則是，不成功嘗試的次數(shù)越多，發(fā)生爭用的可能性就越高，線程需要后退的時(shí)間就應(yīng)越長。

public class BackoffLock {
    private AtomicBoolean state = new AtomicBoolean(false);
    private static final int MIN_DELAY = 10;
    private static final int MAX_DELAY = 100;

    public void lock() {
        Backoff backoff = new Backoff(MIN_DELAY, MAX_DELAY);
        while (true) {
            while (state.get()) {
                ;
            }
            if (! state.getAndSet(true)) {
                return;
            } else {
                backoff.backoff();
            }
        }
    }

    public void unlock() {
        state.set(false);
    }
}

class Backoff {
    private final int minDelay, maxDelay;
    int limit;
    final Random random;

    public Backoff(int min, int max) {
        minDelay = min;
        maxDelay = max;
        limit = minDelay;
        random = new Random();
    }

    public void backoff() {
        int delay = random.nextInt(limit);
        limit = Math.min(maxDelay, 2 * limit);
        try {
            Thread.sleep(delay);
        } catch (InterruptedException e) {
            ;
        }
    }
}

指數(shù)后退算法解決了TTASLock釋放鎖時(shí)的高爭用問題，但是它的性能與minDelay和maxDelay的選取密切相關(guān)，并且很難找到一個(gè)通用兼容的值。

另外，BackoffLock算法還有兩個(gè)問題：

cache一致性流量：所有線程都在同一個(gè)共享存儲(chǔ)單元上旋轉(zhuǎn)；

臨界區(qū)利用率低：后退時(shí)間無法確定，線程延遲可能過長。

下面的這些是隊(duì)列鎖，名字看上去奇形怪狀的，其實(shí)是發(fā)明者名字的首字母。隊(duì)列鎖就是將線程組織成一個(gè)隊(duì)列，讓每個(gè)線程在不同的存儲(chǔ)單元上旋轉(zhuǎn)，從而降低cache一致性流量。

基于循環(huán)數(shù)組實(shí)現(xiàn)隊(duì)列鎖ALock，每個(gè)線程檢測自己的slot對(duì)應(yīng)的flag[]域來判斷是否輪到自己。

一個(gè)線程想獲得鎖，就要調(diào)用lock()方法，獲得自增tail獲得分配的slot號(hào)，然后等待這個(gè)slot空閑；當(dāng)釋放鎖時(shí)，就要阻塞當(dāng)前slot，然后讓下一個(gè)slot可運(yùn)行。
當(dāng)flag[i]為true時(shí)，那么這個(gè)線程就有權(quán)獲得鎖。任意時(shí)刻的flag[]數(shù)組中，應(yīng)該只有一個(gè)slot的值為true。

public class ALock {
    ThreadLocal mySlotIndex = new ThreadLocal();
    AtomicInteger tail;
    volatile boolean [] flag;
    int size;

    public ALock() {
        size = 100;
        tail = new AtomicInteger(0);
        flag = new boolean[size];
        flag[0] = true;
    }

    public void lock() {
        int slot = tail.getAndIncrement() % size;
        mySlotIndex.set(slot);
        while (! flag[slot]) {
            ;
        }
    }

    public void unlock() {
        int slot = mySlotIndex.get();
        flag[slot] = false;
        flag[(slot + 1) % size] = true;
    }
}

mySlotIndex是線程的局部變量，只能被一個(gè)線程訪問，每個(gè)線程都有自己獨(dú)立初始化的副本。不需要保存在共享存儲(chǔ)器，不需要同步，不會(huì)產(chǎn)生一致性流量。使用get()和set()方法來訪問局部變量的值。

tail是常規(guī)變量，域被所有的線程共享，支持原子操作。

數(shù)組flag[]也是被多個(gè)線程共享的，但是每個(gè)線程都是在一個(gè)數(shù)組元素的本地cache副本上旋轉(zhuǎn)。

ALock對(duì)BackoffLock的改進(jìn)：在多個(gè)共享存儲(chǔ)單元上旋轉(zhuǎn)，將cache無效性降到最低；把一個(gè)線程釋放鎖和另一個(gè)線程獲得該鎖之間的時(shí)間間隔最小化；先來先服務(wù)的公平性。但是，數(shù)組的大小至少與最大的并發(fā)線程數(shù)相同，并不是空間有效的，當(dāng)并發(fā)線程最大個(gè)數(shù)為n時(shí)，同步L個(gè)不同對(duì)象就需要O(Ln)大小的空間。

CLH隊(duì)列鎖表示為QNode對(duì)象的鏈表，每個(gè)線程通過一個(gè)線程局部變量pred指向其前驅(qū)。每個(gè)線程通過檢測前驅(qū)結(jié)點(diǎn)的locked域來判斷是否輪到自己。如果該域?yàn)?b>true，則前驅(qū)線程要么已經(jīng)獲得鎖要么正在等待鎖；如果該域?yàn)?b>false，則前驅(qū)進(jìn)程已釋放鎖，輪到自己了。正常情況下，隊(duì)列鏈中只有一個(gè)結(jié)點(diǎn)的locked域?yàn)?b>false。

當(dāng)一個(gè)線程調(diào)用lock()方法想獲得鎖時(shí)，將自己的locked域置為true，表示該線程不準(zhǔn)備釋放鎖，然后并將自己的結(jié)點(diǎn)加入到隊(duì)列鏈尾部。最后就是在前驅(qū)的locked域上旋轉(zhuǎn)，等待前驅(qū)釋放鎖。當(dāng)這個(gè)線程調(diào)用unlock()方法要釋放鎖時(shí)，線程要將自己的locked域置為false，表示已經(jīng)釋放鎖，然后將前驅(qū)結(jié)點(diǎn)作為自己的新結(jié)點(diǎn)以便日后訪問。

那么問題來了，為什么要在釋放鎖時(shí)做myNode.set(myPred.get())這個(gè)處理呢？假設(shè)線程A釋放鎖，A的后繼結(jié)點(diǎn)為B，如果不使用這種處理方式，A在釋放鎖后馬上申請(qǐng)鎖將自己的locked域置為true，整個(gè)動(dòng)作在B檢測到前驅(qū)A釋放鎖之前，那么將發(fā)生死鎖。

public class CLHLock {
    AtomicReference tail;
    ThreadLocal myPred;
    ThreadLocal myNode;

    public CLHLock() {
        tail = new AtomicReference(new QNode());
        myPred = new ThreadLocal() {
            protected QNode initialValue() {
                return null;
            }
        };
        myNode = new ThreadLocal() {
            protected QNode initialValue() {
                return new QNode();
            }
        };
    }

    public void lock() {
        QNode qnode = myNode.get();
        qnode.locked = true;
        QNode pred = tail.getAndSet(qnode);
        myPred.set(pred);
        while (pred.locked) {
            ;
        }
    }

    public void unlock() {
        QNode qnode = myNode.get();
        qnode.locked = false;
        myNode.set(myPred.get());
    }

    class QNode {
        boolean locked = false;
    }
}

如果最大線程數(shù)為n，有L個(gè)不同對(duì)象，那么CLHLock只需要O(L+n)空間。比ALock所需空間少，并且不需要知道可能使用鎖的最大線程數(shù)量。但是，在無cache的系統(tǒng)上性能較差，因?yàn)橐淮我L問兩個(gè)結(jié)點(diǎn)，若這兩個(gè)結(jié)點(diǎn)內(nèi)存位置較遠(yuǎn)，性能損失會(huì)很大。

MCS隊(duì)列鎖通過檢測自己所在結(jié)點(diǎn)的locked的值來判斷是否輪到自己，等待這個(gè)域被前驅(qū)釋放鎖時(shí)改變。

線程若要獲得鎖，需把自己結(jié)點(diǎn)添加到鏈表的尾部。若隊(duì)列鏈表原先為空，則獲得鎖。否則，將前驅(qū)結(jié)點(diǎn)的next域指向自己，在自己的locked域上自旋等待，直到前驅(qū)將該域置為false。線程若要釋放鎖，判斷是否在隊(duì)尾，如果是只需將tail置為null，如果不是需將后繼的locked域置為false且將自己結(jié)點(diǎn)的next域置為默認(rèn)的null。注意在隊(duì)尾的情況，可能有個(gè)線程正在獲得鎖，要等一下變?yōu)楹笠环N情況。

public class MCSLock {
    AtomicReference tail;
    ThreadLocal myNode;

    public MCSLock() {
        tail = new AtomicReference(null);
        myNode = new ThreadLocal() {
            protected QNode initialValue() {
                return new QNode();
            }
        };
    }

    public void lock() {
        QNode qnode = myNode.get();
        QNode pred = tail.getAndSet(qnode);
        if (pred != null) {
            qnode.locked = true;
            pred.next = qnode;
            while (qnode.locked) {
                ;
            }
        }
    }

    public void unlock() {
        QNode qnode = myNode.get();
        if (qnode.next == null) {
            if (tail.compareAndSet(qnode, null)) {
                return;
            }
            while (qnode.next == null) {
                ;
            }
        }
        qnode.next.locked = false;
        qnode.next = null;
    }

    class QNode {
        boolean locked = false;
        QNode next = null;
    }
}

結(jié)點(diǎn)能被重復(fù)使用，該鎖的空間復(fù)雜度為O(L+n)。MCSLock算法適合無cache的系統(tǒng)結(jié)構(gòu)，因?yàn)槊總€(gè)線程只需控制自己自旋的存儲(chǔ)單元。但是，釋放鎖也需要旋轉(zhuǎn)，另外讀寫比較次數(shù)比CLHLock多。

Lock接口中有個(gè)一個(gè)tryLock()方法，可以指定一個(gè)時(shí)限（獲得鎖可等待的最大時(shí)間），若獲得鎖超時(shí)則放棄嘗試。最后返回一個(gè)布爾值說明鎖是否申請(qǐng)成功。

時(shí)限隊(duì)列鎖TOLock是基于CLHLock類的，鎖是一個(gè)結(jié)點(diǎn)的虛擬隊(duì)列，每個(gè)結(jié)點(diǎn)在它的前驅(qū)結(jié)點(diǎn)上自旋等待鎖釋放。但是當(dāng)這個(gè)線程超時(shí)時(shí)，不能簡單的拋棄它的隊(duì)列結(jié)點(diǎn)，而是將這個(gè)結(jié)點(diǎn)標(biāo)記為已廢棄，這樣它的后繼們（如果有）就會(huì)注意到這個(gè)結(jié)點(diǎn)已放棄，轉(zhuǎn)而在放棄結(jié)點(diǎn)的前驅(qū)上自旋。為了保證隊(duì)列鏈表的連續(xù)性，每次申請(qǐng)鎖都會(huì)new一個(gè)QNode。

時(shí)限隊(duì)列鎖結(jié)點(diǎn)的域pred會(huì)特殊一點(diǎn)，它有3種類型的取值：

null：初始狀態(tài)，未獲得鎖或已釋放鎖；

AVAILABLE：一個(gè)靜態(tài)結(jié)點(diǎn)，表示對(duì)應(yīng)結(jié)點(diǎn)已釋放鎖，申請(qǐng)鎖成功；

QNode：pred域?yàn)?b>null的前驅(qū)結(jié)點(diǎn)，表示對(duì)應(yīng)結(jié)點(diǎn)因超時(shí)放棄鎖請(qǐng)求，在放棄請(qǐng)求時(shí)才會(huì)設(shè)置這個(gè)值。

申請(qǐng)鎖時(shí)，創(chuàng)建一個(gè)pred域?yàn)?b>null的新結(jié)點(diǎn)并加入到鏈表尾部，若原先鏈表為空或前驅(qū)結(jié)點(diǎn)已釋放鎖，則獲得鎖。否則，在嘗試時(shí)間內(nèi)，找到pred域?yàn)?b>null的前驅(qū)結(jié)點(diǎn)，等待它釋放鎖。若在等待前驅(qū)結(jié)點(diǎn)釋放鎖的過程中超時(shí)，就嘗試從鏈表中刪除這個(gè)結(jié)點(diǎn)，要分這個(gè)結(jié)點(diǎn)是否有后繼兩種情況。
釋放鎖時(shí)，檢查該結(jié)點(diǎn)是否有后繼，若無后繼可直接把tail設(shè)置為null，否則將該結(jié)點(diǎn)的pred域指向AVAILABLE。

public class TOLock {
    static QNode AVAILABLE = new QNode();
    AtomicReference tail;
    ThreadLocal myNode;

    public TOLock() {
        tail = new AtomicReference(null);
        myNode = new ThreadLocal();
    }

    public boolean tryLock(long time, TimeUnit unit) {
        long startTime = System.currentTimeMillis();
        long patience = TimeUnit.MILLISECONDS.convert(time, unit);
        QNode qnode = new QNode();
        myNode.set(qnode);
        qnode.pred = null;
        QNode myPred = tail.getAndSet(qnode);
        if (myPred == null || myPred.pred == AVAILABLE) {
            return true;
        }
        while (System.currentTimeMillis() - startTime < patience) {
            QNode predPred = myPred.pred;
            if (predPred == AVAILABLE) {
                return true;
            } else {
                if (predPred != null) {
                    myPred = predPred;
                }
            }
        }
        if (! tail.compareAndSet(qnode, myPred)) {
            qnode.pred = myPred;
        }
        return false;
    }

    public void unlock() {
        QNode qnode = myNode.get();
        if (! tail.compareAndSet(qnode, null)) {
            qnode.pred = AVAILABLE;
        }
    }

    static class QNode {
        public QNode pred = null;
    }
}

優(yōu)點(diǎn)：同CLHLock。
缺點(diǎn)：每次申請(qǐng)鎖都要分配一個(gè)新結(jié)點(diǎn)，在鎖上旋轉(zhuǎn)的線程可能要回溯一個(gè)超時(shí)結(jié)點(diǎn)鏈。

上面實(shí)現(xiàn)的這些鎖算法不支持重入。我們可以使用銀行轉(zhuǎn)賬的例子來測試一下鎖的效果，任意賬戶間可以隨意轉(zhuǎn)賬，鎖生效時(shí)所有賬戶的總金額是不變的。完整的算法實(shí)現(xiàn)和測試代碼在這里。