詳細(xì)聊聊C#的并發(fā)機(jī)制優(yōu)秀在哪

更新時(shí)間：2022年02月08日 12:11:10 作者：beyondma

并發(fā)其實(shí)是一個(gè)很泛的概念,字面意思就是"同時(shí)做多件事",不過方式有所不同,下面這篇文章主要給大家介紹了關(guān)于C#并發(fā)機(jī)制的相關(guān)資料,文中通過實(shí)例代碼介紹的非常詳細(xì),需要的朋友可以參考下

前言

筆者上次用C#寫.Net代碼差不多還是10多年以前，由于當(dāng)時(shí)Java已經(jīng)頗具王者風(fēng)范，Net幾乎被打得潰不成軍。因此當(dāng)時(shí)筆者對(duì)于這個(gè).Net的項(xiàng)目態(tài)度比較敷衍了事，沒有對(duì)其中一些優(yōu)秀機(jī)制有很深的了解，在去年寫《C和Java沒那么香了，高并發(fā)時(shí)代誰能稱王》時(shí)都沒給.Net以一席之地，不過最近恰好機(jī)緣巧合，我又接手了一個(gè)Windows方面的項(xiàng)目，這也讓我有機(jī)會(huì)重新審視一下自己關(guān)于.Net框架的相關(guān)知識(shí)。

項(xiàng)目原型要實(shí)現(xiàn)的功能并不復(fù)雜，主要就是記錄移動(dòng)存儲(chǔ)設(shè)備中文件拷出的記錄，而且需要盡可能少的占用系統(tǒng)資源，而在開發(fā)過程中我無意中加了一行看似沒有任何效果的代碼，使用Invoke方法記錄文件拷出情況，這樣的操作卻讓程序執(zhí)行效率明顯會(huì)更高，這背后的原因特別值得總結(jié)。

一行沒用的代碼卻提高了效率？

由于我需要記錄的文件拷出信息并沒有回顯在UI的需要，因此也就沒考慮并發(fā)沖突的問題，在最初版本的實(shí)現(xiàn)中，我對(duì)于filesystemwatcher的回調(diào)事件，都是直接處理的，如下：

private void DeleteFileHandler(object sender, FileSystemEventArgs e)
        {
            if(files.Contains(e.FullPath))
            {
                files.Remove(e.FullPath);
               //一些其它操作
            }
        }

這個(gè)程序的處理效率在普通的辦公PC上如果同時(shí)拷出20個(gè)文件，那么在拷貝過程中，U盤監(jiān)測(cè)程序的CPU使用率大約是0.7%。

但是一個(gè)非常偶然的機(jī)會(huì)，我使用了Event/Delegate的Invoke機(jī)制，結(jié)果發(fā)現(xiàn)這樣一個(gè)看似的廢操作，卻讓程序的CPU占用率下降到0.2%左右

 private void UdiskWather_Deleted(object sender, FileSystemEventArgs e)
        {
            if(this.InvokeRequired)
            {
                this.Invoke(new DeleteDelegate(DeleteFileHandler), new object[] { sender,e });               }
            else
            {
                DeleteFileHandler(sender, e);
            }
        }

在我最初的認(rèn)識(shí)中.net中的Delegate機(jī)制在調(diào)用過程中是要進(jìn)行拆、裝箱操作的，因此這不拖慢操作就不錯(cuò)了，但實(shí)際的驗(yàn)證結(jié)果卻相反。

? 看似沒用的Invoke到底有什么用

這里先給出結(jié)論，Invoke能提升程序執(zhí)行效率，其關(guān)鍵還是在于線程在多核之間切換的消耗要遠(yuǎn)遠(yuǎn)高于拆、裝箱的資源消耗，我們知道我們程序的核心就是操作files這個(gè)共享變量，每次在被檢測(cè)的U盤目錄中如果發(fā)生文件變動(dòng)，其回調(diào)通知函數(shù)可能都運(yùn)行在不同的線程，如下：

Invoke機(jī)制的背后其實(shí)就是保證所有對(duì)于files這個(gè)共享變量的操作，全部都是由一個(gè)線程執(zhí)行完成的。

目前.Net的代碼都開源的，下面我們大致講解一下Invoke的調(diào)用過程，不管是BeginInvoke還是Invoke背后其實(shí)都是調(diào)用的MarshaledInvoke方法來完成的，如下：

public IAsyncResult BeginInvoke(Delegate method, params Object[] args) {
            using (new MultithreadSafeCallScope()) {
                Control marshaler = FindMarshalingControl();
                return(IAsyncResult)marshaler.MarshaledInvoke(this, method, args, false);
            }
        }

MarshaledInvoke的主要工作是創(chuàng)建ThreadMethodEntry對(duì)象，并把它放在一個(gè)鏈表里進(jìn)行管理，然后調(diào)用PostMessage將相關(guān)信息發(fā)給要通信的線程，如下：

?
private Object MarshaledInvoke(Control caller, Delegate method, Object[] args, bool synchronous) {
            if (!IsHandleCreated) {
                throw new InvalidOperationException(SR.GetString(SR.ErrorNoMarshalingThread));
            }
 
            ActiveXImpl activeXImpl = (ActiveXImpl)Properties.GetObject(PropActiveXImpl);
            if (activeXImpl != null) {
                IntSecurity.UnmanagedCode.Demand();
            }
 
            // We don't want to wait if we're on the same thread, or else we'll deadlock.
 
            // It is important that syncSameThread always be false for asynchronous calls.
 
            //
 
            bool syncSameThread = false;
            int pid; // ignored
            if (SafeNativeMethods.GetWindowThreadProcessId(new HandleRef(this, Handle), out pid) == SafeNativeMethods.GetCurrentThreadId()) {
                if (synchronous)
                    syncSameThread = true;
            }
 
            // Store the compressed stack information from the thread that is calling the Invoke()
            // so we can assign the same security context to the thread that will actually execute
            // the delegate being passed.
            //
 
            ExecutionContext executionContext = null;
            if (!syncSameThread) {
                executionContext = ExecutionContext.Capture();
            }
 
            ThreadMethodEntry tme = new ThreadMethodEntry(caller, this, method, args, synchronous, executionContext);
            lock (this) {
                if (threadCallbackList == null) {
                    threadCallbackList = new Queue();
                }
            }
 
            lock (threadCallbackList) {
                if (threadCallbackMessage == 0) {
                    threadCallbackMessage = SafeNativeMethods.RegisterWindowMessage(Application.WindowMessagesVersion + "_ThreadCallbackMessage");
                }
                threadCallbackList.Enqueue(tme);
            }
 
            if (syncSameThread) {
                InvokeMarshaledCallbacks();
            }  else {
                //
                UnsafeNativeMethods.PostMessage(new HandleRef(this, Handle), threadCallbackMessage, IntPtr.Zero, IntPtr.Zero);
            }
 
            if (synchronous) {
                if (!tme.IsCompleted) {
                    WaitForWaitHandle(tme.AsyncWaitHandle);
                }
 
                if (tme.exception != null) {
                    throw tme.exception;
                }
                return tme.retVal;
            }
            else {
                return(IAsyncResult)tme;
            }
        }

Invoke的機(jī)制就保證了一個(gè)共享變量只能由一個(gè)線程維護(hù)，這和GO語言使用通信來替代共享內(nèi)存的設(shè)計(jì)是暗合的，他們的理念都是 "讓同一塊內(nèi)存在同一時(shí)間內(nèi)只被一個(gè)線程操作" 。這和現(xiàn)代計(jì)算體系結(jié)構(gòu)的多核CPU（SMP)有著密不可分的聯(lián)系，

這里我們先來科普一下CPU之間的通信MESI協(xié)議的內(nèi)容。我們知道現(xiàn)代的CPU都配備了高速緩存，按照多核高速緩存同步的MESI協(xié)議約定，每個(gè)緩存行都有四個(gè)狀態(tài)，分別是E（exclusive）、M（modified）、S（shared）、I（invalid），其中：

M：代表該緩存行中的內(nèi)容被修改，并且該緩存行只被緩存在該CPU中。這個(gè)狀態(tài)代表緩存行的數(shù)據(jù)和內(nèi)存中的數(shù)據(jù)不同。

E：代表該緩存行對(duì)應(yīng)內(nèi)存中的內(nèi)容只被該CPU緩存，其他CPU沒有緩存該緩存對(duì)應(yīng)內(nèi)存行中的內(nèi)容。這個(gè)狀態(tài)的緩存行中的數(shù)據(jù)與內(nèi)存的數(shù)據(jù)一致。

I：代表該緩存行中的內(nèi)容無效。

S:該狀態(tài)意味著數(shù)據(jù)不止存在本地CPU緩存中，還存在其它CPU的緩存中。這個(gè)狀態(tài)的數(shù)據(jù)和內(nèi)存中的數(shù)據(jù)也是一致的。不過只要有CPU修改該緩存行都會(huì)使該行狀態(tài)變成 I 。

四種狀態(tài)的狀態(tài)轉(zhuǎn)移圖如下：

?我們上文也提到了，不同的線程是有大概率是運(yùn)行在不同CPU核上的，在不同CPU操作同一塊內(nèi)存時(shí)，站在CPU0的角度上看，就是CPU1會(huì)不斷發(fā)起remote write的操作，這會(huì)使該高速緩存的狀態(tài)總是會(huì)在S和I之間進(jìn)行狀態(tài)遷移，而一旦狀態(tài)變?yōu)镮將耗費(fèi)比較多的時(shí)間進(jìn)行狀態(tài)同步。

因此我們可以基本得出 this.Invoke(new DeleteDelegate(DeleteFileHandler), new object[] { sender,e }); ;這行看似無關(guān)緊要的代碼之后，無意中使files共享變量的維護(hù)操作，由多核多線程共同操作，變成了眾多子線程向主線程通信，所有維護(hù)操作均由主線程進(jìn)行，這也使最終的執(zhí)行效率有所提高。

?深度解讀，為何要加兩把鎖

在當(dāng)前使用通信替代共享內(nèi)存的大潮之下，鎖其實(shí)是最重要的設(shè)計(jì)。

我們看到在.Net的Invoke實(shí)現(xiàn)中，使用了兩把鎖lock (this) 與lock (threadCallbackList)。

lock (this) {
                if (threadCallbackList == null) {
                    threadCallbackList = new Queue();
                }
            }
 
            lock (threadCallbackList) {
                if (threadCallbackMessage == 0) {
                    threadCallbackMessage = SafeNativeMethods.RegisterWindowMessage(Application.WindowMessagesVersion + "_ThreadCallbackMessage");
                }
                threadCallbackList.Enqueue(tme);
            }

在.NET當(dāng)中l(wèi)ock關(guān)鍵字的基本可以理解為提供了一個(gè)近似于CAS的鎖（Compare And Swap）。CAS的原理不斷地把"期望值"和"實(shí)際值"進(jìn)行比較，當(dāng)它們相等時(shí)，說明持有鎖的CPU已經(jīng)釋放了該鎖，那么試圖獲取這把鎖的CPU就會(huì)嘗試將"new"的值(0)寫入"p"（交換），以表明自己成為spinlock新的owner。偽代碼演示如下：

void CAS(int p, int old,int new)
{
    if *p != old
        do nothing
    else 
     *p ← new
}

基于CAS的鎖效率沒問題，尤其是在沒有多核競(jìng)爭(zhēng)的情況CAS表現(xiàn)得尤其優(yōu)秀，但CAS最大的問題就是不公平，因?yàn)槿绻卸鄠€(gè)CPU同時(shí)在申請(qǐng)一把鎖，那么剛剛釋放鎖的CPU極可能在下一輪的競(jìng)爭(zhēng)中獲取優(yōu)勢(shì)，再次獲得這把鎖，這樣的結(jié)果就是一個(gè)CPU忙死，而其它CPU卻很閑，我們很多時(shí)候詬病多核SOC“一核有難，八核圍觀”其實(shí)很多時(shí)候都是由這種不公平造成的。

為了解決CAS的不公平問題，業(yè)界大神們又引入了TAS（Test And Set Lock）機(jī)制，個(gè)人感覺還是把TAS中的T理解為Ticket更好記一些，TAS方案中維護(hù)了一個(gè)請(qǐng)求該鎖的頭尾索引值，由"head"和"tail"兩個(gè)索引組成。

struct lockStruct{
    int32 head;
    int32 tail;
} ;

"head"代表請(qǐng)求隊(duì)列的頭部，"tail"代表請(qǐng)求隊(duì)列的尾部，其初始值都為0。

最一開始時(shí)，第一個(gè)申請(qǐng)的CPU發(fā)現(xiàn)該隊(duì)列的tail值是0,那么這個(gè)CPU會(huì)直接獲取這把鎖，并會(huì)把tail值更新為1，并在釋放該鎖時(shí)將head值更新為1。

在一般情況下當(dāng)鎖被持有的CPU釋放時(shí)，該隊(duì)列的head值會(huì)被加1，當(dāng)其他CPU在試圖獲取這個(gè)鎖時(shí)，鎖的tail值獲取到，然后把這個(gè)tail值加1，并存儲(chǔ)在自己專屬的寄存器當(dāng)中，然后再把更新后的tail值更新到隊(duì)列的tail當(dāng)中。接下來就是不斷地循環(huán)比較，判斷該鎖當(dāng)前的"head"值，是否和自己存儲(chǔ)在寄存器中的"tail"值相等，相等時(shí)則代表成功獲得該鎖。

TAS這類似于用戶到政務(wù)大廳去辦事時(shí)，首先要在叫號(hào)機(jī)取號(hào)，當(dāng)工作人員廣播叫到的號(hào)碼與你手中的號(hào)碼一致時(shí)，你就獲取了辦事柜臺(tái)的所有權(quán)。

但是TAS卻存在一定的效率問題，根據(jù)我們上文介紹的MESI協(xié)議，這個(gè)lock的頭尾索引其實(shí)是在各個(gè)CPU之間共享的，因此tail和head頻繁更新，還是會(huì)引發(fā)調(diào)整緩存不停的invalidate，這會(huì)極大的影響效率。

因此我們看到在.Net的實(shí)現(xiàn)中干脆就直接引入了threadCallbackList的隊(duì)列，并不斷將tme(ThreadMethodEntry)加入隊(duì)尾，而接收消息的進(jìn)程，則不斷從隊(duì)首獲取消息.

lock (threadCallbackList) {
                if (threadCallbackMessage == 0) {
                    threadCallbackMessage = SafeNativeMethods.RegisterWindowMessage(Application.WindowMessagesVersion + "_ThreadCallbackMessage");
                }
                threadCallbackList.Enqueue(tme);
            }

當(dāng)隊(duì)首指向這個(gè)tme時(shí)，消息才被發(fā)送，其實(shí)是一種類似于MAS的實(shí)現(xiàn)，當(dāng)然MAS實(shí)際是為每個(gè)CPU都建立了一個(gè)專屬的隊(duì)列，和Invoke的設(shè)計(jì)略有不同，不過基本的思想是一致的。

很多時(shí)候年少時(shí)不是品不出很多東西背后味道的，這也讓我錯(cuò)過了很多非常值得總結(jié)的技術(shù)要點(diǎn)，因此在春節(jié)假期總結(jié)一下最近使用C#的心得，以饗讀者，順祝大家新春愉快！