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

1、Lock：

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

優勢：最爲大衆所知的一種多線程處理方法，最爲廣泛的解決方案。

缺點：沒法徹底適應高併發場合下處理需求——緣由：每次讓大量線程在外邊排隊等候（由於一次只能一個線程處理lock區塊），並且外邊排隊的線程老是要在Windows操做系統、臨界區等上下文環境切換。（想想：假設有一個廁所，每次一我的進去大便，由於有人你進不了，也不知道人家啥時候出來……因此尷尬的你只能到休息區去暫時休息打盹、看報、聊天……，等一陣而後再去看看廁所是否處於可用狀態……）。這裏的「廁所」的門就是「臨界區」，「臨界區」裏邊就是受保護的代碼，「廁所」和「休息區」是兩個不一樣的上下文地區，整個公司就是一個Windows操做系統。因此這種lock方式的上下文切換自己就是嘗試性地試探是否能夠進入，不能夠則上下文切換休眠。固然消耗資源也是大的——除非你能夠「估計」到每一個線程處理的時間差很少等於線程休眠從上下文切換的時間，並且外部排隊線程不是不少的狀況下。

2、取代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);
      }
}

3、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;
                 }
          }
      }
}