關於高並發下多線程數據處理


一、Lock:

C#中關鍵字lock(VB.NET中SyncLock,等同於try+finally的Monitor.Enter……Monitor.Exit)。原理是“每次線程進入后鎖住當前所有的內存區塊等相關區域,由該線程自行處理完畢全部的線程后自動釋放”,接着其余線程搶先進入。

優點:最為大眾所知的一種多線程處理方法,最為普遍的解決方案。

缺點:無法完全適應高並發場合下處理需求——原因:每次讓大量線程在外邊排隊等候(因為一次只能一個線程處理lock區塊),而且外邊排隊的線程總是要在Windows操作系統、臨界區等上下文環境切換。(想一想:假設有一個廁所,每次一個人進去大便,因為有人你進不了,也不知道人家啥時候出來……所以尷尬的你只能到休息區去暫時休息打盹、看報、聊天……,等一陣然后再去看看廁所是否處於可用狀態……)。這里的“廁所”的門就是“臨界區”,“臨界區”里邊就是受保護的代碼,“廁所”和“休息區”是兩個不同的上下文地區,整個公司就是一個Windows操作系統。所以這種lock方式的上下文切換本身就是嘗試性地試探是否可以進入,不可以則上下文切換休眠。當然消耗資源也是大的——除非你可以“估計”到每個線程處理的時間差不多等於線程休眠從上下文切換的時間,而且外部排隊線程不是很多的情況下。

二、取代lock的“原子鎖”:

本質上是CPU級別的鎖,最小粒度的。用於保證一次性數據原子量的完成。在NET中有InterLock類可以提供最簡單的諸如Increment,Decrement等原子性操作方法。如果要實現共享代碼,完全可以用以下方法嘗試實現:

復制代碼
private volatile int _isFree = 0;

public void GeneralLockCode(Action act = null)
{
     try
     {
        if(InterLock.Exchange(ref _isFree, 1) == 0)  
        {
              if(act != null)
               act();
        }
     }
     finally
     {
           InterLock.Exchange(ref _isFree, 0);
     }
}
復制代碼

“_isFree”是一個類級別的變量,如果是0表示尚無線程使用act代碼(表示允許線程進來)。每次進入此代碼之后,有一個Exchange函數,這個函數是內存原子操作函數,它的作用是:1)把“1”賦值給_isFree。2)返回_isFree以前狀態的值(在這兩步中不允許其它線程干擾對_isFree進行變化)。

假設有2個線程(A和B)。A搶到了執行了Exchange方法,得到結果true;此時_isFree立馬變成了1。此時B線程再次執行Exchange,_isFree重新被賦值為1,但是返回了線程A時候處理的狀態1,因為1不會等於0,所以act方法不會執行——直到A執行了finally方法為止。

上面的代碼加以改進,就完全可以變成一個通用的,比lock更好一點的“排它鎖”:

復制代碼
private volatile int _isFree = 0;

public void GeneralLock(Action act)
{
      try
      {
            while(InterLock.Exchange(ref _isFree,1) == 1);
            act();
      }
      finally
      {
           InterLock.Exchange(ref _isFree,0);
      }
}
復制代碼

該鎖一目了然——也只允許一個線程進入,其余線程外面等待。但是該方法的好處在於——如果act方法足夠小(耗時小,時間短),那么當某個線程搶先進入的時候,其余的線程不是像Lock塊一樣上下文切換,而是直接在臨界區“候着”自旋,直到act被執行完畢,鎖標記釋放為止。之所以要“act耗時足夠小”,是因為這里如果高並發的話,沒有搶到機會執行act的線程不得不一直處於“自旋”狀態,空耗CPU資源。

我們可以通過嘗試在while里邊加Thread.Sleep,Sleep的毫秒可以隨着失敗的輪詢次數不斷試探性增長,或者干脆使用SpinWait或者Thread的SpinUntil代替while,以便於這個CAS的鎖更好地適用於一般高並發場合下的應用。

【方案1】

復制代碼
private volatile int _isFree = 0;

public void GeneralLock(Action act)
{
     SpinWait wait = new SpinWait();

      try
      {
             
            while(InterLock.Exchange(ref _isFree,1) == 1)                    
                  {wait.SpinOnce();}
   act();
      }
      finally
      {
           InterLock.Exchange(ref _isFree,0);
      }
}
復制代碼

【方案2】

復制代碼
private volatile int _isFree = 0;

public void GeneralLock(Action act)
{
      try
      {
           Thread.SpinUntil (() => InterLock.Exchange(ref _isFree,1) == 0);  //Until,直到isFree=0,即等到可用狀態。
           act();
      }
      finally
      {
           InterLock.Exchange(ref _isFree,0);
      }
}
復制代碼

三、CAS法則:

InternLock中有一個Compare And Swap原子操作,其函數是(以NET為主):

InternLock.CompareExchange(ref 原來變量,替代變量,比較值),函數結果返回“原來變量”前一次的值(被“替代變量”取代前的數值)。

根據這個函數,我們完全可以在把每個線程中拷貝自己的一份“原有變量”的值作為“比較值”,“替代變量”在“比較值”基礎上進行某些操作,然后使用這個CAS方法保證每次操作的時候都是原子性的,該函數做3件事情:

1)判斷“原來變量”是否等於“比較值”。

2)如果1)步返回true,則用“替代變量”替換“原來變量”。

3)返回“原有變量”前一次的值。

再次聲明,這個方法也是原子性的。意味着在這個函數的時候無論如何不會被其它線程干擾打斷,修改變量。那么我們輕而易舉可以得到結論——如果有A和B線程,同時修改變量a(假設a共用,初始值為1),那么A線程搶到進去(A線程對a做了拷貝,比較值為1,替代變量為2)。那么當A做CAS的時候,原來變量=比較值,然后把“替代變量”替換“原來變量”,之后返回之前的值1,1=A線程的比較值1,所以成功了。

假設A在執行CAS過程中(或者之前某些步驟),B線程也進來了,(B的比較值也為1,替代變量為2),此時A率先做了CAS,“原來變量”變成了2,此時B做CAS發現原來變量不等於比較值,因此不會進行替代。當然返回的結果也也是被A線程替代后的原來變量的值2,自然也不等於B線程的比較值。所以B線程不得不再次去搶——直到滿足條件為止。

類似這樣的方法,相對於之前講的“原子鎖”而言不存在空轉(因為他們每次都嘗試生產一個自己的方案,然后分別去搶;不像“原子鎖”中的while處於Sleep空轉或者忙轉情況),但是他們也是死循環嘗試探測是否可以搶到原子鎖的,所以仍然不排除CPU資源被大量占用的情況。所以也完全可以嘗試先判斷一次是否搶到,搶到直接退出循環;否則繼續搶或者進行優化的“自旋”,下面給出偽代碼結構:

復制代碼
volatile _globalVar = 初始化值;

public void GeneralCAS()
{
      聲明 _copyVar和_repVar;
do
           {
              _copyVar = _globalVar;
              _repVar = 在_globalVar基礎上改變或者新的值;
           }while(CAS(ref _globalVar,_repVar,_copyVar)!=_copyVar);
}
復制代碼

 最后給出一個多線程累加的例子:

復制代碼
public struct AtomicNumber
{
      private volatile int _number = 0;
      
      public int GetProcessedNumber (int stepToIncrease = 1)
      {
          int copyNum = _number,newNum = copyNum + stepToIncrease;
          for(;;copyNumber = _number,newNum = copyNum+stepToIncrease)
          {
                 if(Interlock.CompareExchange(ref _number, newNum, copyNum) == copyNum)
                 {
                       return newNum;
                 }
          }
      }
}
復制代碼


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM