簡介

在AQS的源碼中有介紹到一種叫CLH的隊列(AQS中使用的是一種變種).CLH它是一種基于單向鏈表實現(xiàn)的隊列,即后一個節(jié)點保存前一個節(jié)點的引用形成單向鏈接.我們可以使用CLH來實現(xiàn)一個高效的自旋鎖.每個線程請求鎖時,都先判斷它的前節(jié)點是否需要鎖,如果前節(jié)點需要鎖則自己自旋等待.如果前節(jié)點不需要鎖則獲取鎖成功.
正式因為它的這種實現(xiàn)方式,它具有先來先獲取的公平性.而它的核心基于一個CAS操作來實現(xiàn)具有很高的性能.但因為其自旋等待,所以不太適合占用鎖時間過長的場景,因為自旋是會需要消耗CPU資源的.

簡單實現(xiàn)

/**
 * 聲明鎖接口
 */
public interface Lock {
    /**
     * 獲取鎖
     */
    void lock();

    /**
     * 解鎖
     */
    void unlock();
}

/**
 * 定義Node
 */
public class CLHNode {
    /**
     * 使用volatile修飾保證可見性.
     */
    volatile boolean locked;
}

/**
 * CLHLock實現(xiàn)
 */
public class CLHLock implements Lock{
    /**
     * 保存當前節(jié)點,一個線程一個
     */
    private final ThreadLocal<CLHNode> node;
    /**
     * 尾節(jié)點.所有線程共享一個.支持CAS操作
     */
    private final AtomicReference<CLHNode> tail;
    /**
     * 父節(jié)點.一個線程一個
     */
    private final ThreadLocal<CLHNode> pred;

    public CLHLock() {
        this.node = ThreadLocal.withInitial(CLHNode::new);
        this.tail = new AtomicReference<>(new CLHNode());
        this.pred = new ThreadLocal<>();
    }

    @Override
    public void lock() {
        //獲取當前節(jié)點
        CLHNode current = node.get();
        //將當前節(jié)點設置成需要鎖
        current.locked = true;
        //將當前節(jié)點設置成尾節(jié)點,CAS操作
        CLHNode currentPred = tail.getAndSet(current);
        //設置當前節(jié)點的前節(jié)點
        pred.set(currentPred);
        //查看當前節(jié)點是否需要鎖,如果需要則自旋等待
        while (currentPred.locked){

        }
        //獲取鎖成功
        System.out.println(Thread.currentThread().getName()+":獲取到鎖");
    }

    @Override
    public void unlock() {
        //獲取當前節(jié)點
        CLHNode current = node.get();
        //將當前節(jié)點locked設置成false,它后面的節(jié)點將會推出自旋
        current.locked = false;
        //將當前節(jié)點設置成前節(jié)點
        node.set(pred.get());
    }
}

/**
 * 使用測試
 */
public class App {

    public static void main(String[] args) {
        Lock lock = new CLHLock();

        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                lock.lock();
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                lock.unlock();
            }
        };

        Thread t1 = new Thread(runnable);
        t1.setName("t1");
        t1.start();

        Thread t2 = new Thread(runnable);
        t2.setName("t2");
        t2.start();

        lock.lock();
        lock.unlock();

    }
}

• 獲取鎖:每個線程獲取鎖時都自己節(jié)點的locked設置成true代表自己需要鎖,然后將自己節(jié)點設置成尾節(jié)點tail(這里使用AtomicReference來實現(xiàn)一個CAS操作),并獲取到當前節(jié)點的前置節(jié)點.接著獲取的前置節(jié)點設置成自己的前置節(jié)點.然后就是判斷通過前置節(jié)點的locked屬性來判斷前置節(jié)點是否需要鎖.這里需要注意的是locked屬性需要使用volatile修飾來保證其他線程修改它的值而當前線程能感知到該值的變化.如果去掉volatile可能導致當前線程一直自旋無法獲取到鎖.
• 釋放鎖:釋放鎖就比較好理解了,直接獲取到當前節(jié)點(這就是為什么使用ThreadLocal來保存當前節(jié)點的值了).然后將當前節(jié)點的locked設置成false,而這個操作將會使該節(jié)點的后置節(jié)點退出自旋獲取到鎖.

關于解鎖操作中的node.set(pred.get()).它主要是為了防止獲取鎖解鎖之后再次獲取鎖無法獲取鎖的bug.如果去掉該行代碼,第一次獲取鎖當前節(jié)點已經(jīng)是tail節(jié)點了,然后再次獲取鎖.再次將當前節(jié)點作為尾節(jié)點.即當前節(jié)點的頭節(jié)點和當前節(jié)點是同一個節(jié)點.如果這個時候設置locked為true,這將導致當前節(jié)點再判斷前節(jié)點是否需要鎖時一直都是true.線程將在獲取鎖時無限自旋等待.

總結(jié)

總體來說使用這種方式的有點如下:

• 簡單高效.
• 支持FIFO公平性.
• 自旋等待避免線程切換的性能損耗

同樣它也有下面的缺點:

• 不可重入.即獲取鎖之后必須釋放鎖才能再次獲取鎖.
• 鎖占用時間過長導致CPU空轉(zhuǎn).