隱語義模型（Latent factor model，以下簡稱LFM），是基於矩陣分解的推薦算法，在其基本算法上引入L2正則的FunkSVD算法在推薦系統領域更是廣泛使用，在Spark上也有其實現。本文將對 LFM原理進行詳細闡述，給出其基本算法原理。此外，還將介紹使得隱語義模型聲名大噪的算法FunkSVD和在其基礎上改進較為成功的BiasSVD。最後，對LFM進行一個較為全面的總結。

1. 矩陣分解應用於推薦算法要解決的問題

在推薦系統中，我們經常可能面臨的場景是：現有大量用戶和物品，以及少部分用戶對少部分物品的評分，我們需要使用現有的用戶對少部分物品的評分去推測用戶對物品集中其他物品的可能的評分，從而將預測中評分高的物品推薦給用戶。例如下面的用戶物品評分表：

對於每個用戶，我們希望較準確的預測出其對未評分物品的評分。將m個用戶和n個物品的評分看做一個矩陣M，從而將矩陣分解應用到該場景，即可解決這一問題。而本文，將關注於矩陣分解用於到推薦的方法之一，即LFM算法的解決方案。

2. LFM

LFM算法的核心思想是通過隱含特徵（Latent factor）聯繫用戶和物品，該算法最早在文本挖掘領域中被提出用於找到文本的隱含語義，相關名詞還有LDA和Topic Model等。

2.1 如何表示用戶的偏好和物品（item）屬性？

在被問到這個問題時，針對MovieLens（電影評分）數據集，你可能會說用戶是否喜歡動作片，物品所屬電影類型去回答。但用戶對其他類型的電影的偏好程度呢？物品在其它類型所佔的權重又是多少呢？就算針對現有的電影類型去表徵用戶偏好和物品，那麼能否能夠完全的表示出用戶的偏好和物品屬性呢？答案是不能，比如用戶喜歡看成龍的電影這個偏好沒法表示出來，電影由誰導演，主演是誰沒法表示。但你要問我用哪些屬性去表徵呢？這個誰也無法給出一個很好的答案，粒度很難控制。

2.2 LFM來救場

隱語義模型較好的解決了該問題，它從數據出發，通過基於用戶行為統計的自動聚類，可指定出表徵用戶偏好和物品的向量維度，最終得到用戶的偏好向量以及物品的表徵向量。LFM通過以下公式計算用戶u對物品i的偏好：
\[ preference(u, i)=p^T_u q_i=\sum_f^F{p_{u,k}q_{i,k}} \]
這個公式，\(p_{u,k}\)度量了用戶u的偏好和第f個隱類的關係，\(q_{i,k}\)度量了物品i和第f個隱類的關係。

那麼現在，我們期望用戶的評分矩陣M這樣分解：
\[ M_{m*n}=P^T_{m*k}Q_{k*n} \]
那麼，我們如何將矩陣分解呢？這裏採用了線性回歸的思想，即盡可能的讓用戶的評分和我們預測的評分的殘差盡可能小，也就是說，可以用均方差作為損失函數來尋找最終的P和Q。考慮所有的用戶和樣本的組合，我們期望的最小化損失函數為：
\[ \sum_{i,j}{(m_{ij}-p_i^Tq_j)^2} \]
只要我們能夠最小化上面的式子，並求出極值所對應的\(p_i\) 和\(q_j\)，則我們最終可以得到矩陣P和Q，那麼對於任意矩陣M任意一個空白評分的位置，我們就可以通過\(p^T_i q_j\)計算預測評分。

2.3 FunkSVD用於推薦

上面是隱語義模型LFM的基本原理，但在實際業務中，為防止過擬合，我們常常會加入一個L2的正則化項，這也就誕生了我們的FunkSVD算法。其優化目標函數\(J(p,q)\)定義為：
\[ \underbrace{argmin}_{p_i,q_j}\sum_{i,j}{(m_{ij}-p^T_iq_j)^2+\lambda({\Arrowvert{p_i}\Arrowvert}^2_2+{\Arrowvert{q_i}\Arrowvert}^2_2)} \]
其中λ為正則化係數，需要調參。對於這個優化問題，我們一般通過梯度下降法來進行優化得到結果。

將上式分別對\(p_i\)和\(q_j\)求導我們得到：
\[ \frac{\partial{J}}{\partial{p_i}}=-2(m_{ij}-p^T_iq_j)q_j+2\lambda{p_i} \]

\[ \frac{\partial{J}}{\partial{q_j}}=-2(m_{ij}-p^T_iq_j)p_i+2\lambda{q_j} \]

則梯度下降中迭代公式為：
\[ p_i = p_i +\alpha((m_{ij}-p^T_iq_j)q_j-\lambda{p_i}) \]

\[ q_j = q_j+\alpha((m_{ij}-p^T_iq_j)p_i-\lambda{q_j}) \]

通過迭代我們最終可以得到P和Q，進而用於推薦。

為讀者進一步理解，筆者實現了基於MovieLens數據集實現了該方法。代碼詳見github:

2.4 BiasSVD用於推薦

BiasSVD是FunkSVD較為成功的改進版算法。BiasSVD假設評分系統包括三部分的偏置因素：一些和用戶物品無關的評分因素。用戶有一些和物品無關的評分因素，稱為用戶偏置項。而物品也有一些和用戶無關的評分因素，稱為物品偏置項。這很好理解，對於樂觀的用戶來說，它的評分行為普遍偏高，而對批判性用戶來說，他的評分記錄普遍偏低，即使他們對同一物品的評分相同，但是他們對該物品的喜好程度卻並不一樣。同理，對物品來說，以電影為例，受大眾歡迎的電影得到的評分普遍偏高，而一些爛片的評分普遍偏低，這些因素都是獨立於用戶或產品的因素，而和用戶對產品的的喜好無關。

假設評分系統平均分為μ,第i個用戶的用戶偏置項為\(b_i\),而第j個物品的物品偏置項為\(b_j\)，則加入了偏置項以後的優化目標函數\(J(p_i,q_j)\)是這樣的:
\[ \underbrace{argmin}_{p_i,q_j}\sum_{i,j}{(m_{ij}-p^T_iq_j-u-b_i-b_j)^2+\lambda({\Arrowvert{p_i}\Arrowvert}^2_2+{\Arrowvert{q_i}\Arrowvert}^2_2+{\Arrowvert{b_i}\Arrowvert}^2_2+{\Arrowvert{b_j}\Arrowvert}^2_2)} \]
這個優化目標也可以採用梯度下降法求解。和FunkSVD不同的是，此時我們多了兩個偏執項\(b_i\)和 \(b_j\)，\(p_i\)和\(q_j\)的迭代公式和FunkSVD類似，只是每一步的梯度導數稍有不同而已。\(b_i\)和 \(b_j\)一般可以初始設置為0，然後參与迭代。迭代公式為：
\[ p_i = p_i +\alpha((m_{ij}-p^T_iq_j-u-b_i-b_j)q_j-\lambda{p_i}) \]

\[ q_j = q_j+\alpha((m_{ij}-p^T_iq_j-u-b_i-b_j)p_i-\lambda{q_j}) \]

\[ b_i=b_i+\alpha(m_{ij}-p^T_iq_j-u-b_i-b_j-\lambda{b_i}) \]

\[ b_j=b_j+\alpha(m_{ij}-p^T_iq_j-u-b_i-b_j-\lambda{b_j}) \]

通過迭代我們最終可以得到P和Q，進而用於推薦。BiasSVD增加了一些額外因素的考慮，因此在某些場景會比FunkSVD表現好。

為讀者進一步理解，筆者實現了基於MovieLens數據集實現了該方法。代碼詳見github：

小結

LFM 是一種基於機器學習的方法，具有比較好的理論基礎，通過優化一個設定的指標建立最優的模型。它實質上是矩陣分解應用到推薦的方法，其中FunkSVD更是將矩陣分解用於推薦方法推到了新的高度，在實際應用中使用非常廣泛。當然矩陣分解方法也在不停的進步，目前矩陣分解推薦算法中分解機方法（factorization machine, FM）已成為一個趨勢。

對於矩陣分解用於推薦方法本身來說，它容易編程實現，實現複雜度低，預測效果也好，同時還能保持擴展性。這些都是其寶貴的優點。但是LFM 無法給出很好的推薦解釋，它計算出的隱類雖然在語義上確實代表了一類興趣和物品，卻很難用自然語言描述並生成解釋展現給用戶。

LFM 在建模過程中，假設有 M 個用戶、 N 個物品、 K 條用戶對物品的行為記錄，如果是 F 個隱類，那麼它離線計算的空間複雜度是 \(O(F*(M+N))\) ，迭代 S次則時間複雜度為 \(O(K * F * S)\)。當 M(用戶數量)和 N(物品數量)很大時LFM相對於ItemCF和UserCF可以很好地節省離線計算的內存，在時間複雜度由於LFM會多次迭代上所以和ItemCF、UserCF沒有質的差別。

同時，遺憾的是，LFM 無法進行在線實時推薦，即當用戶有了新的行為後，他的推薦列表不會發生變化。而從 LFM的預測公式可以看到， LFM 在給用戶生成推薦列表時，需要計算用戶對所有物品的興趣權重，然後排名，返回權重最大的 N 個物品。那麼，在物品數很多時，這一過程的時間複雜度非常高，可達 \(O(M*N*F)\) 。因此， LFM 不太適合用於物品數非常龐大的系統，如果要用，我們也需要一個比較快的算法給用戶先計算一個比較小的候選列表，然後再用LFM重新排名。另一方面，LFM 在生成一個用戶推薦列表時速度太慢，因此不能在線實時計算，而需要離線將所有用戶的推薦結果事先計算好存儲在數據庫中。

參考：

• 推薦系統實戰—項亮

（歡迎轉載，轉載請註明出處。歡迎溝通交流： losstie@outlook.com）

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理【其他文章推薦】

※網頁設計公司推薦更多不同的設計風格，搶佔消費者視覺第一線

※廣告預算用在刀口上，網站設計公司幫您達到更多曝光效益

※自行創業 缺乏曝光? 下一步"網站設計"幫您第一時間規劃公司的門面形象

　　為充分利用機器性能，人們發明了多線程。但同時帶來了線程安全問題，於是人們又發明了同步鎖。

　　這個問題自然人人知道，但你真的了解同步鎖嗎？還是說你會用其中的上鎖與解鎖功能？

　　今天我們就一起來深入看同步鎖的原理和實現吧！

一、同步鎖的職責

　　同步鎖的職責可以說就一個，限制資源的使用（線程安全從屬）。

　　它一般至少會包含兩個功能: 1. 給資源加鎖； 2. 給資源解鎖；另外，它一般還有 等待/通知 即 wait/notify 的功能；

　　同步鎖的應用場景：多個線程同時操作一個事務必須保證正確性；一個資源只能同時由一線程訪問操作；一個資源最多只能接入k的併發訪問；保證訪問的順序性；

　　同步鎖的實現方式：操作系統調度實現；應用自行實現；CAS自旋；

　　同步鎖的幾個問題：

　　　　為什麼它能保證線程安全？

　　　　鎖等待耗CPU嗎？

　　　　使用鎖后性能下降嚴重的原因是啥？

二、同步鎖的實現一：lock/unlock

　　其實對於應用層來說，非常多就是 lock/unlock , 這也是鎖的核心。

　　AQS 是java中很多鎖實現的基礎，因為它屏蔽了很多繁雜而底層的阻塞操作，為上層抽象出易用的接口。

　　我們就以AQS作為跳板，先來看一下上鎖的過程。為不至於陷入具體鎖的業務邏輯中，我們先以最簡單的 CountDownLatch 看看。

    // 先看看 CountDownLatch 的基礎數據結構，可以說是不能再簡單了，就繼承了 AQS，然後簡單覆寫了幾個必要方法。
    // java.util.concurrent.CountDownLatch.Sync
    /**
     * Synchronization control For CountDownLatch.
     * Uses AQS state to represent count.
     */
    private static final class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 4982264981922014374L;

        Sync(int count) {
            setState(count);
        }

        int getCount() {
            return getState();
        }

        protected int tryAcquireShared(int acquires) {
            // 只有一種情況會獲取鎖成功，即 state == 0 的時候
            return (getState() == 0) ? 1 : -1;
        }

        protected boolean tryReleaseShared(int releases) {
            // Decrement count; signal when transition to zero
            for (;;) {
                int c = getState();
                if (c == 0)
                    return false;
                // 原始的鎖數量是在初始化時指定的不可變的，每次釋放一個鎖標識
                int nextc = c-1;
                if (compareAndSetState(c, nextc))
                    // 只有一情況會釋放鎖成功，即本次釋放后 state == 0
                    return nextc == 0;
            }
        }
    }
    private final Sync sync;

重點1，我們看看上鎖過程，即 await() 的調用。

    public void await() throws InterruptedException {
        // 調用 AQS 的接口，由AQS實現了鎖的骨架邏輯
        sync.acquireSharedInterruptibly(1);
    }
    
    // java.util.concurrent.locks.AbstractQueuedSynchronizer#acquireSharedInterruptibly
    /**
     * Acquires in shared mode, aborting if interrupted.  Implemented
     * by first checking interrupt status, then invoking at least once
     * {@link #tryAcquireShared}, returning on success.  Otherwise the
     * thread is queued, possibly repeatedly blocking and unblocking,
     * invoking {@link #tryAcquireShared} until success or the thread
     * is interrupted.
     * @param arg the acquire argument.
     * This value is conveyed to {@link #tryAcquireShared} but is
     * otherwise uninterpreted and can represent anything
     * you like.
     * @throws InterruptedException if the current thread is interrupted
     */
    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        // 首先嘗試獲取鎖，如果成功就不用阻塞了
        // 而從上面的邏輯我們看到，獲取鎖相當之簡單，所以，獲取鎖本身並沒有太多的性能消耗喲
        // 如果獲取鎖失敗，則會進行稍後嘗試，這應該是複雜而精巧的
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }
    
    /**
     * Acquires in shared interruptible mode.
     * @param arg the acquire argument
     */
    private void doAcquireSharedInterruptibly(int arg)
        throws InterruptedException {
        // 首先將當前線程添加排隊隊尾，此處會保證線程安全，稍後我們可以看到
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            for (;;) {
                // 獲取其上一節點，如果上一節點是頭節點，就代表當前線程可以再次嘗試獲取鎖了
                final Node p = node.predecessor();
                if (p == head) {
                    int r = tryAcquireShared(arg);
                    if (r >= 0) {
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        failed = false;
                        return;
                    }
                }
                // 先檢測是否需要阻塞，然後再進行阻塞等待，阻塞由 LockSupport 底層支持
                // 如果阻塞后，將不會主動喚醒，只會由 unlock 時，主動被通知
                // 因此，此處即是獲取鎖的最終等待點
                // 操作系統將不會再次調度到本線程，直到獲取到鎖
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

    // 如此線程安全地添加當前線程到隊尾？ CAS 保證
    /**
     * Creates and enqueues node for current thread and given mode.
     *
     * @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared
     * @return the new node
     */
    private Node addWaiter(Node mode) {
        Node node = new Node(Thread.currentThread(), mode);
        // Try the fast path of enq; backup to full enq on failure
        Node pred = tail;
        if (pred != null) {
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        enq(node);
        return node;
    }
    /**
     * Inserts node into queue, initializing if necessary. See picture above.
     * @param node the node to insert
     * @return node's predecessor
     */
    private Node enq(final Node node) {
        for (;;) {
            Node t = tail;
            if (t == null) { // Must initialize
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }
    
    // 檢測是否需要進行阻塞
    /**
     * Checks and updates status for a node that failed to acquire.
     * Returns true if thread should block. This is the main signal
     * control in all acquire loops.  Requires that pred == node.prev.
     *
     * @param pred node's predecessor holding status
     * @param node the node
     * @return {@code true} if thread should block
     */
    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
            /*
             * This node has already set status asking a release
             * to signal it, so it can safely park.
             */
             // 只有前置節點是 SIGNAL 狀態的節點，才需要進行 阻塞等待，當然前置節點會在下一次循環中被設置好
            return true;
        if (ws > 0) {
            /*
             * Predecessor was cancelled. Skip over predecessors and
             * indicate retry.
             */
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            /*
             * waitStatus must be 0 or PROPAGATE.  Indicate that we
             * need a signal, but don't park yet.  Caller will need to
             * retry to make sure it cannot acquire before parking.
             */
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }
    
    // park 阻塞實現
    /**
     * Convenience method to park and then check if interrupted
     *
     * @return {@code true} if interrupted
     */
    private final boolean parkAndCheckInterrupt() {
        // 將當前 AQS 實例作為鎖對象 blocker, 進行操作系統調用阻塞, 所以所有等待鎖的線程將會在同一個鎖前提下執行
        LockSupport.park(this);
        return Thread.interrupted();
    }

　　如上，上鎖過程是比較簡單明了的。加入一隊列，然後由操作系統將線程調出。（那麼操作系統是如何把線程調出的呢？有興趣自行研究）

重點2. 解鎖過程，即 countDown() 調用

    public void countDown() {
        // 同樣直接調用 AQS 的接口，由AQS實現了鎖的釋放骨架邏輯
        sync.releaseShared(1);
    }
    // java.util.concurrent.locks.AbstractQueuedSynchronizer#releaseShared
    /**
     * Releases in shared mode.  Implemented by unblocking one or more
     * threads if {@link #tryReleaseShared} returns true.
     *
     * @param arg the release argument.  This value is conveyed to
     *        {@link #tryReleaseShared} but is otherwise uninterpreted
     *        and can represent anything you like.
     * @return the value returned from {@link #tryReleaseShared}
     */
    public final boolean releaseShared(int arg) {
        // 調用業務實現的釋放邏輯，如果成功，再執行底層的釋放，如隊列移除，線程通知等等
        // 在 CountDownLatch 的實現中，只有 state == 0 時才會成功，所以它只會執行一次底層釋放
        // 這也是我們認為 CountDownLatch 能夠做到多線程同時執行的效果的原因之一
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }
    
    /**
     * Release action for shared mode -- signals successor and ensures
     * propagation. (Note: For exclusive mode, release just amounts
     * to calling unparkSuccessor of head if it needs signal.)
     */
    private void doReleaseShared() {
        /*
         * Ensure that a release propagates, even if there are other
         * in-progress acquires/releases.  This proceeds in the usual
         * way of trying to unparkSuccessor of head if it needs
         * signal. But if it does not, status is set to PROPAGATE to
         * ensure that upon release, propagation continues.
         * Additionally, we must loop in case a new node is added
         * while we are doing this. Also, unlike other uses of
         * unparkSuccessor, we need to know if CAS to reset status
         * fails, if so rechecking.
         */
        for (;;) {
            Node h = head;
            // 隊列不為空才進行釋放
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                // 看過上面的 lock 邏輯，我們知道只要在阻塞狀態，一定是 Node.SIGNAL 
                if (ws == Node.SIGNAL) {
                    // 狀態改變成功，才進行後續的喚醒邏輯
                    // 因為先改變狀態成功，才算是線程安全的，再進行喚醒，否則進入下一次循環再檢查
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;            // loop to recheck cases
                    // 將頭節點的下一節點喚醒，如有必要
                    unparkSuccessor(h);
                }
                // 這裏的 propagates, 是要傳播啥呢？？
                // 為什麼只喚醒了一個線程，其他線程也可以動了？
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
            if (h == head)                   // loop if head changed
                break;
        }
    }
    /**
     * Wakes up node's successor, if one exists.
     *
     * @param node the node
     */
    private void unparkSuccessor(Node node) {
        /*
         * If status is negative (i.e., possibly needing signal) try
         * to clear in anticipation of signalling.  It is OK if this
         * fails or if status is changed by waiting thread.
         */
        int ws = node.waitStatus;
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);

        /*
         * Thread to unpark is held in successor, which is normally
         * just the next node.  But if cancelled or apparently null,
         * traverse backwards from tail to find the actual
         * non-cancelled successor.
         */
        // 喚醒下一個節點
        // 但如果下一節點已經取消等待了，那麼就找下一個沒最近的沒被取消的線程進行喚醒
        // 喚醒只是針對一個線程的喲
        Node s = node.next;
        if (s == null || s.waitStatus > 0) {
            s = null;
            for (Node t = tail; t != null && t != node; t = t.prev)
                if (t.waitStatus <= 0)
                    s = t;
        }
        if (s != null)
            LockSupport.unpark(s.thread);
    }

重要3. 線程解鎖的傳播性？

　　因為從上一節的講解中，我們看到，當用戶調用 countDown 時，僅僅是讓操作系統喚醒了 head 的下一個節點線程或者最近未取消的節點。那麼，從哪裡來的所有線程都獲取了鎖從而運行呢？

　　其實是在 獲取鎖的過程中，還有一點我們未看清：

    // java.util.concurrent.locks.AbstractQueuedSynchronizer#doAcquireShared
    /**
     * Acquires in shared uninterruptible mode.
     * @param arg the acquire argument
     */
    private void doAcquireShared(int arg) {
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();
                if (p == head) {
                    // 當countDown被調用后，head節點被喚醒，執行
                    int r = tryAcquireShared(arg);
                    if (r >= 0) {
                        // 獲取到鎖后，設置node為下一個頭節點，並把喚醒狀態傳播下去，而這裏面肯定會做一些喚醒其他線程的操作，請看下文
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        if (interrupted)
                            selfInterrupt();
                        failed = false;
                        return;
                    }
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }
    
    /**
     * Sets head of queue, and checks if successor may be waiting
     * in shared mode, if so propagating if either propagate > 0 or
     * PROPAGATE status was set.
     *
     * @param node the node
     * @param propagate the return value from a tryAcquireShared
     */
    private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head; // Record old head for check below
        setHead(node);
        /*
         * Try to signal next queued node if:
         *   Propagation was indicated by caller,
         *     or was recorded (as h.waitStatus either before
         *     or after setHead) by a previous operation
         *     (note: this uses sign-check of waitStatus because
         *      PROPAGATE status may transition to SIGNAL.)
         * and
         *   The next node is waiting in shared mode,
         *     or we don't know, because it appears null
         *
         * The conservatism in both of these checks may cause
         * unnecessary wake-ups, but only when there are multiple
         * racing acquires/releases, so most need signals now or soon
         * anyway.
         */
        if (propagate > 0 || h == null || h.waitStatus < 0 ||
            (h = head) == null || h.waitStatus < 0) {
            // 如果有必要，則做一次喚醒下一線程的操作
            // 在 countDown() 不會觸發此操作，所以這裏只是一個內部調用傳播
            Node s = node.next;
            if (s == null || s.isShared())
                // 此處鎖釋放邏輯如上，總之，又是另一次的喚醒觸發
                doReleaseShared();
        }
    }

　　到此，我們明白了它是怎麼做到一個鎖釋放，所有線程可通行的。也從根本上回答了我們猜想，所有線程同時併發運行。然而並沒有，它只是通過喚醒傳播性來依次喚醒各個等待線程的。從絕對時間性上來講，都是有先後關係的。以後可別再淺顯說是同時執行了喲。

三、 鎖的切換：wait/notify

　　上面看出，針對一個lock/unlock 的過程還是很簡單的，由操作系統負責大頭，實現代碼也並不多。

　　但是針對稍微有點要求的場景，就會進行條件式的操作。比如：持有某個鎖運行一段代碼，但是，運行時發現某條件不滿足，需要進行等待而不能直接結束，直到條件成立。即所謂的 wait 操作。

　　乍一看，wait/notify 與 lock/unlock 很像，其實不然。區分主要是 lock/unlock 是針對整個代碼段的，而 wait/notify 則是針對某個條件的，即獲取了鎖不代表條件成立了，但是條件成立了一定要在鎖的前提下才能進行安全操作。

　　那麼，是否 wait/notify 也一樣的實現簡單呢？比如java的最基礎類 Object 類就提供了 wait/notify 功能。

　　我們既然想一探究竟，還是以併發包下的實現作為基礎吧，畢竟 java 才是我們的強項。

　　本次，咱們以  ArrayBlockingQueue#put/take 作為基礎看下這種場景的使用先。

　　ArrayBlockingQueue 的put/take 特性就是，put當隊列滿時，一直阻塞，直到有可用位置才繼續運行下一步。而take當隊列為空時一樣阻塞，直到隊列里有數據才運行下一步。這種場景使用鎖主不好搞了，因為這是一個條件判斷。put/take 如下：

    // java.util.concurrent.ArrayBlockingQueue#put
    /**
     * Inserts the specified element at the tail of this queue, waiting
     * for space to become available if the queue is full.
     *
     * @throws InterruptedException {@inheritDoc}
     * @throws NullPointerException {@inheritDoc}
     */
    public void put(E e) throws InterruptedException {
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            // 當隊列滿時，一直等待
            while (count == items.length)
                notFull.await();
            enqueue(e);
        } finally {
            lock.unlock();
        }
    }
    
    // java.util.concurrent.ArrayBlockingQueue#take
    public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            // 當隊列為空時一直等待
            while (count == 0)
                notEmpty.await();
            return dequeue();
        } finally {
            lock.unlock();
        }
    }

　　看起來相當簡單，完全符合人類思維。只是，這裏使用的兩個變量進行控制流程 notFull,notEmpty. 這兩個變量是如何進行關聯的呢？

　　在這之前，我們還需要補充下上面的例子，即 notFull.await(), notEmpty.await(); 被阻塞了，何時才能運行呢？如上代碼在各自的入隊和出隊完成之後進行通知就可以了。

    // 與 put 對應，入隊完成后，隊列自然就不為空了，通知下 notEmpty 就好了
    /**
     * Inserts element at current put position, advances, and signals.
     * Call only when holding lock.
     */
    private void enqueue(E x) {
        // assert lock.getHoldCount() == 1;
        // assert items[putIndex] == null;
        final Object[] items = this.items;
        items[putIndex] = x;
        if (++putIndex == items.length)
            putIndex = 0;
        count++;
        // 我已放入一個元素，不為空了
        notEmpty.signal();
    }
    // 與 take 對應，出隊完成后，自然就不可能是滿的了，至少一個空餘空間。
    /**
     * Extracts element at current take position, advances, and signals.
     * Call only when holding lock.
     */
    private E dequeue() {
        // assert lock.getHoldCount() == 1;
        // assert items[takeIndex] != null;
        final Object[] items = this.items;
        @SuppressWarnings("unchecked")
        E x = (E) items[takeIndex];
        items[takeIndex] = null;
        if (++takeIndex == items.length)
            takeIndex = 0;
        count--;
        if (itrs != null)
            itrs.elementDequeued();
        // 我已移除一個元素，肯定沒有滿了，你們繼續放入吧
        notFull.signal();
        return x;
    }

　　是不是超級好理解。是的。不過，我們不是想看 ArrayBlockingQueue 是如何實現的，我們是要論清 wait/notify 是如何實現的。因為畢竟，他們不是一個鎖那麼簡單。

    // 三個鎖的關係，即 notEmpty, notFull 都是 ReentrantLock 的條件鎖，相當於是其子集吧
    /** Main lock guarding all access */
    final ReentrantLock lock;

    /** Condition for waiting takes */
    private final Condition notEmpty;

    /** Condition for waiting puts */
    private final Condition notFull;
    
    public ArrayBlockingQueue(int capacity, boolean fair) {
        if (capacity <= 0)
            throw new IllegalArgumentException();
        this.items = new Object[capacity];
        lock = new ReentrantLock(fair);
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
    }
    // lock.newCondition() 是什麼鬼？它是 AQS 中實現的 ConditionObject
    // java.util.concurrent.locks.ReentrantLock#newCondition
    public Condition newCondition() {
        return sync.newCondition();
    }
        // java.util.concurrent.locks.ReentrantLock.Sync#newCondition
        final ConditionObject newCondition() {
            // AQS 中定義
            return new ConditionObject();
        }

　　接下來，我們要帶着幾個疑問來看這個 Condition 的對象：

　　　　1. 它的 wait/notify 是如何實現的？
　　　　2. 它是如何與互相進行聯繫的？
　　　　3. 為什麼 wait/notify 必須要在外面的lock獲取之後才能執行？
　　　　4. 它與Object的wait/notify 有什麼相同和不同點？

　　能夠回答了上面的問題，基本上對其原理與實現也就理解得差不多了。

重點1. wait/notify 是如何實現的？

　　我們從上面可以看到，它是通過調用 await()/signal() 實現的，到底做事如何，且看下面。

        // java.util.concurrent.locks.AbstractQueuedSynchronizer.ConditionObject#await()
        /**
         * Implements interruptible condition wait.
         * <ol>
         * <li> If current thread is interrupted, throw InterruptedException.
         * <li> Save lock state returned by {@link #getState}.
         * <li> Invoke {@link #release} with saved state as argument,
         *      throwing IllegalMonitorStateException if it fails.
         * <li> Block until signalled or interrupted.
         * <li> Reacquire by invoking specialized version of
         *      {@link #acquire} with saved state as argument.
         * <li> If interrupted while blocked in step 4, throw InterruptedException.
         * </ol>
         */
        public final void await() throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            // 添加當前線程到 等待線程隊列中，有 lastWaiter/firstWaiter 維護
            Node node = addConditionWaiter();
            // 釋放當前lock中持有的鎖，詳情且看下文
            int savedState = fullyRelease(node);
            // 從以下開始，將不再保證線程安全性，因為當前的鎖已經釋放，其他線程將會重新競爭鎖使用
            int interruptMode = 0;
            // 循環判定，如果當前節點不在 sync 同步隊列中，那麼就反覆阻塞自己
            // 所以判斷是否在 同步隊列上，是很重要的
            while (!isOnSyncQueue(node)) {
                // 沒有在同步隊列，阻塞
                LockSupport.park(this);
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
            }
            // 當條件被滿足后，需要重新競爭鎖，詳情看下文
            // 競爭到鎖后，原樣返回到 wait 的原點，繼續執行業務邏輯
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            // 下面是異常處理，忽略
            if (node.nextWaiter != null) // clean up if cancelled
                unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
        }
    /**
     * Invokes release with current state value; returns saved state.
     * Cancels node and throws exception on failure.
     * @param node the condition node for this wait
     * @return previous sync state
     */
    final int fullyRelease(Node node) {
        boolean failed = true;
        try {
            int savedState = getState();
            // 預期的，都是釋放鎖成功，如果失敗，說明當前線程並並未獲取到鎖，引發異常
            if (release(savedState)) {
                failed = false;
                return savedState;
            } else {
                throw new IllegalMonitorStateException();
            }
        } finally {
            if (failed)
                node.waitStatus = Node.CANCELLED;
        }
    }
    /**
     * Releases in exclusive mode.  Implemented by unblocking one or
     * more threads if {@link #tryRelease} returns true.
     * This method can be used to implement method {@link Lock#unlock}.
     *
     * @param arg the release argument.  This value is conveyed to
     *        {@link #tryRelease} but is otherwise uninterpreted and
     *        can represent anything you like.
     * @return the value returned from {@link #tryRelease}
     */
    public final boolean release(int arg) {
        // tryRelease 由客戶端自定義實現
        if (tryRelease(arg)) {
            Node h = head;
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }
    
    // 如何判定當前線程是否在同步隊列中或者可以進行同步隊列？
    /**
     * Returns true if a node, always one that was initially placed on
     * a condition queue, is now waiting to reacquire on sync queue.
     * @param node the node
     * @return true if is reacquiring
     */
    final boolean isOnSyncQueue(Node node) {
        // 如果上一節點還沒有被移除，當前節點就不能被加入到同步隊列
        if (node.waitStatus == Node.CONDITION || node.prev == null)
            return false;
        // 如果當前節點的下游節點已經存在，則它自身必定已經被移到同步隊列中
        if (node.next != null) // If has successor, it must be on queue
            return true;
        /*
         * node.prev can be non-null, but not yet on queue because
         * the CAS to place it on queue can fail. So we have to
         * traverse from tail to make sure it actually made it.  It
         * will always be near the tail in calls to this method, and
         * unless the CAS failed (which is unlikely), it will be
         * there, so we hardly ever traverse much.
         */
         // 最終直接從同步隊列中查找，如果找到，則自身已經在同步隊列中
        return findNodeFromTail(node);
    }

    /**
     * Returns true if node is on sync queue by searching backwards from tail.
     * Called only when needed by isOnSyncQueue.
     * @return true if present
     */
    private boolean findNodeFromTail(Node node) {
        Node t = tail;
        for (;;) {
            if (t == node)
                return true;
            if (t == null)
                return false;
            t = t.prev;
        }
    }
    
    // 當條件被滿足后，需要重新競爭鎖，以保證外部的鎖語義，因為之前自己已經將鎖主動釋放
    // 這個鎖與 lock/unlock 時的一毛一樣，沒啥可講的
    // java.util.concurrent.locks.AbstractQueuedSynchronizer#acquireQueued
    /**
     * Acquires in exclusive uninterruptible mode for thread already in
     * queue. Used by condition wait methods as well as acquire.
     *
     * @param node the node
     * @param arg the acquire argument
     * @return {@code true} if interrupted while waiting
     */
    final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

　　總結一下 wait 的邏輯:

　　　　1. 前提：自身已獲取到外部鎖；
　　　　2. 將當前線程添加到 ConditionQueue 等待隊列中；
　　　　3. 釋放已獲取到的鎖；
　　　　4. 反覆檢查進入等待，直到當前節點被移動到同步隊列中；
　　　　5. 條件滿足被喚醒，重新競爭外部鎖，成功則返回，否則繼續阻塞；(外部鎖是同一個，這也是要求兩個對象必須存在依賴關係的原因)
　　　　6. wait前線程持有鎖，wait后線程持有鎖，沒有一點外部鎖變化；

重點2. 釐清了 wait, 接下來，我們看 signal() 通知喚醒的實現：

        // java.util.concurrent.locks.AbstractQueuedSynchronizer.ConditionObject#signal
        /**
         * Moves the longest-waiting thread, if one exists, from the
         * wait queue for this condition to the wait queue for the
         * owning lock.
         *
         * @throws IllegalMonitorStateException if {@link #isHeldExclusively}
         *         returns {@code false}
         */
        public final void signal() {
            // 只有獲取鎖的實例，才可以進行signal，否則你拿什麼去保證線程安全呢
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            Node first = firstWaiter;
            // 通知 firstWaiter 
            if (first != null)
                doSignal(first);
        }
        
        /**
         * Removes and transfers nodes until hit non-cancelled one or
         * null. Split out from signal in part to encourage compilers
         * to inline the case of no waiters.
         * @param first (non-null) the first node on condition queue
         */
        private void doSignal(Node first) {
            // 最多只轉移一個 節點
            do {
                if ( (firstWaiter = first.nextWaiter) == null)
                    lastWaiter = null;
                first.nextWaiter = null;
            } while (!transferForSignal(first) &&
                     (first = firstWaiter) != null);
        }
    // 將一個節點從 等待隊列 移動到 同步隊列中，即可參与下一輪競爭
    // 只有確實移動成功才會返回 true
    // 說明：當前線程是持有鎖的線程
    // java.util.concurrent.locks.AbstractQueuedSynchronizer#transferForSignal
    /**
     * Transfers a node from a condition queue onto sync queue.
     * Returns true if successful.
     * @param node the node
     * @return true if successfully transferred (else the node was
     * cancelled before signal)
     */
    final boolean transferForSignal(Node node) {
        /*
         * If cannot change waitStatus, the node has been cancelled.
         */
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
            return false;

        /*
         * Splice onto queue and try to set waitStatus of predecessor to
         * indicate that thread is (probably) waiting. If cancelled or
         * attempt to set waitStatus fails, wake up to resync (in which
         * case the waitStatus can be transiently and harmlessly wrong).
         */
        // 同步隊列由 head/tail 指針維護
        Node p = enq(node);
        int ws = p.waitStatus;
        // 注意，此處正常情況下並不會喚醒等待線程，僅是將隊列轉移。 
        // 因為當前線程的鎖保護區域並未完成,完成后自然會喚醒其他等待線程
        // 否則將會存在當前線程任務還未執行完成，卻被其他線程搶了先去，那接下來的任務當如何？？
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
            LockSupport.unpark(node.thread);
        return true;
    }

　　總結一下，notify 的功能原理如下：

　　　　1. 前提：自身已獲取到外部鎖；
　　　　2. 轉移下一個等待隊列的節點到同步隊列中；
　　　　3. 如果遇到下一節點被取消情況，順延到再下一節點直到為空，至多轉移一個節點；
　　　　4. 正常情況下不做線程的喚醒操作；

　　所以，實現 wait/notify, 最關鍵的就是維護兩個隊列，等待隊列與同步隊列，而且都要求是在有外部鎖保證的情況下執行。

　　到此，我們也能回答一個問題：為什麼wait/notify一定要在鎖模式下才能運行？

　　因為wait是等待條件成立，此時必定存在競爭需要做保護，而它自身又必須釋放鎖以使外部條件可成立，且後續需要做恢復動作；而notify之後可能還有後續工作必須保障安全，notify只是鎖的一個子集。。。

四、通知所有線程的實現：notifyAll

　　有時條件成立后，可以允許所有線程通行，這時就可以進行 notifyAll, 那麼如果達到通知所有的目的呢？是一起通知還是？？

　　以下是 AQS 中的實現：

        // java.util.concurrent.locks.AbstractQueuedSynchronizer.ConditionObject#signalAll
        public final void signalAll() {
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            Node first = firstWaiter;
            if (first != null)
                doSignalAll(first);
        }
        /**
         * Removes and transfers all nodes.
         * @param first (non-null) the first node on condition queue
         */
        private void doSignalAll(Node first) {
            lastWaiter = firstWaiter = null;
            do {
                Node next = first.nextWaiter;
                first.nextWaiter = null;
                transferForSignal(first);
                first = next;
            } while (first != null);
        }

　　可以看到，它是通過遍歷所有節點，依次轉移等待隊列到同步隊列（通知）的，原本就沒有人能同時干幾件事的！

　　本文從java實現的角度去解析同步鎖的原理與實現，但並不局限於java。道理總是相通的，只是像操作系統這樣的大佬，能幹的活更純粹：比如讓cpu根本不用調度一個線程。

本站聲明:網站內容來源於博客園,如有侵權,請聯繫我們,我們將及時處理【其他文章推薦】

※網頁設計公司推薦更多不同的設計風格，搶佔消費者視覺第一線

※廣告預算用在刀口上，網站設計公司幫您達到更多曝光效益

※自行創業 缺乏曝光? 下一步"網站設計"幫您第一時間規劃公司的門面形象