.NET中保證線程安全的高級方法Interlocked類使用介紹

更新時間：2014年07月16日 09:38:21 投稿：junjie

這篇文章主要介紹了.NET中保證線程安全的高級方法Interlocked類使用介紹,Interlocked類可以為為多個線程共享的變量提供原子操作,需要的朋友可以參考下

說到線程安全，不要一下子就想到加鎖，尤其是可能會調(diào)用頻繁或者是要求高性能的場合。

對于性能要求不高或者同步的對象數(shù)量不多的時候，加鎖是一個比較簡單而且易于實現(xiàn)的選擇。比方說.NET提供的一些基礎類庫，比如線程安全的堆棧和隊列，如果使用加鎖的方式那么會使性能大打折扣（速度可能會降低好幾個數(shù)量級），而且如果設計得不好的話還有可能發(fā)生死鎖。

現(xiàn)在通過查看微軟的源代碼來學習一些不直接lock（等價于Monitor類）的線程同步技巧吧。

這里我們主要用的是Interlocked類，這個類按照M$的描述，是“為多個線程共享的變量提供原子操作”，當然這個類是一個靜態(tài)類。這個類的源代碼看不到，因為是調(diào)用的CLR內(nèi)部的方法，不過基本思想應該是通過硬件原語try and set來實現(xiàn)的。

該類提供的Add、Increment、Decrement能夠完成簡單的原子操作。

假如我們要提供一個計數(shù)器，每訪問一次就遞增地返回一個新的數(shù)值用于計數(shù)。在多線程環(huán)境下，s++這一條語句不是線程安全的。因為執(zhí)行這個語句要經(jīng)過：移到寄存器（讀?。⑦\算、寫入這幾個步驟，在任何時候都可能會切換到其他線程，這樣子s被多個線程訪問值可能會在切換的過程中丟失。有了Interlocked提供的這幾個原子操作的方法，就不用自己去加鎖實現(xiàn)這些簡單的運算了。由于是使用的硬件原語，其效率自然也比加鎖高得多。

但是大多數(shù)情況下，問題并沒有執(zhí)行相加相減運算那么簡單，這時如果不想用鎖的話就要想想辦法了。

以微軟的ConcurrentStack提供的線程安全的堆棧為例，分析一下如何實現(xiàn)如果往棧頭添加數(shù)據(jù)。

m_head是指向堆頂?shù)闹羔槪诙x的時候由于是多線程訪問的，所以要加上volatile修飾符：

復制代碼代碼如下:

private volatile Node m_head;

如果是單線程的，那么入棧語句就是下面這個樣子：

復制代碼代碼如下:

    Node newNode = new Node(item);

    newNode.m_next = m_head; 

    m_head = newNode;

假如有兩個線程并發(fā)訪問入棧方法的話，那么可能會產(chǎn)生如下情況：第一個線程執(zhí)行完第二條語句被打斷，第二個線程執(zhí)行到第二條語句又切換回第一個線程，兩個線程執(zhí)行完后有一個入棧的元素就不見了。那么如何實現(xiàn)線程安全呢？M$的代碼是這樣寫的：

復制代碼代碼如下:

Node newNode = new Node(item);
newNode.m_next = m_head;
if (Interlocked.CompareExchange(ref m_head, newNode, newNode.m_next) == newNode.m_next)
{
return;
}

// If we failed, go to the slow path and loop around until we succeed.
PushCore(newNode, newNode);

首先，Interlocked.CompareExchange比較兩個元素是否相等，并根據(jù)比較的結(jié)果替換其中一個元素，返回結(jié)果始終是第一個元素的原值。這個方法是原子操作。

那么這段代碼首先設置newNode的下一節(jié)點為堆棧頂部的元素，接下來CompareExchange，判斷棧頂元素有沒有被修改過。假如此時沒有另一個線程修改棧頂元素，那么m_head還是原來的值（上一條語句設置的新棧頂?shù)南乱粋€元素），此時就可以安全地把棧頂指針指向新元素，操作完成（return）。注意CompareExchange是原子操作的，所以在這期間棧頂元素不可能再被修改。

如果比較結(jié)果不相等，那么說明棧頂元素已經(jīng)被其他線程修改了（此時返回值就是被修改后的棧頂，和上一條語句設置m_next不一樣），這樣CompareExchange就不會修改m_head，說明入棧不成功，執(zhí)行PushCore方法。

這個東東的代碼如下：

復制代碼代碼如下:

        private void PushCore(Node head, Node tail)
        {
            SpinWait spin = new SpinWait();

            // Keep trying to CAS the exising head with the new node until we succeed.
            do
            {
                spin.SpinOnce();
                // Reread the head and link our new node.
                tail.m_next = m_head;
            }
            while (Interlocked.CompareExchange(
                ref m_head, head, tail.m_next) != tail.m_next);

#if !FEATURE_PAL && !FEATURE_CORECLR
            if (CDSCollectionETWBCLProvider.Log.IsEnabled())
            {
                CDSCollectionETWBCLProvider.Log.ConcurrentStack_FastPushFailed(spin.Count);
            }
#endif //!FEATURE_PAL && !FEATURE_CORECLR
        }

可以看到其邏輯還是和上面那個一樣，只是加了一個循環(huán)直到操作完成。在這期間使用了一個SpinWait對象和SpinOnce方法，那么我們又要了解一下這是干嘛的。

關(guān)于SpinWait對象，M$的說明是：System.Threading.SpinWait 是一個輕量同步類型，可以在低級別方案中使用它來避免內(nèi)核事件所需的高開銷的上下文切換和內(nèi)核轉(zhuǎn)換。

關(guān)于它的說明還有一堆，你可以參考這里。如果不想看那么多，那么只需了解它的使用場合是在資源不會被占用很長時間的時候進行等待，以用戶模式自旋以避免高額的開銷。

SpinOnce的說明很簡單，就是執(zhí)行單一自旋，可以理解為等待一個很短的時間。總的來說，當自旋此時達到5次時，會切換到同一處理器上的另一個線程，當達到20次時，會調(diào)用Thread的Sleep方法阻塞當前線程，此時可以切換到其他同優(yōu)先級或更高優(yōu)先級的線程上去。

這樣，就可以避免加鎖（lock free）的高昂代價來實現(xiàn)線程的同步。

但是有的時候我們不能保證線程安全。比如堆棧的Count屬性，在程序調(diào)用這個屬性后，我們并不能保證這個屬性返回的時候是正確的，在返回到應用程序的線程前元素數(shù)量是有可能變化的，因此我們也就只能保證我們的返回值曾經(jīng)正確。

不過顯而易見，這是可以接受的。對于開發(fā)者來說，假如我們要訪問一個多線程字典（ConcurrentDictionary）中的指定元素，我們不應該是先判斷是否為空再取元素（因為元素可能在這兩步操作之間被刪掉），而是應該使用TryGetValue這種保證線程安全的方法來進行操作。

您可能感興趣的文章: