C#并行編程之信號量

更新時間：2022年05月09日 14:47:52 作者：springsnow

這篇文章介紹了C#并行編程之信號量，文中通過示例代碼介紹的非常詳細。對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下

一：CountdownEvent

這種采用信號狀態(tài)的同步基元非常適合在動態(tài)的fork,join的場景，它采用“信號計數(shù)”的方式，就比如這樣，一個麻將桌只能容納4個人打麻將，如果后來的人也想搓一把碰碰運氣，那么他必須等待直到麻將桌上的人走掉一位。好，這就是簡單的信號計數(shù)機制，從技術角度上來說它是定義了最多能夠進入關鍵代碼的線程數(shù)。

但是CountdownEvent更牛X之處在于我們可以動態(tài)的改變“信號計數(shù)”的大小，比如一會兒能夠容納8個線程，一下又4個，一下又10個，這樣做有什么好處呢？還是承接上一篇文章所說的，比如一個任務需要加載1w條數(shù)據(jù)，那么可能出現(xiàn)這種情況。

加載User表：根據(jù)user表的數(shù)據(jù)量，我們需要開5個task。

加載Product表：產(chǎn)品表數(shù)據(jù)相對比較多，計算之后需要開8個task。

加載order表：由于我的網(wǎng)站訂單豐富，計算之后需要開12個task。

先前的文章也說了，我們需要協(xié)調(diào)task在多階段加載數(shù)據(jù)的同步問題，那么如何應對這里的5，8，12，幸好，CountdownEvent給我們提供了可以動態(tài)修改的解決方案。

我們看到有兩個主要方法：Wait和Signal。每調(diào)用一次Signal相當于麻將桌上走了一個人，直到所有人都搓過麻將wait才給放行，這里同樣要注意也就是“超時“問題的存在性，尤其是在并行計算中，輕量級別給我們提供了”取消標記“的機制，這是在重量級別中不存在的，比如下面的重載public bool Wait(int millisecondsTimeout, CancellationToken cancellationToken)，具體使用可以看前一篇文章的介紹。

//默認的容納大小為“硬件線程“數(shù)
static CountdownEvent cde = new CountdownEvent(Environment.ProcessorCount);

static void Main(string[] args)
{
    //加載User表需要5個任務
    var userTaskCount = 5;

    //重置信號
    cde.Reset(userTaskCount);

    for (int i = 0; i < userTaskCount; i++)
    {
        Task.Factory.StartNew((obj) =>
        {
            LoadUser(obj);
        }, i);
    }

    //等待所有任務執(zhí)行完畢
    cde.Wait();

    Console.WriteLine("\nUser表數(shù)據(jù)全部加載完畢！\n");

    //加載product需要8個任務
    var productTaskCount = 8;

    //重置信號
    cde.Reset(productTaskCount);

    for (int i = 0; i < productTaskCount; i++)
    {
        Task.Factory.StartNew((obj) =>
        {
            LoadProduct(obj);
        }, i);
    }

    cde.Wait();

    Console.WriteLine("\nProduct表數(shù)據(jù)全部加載完畢！\n");

    //加載order需要12個任務
    var orderTaskCount = 12;

    //重置信號
    cde.Reset(orderTaskCount);

    for (int i = 0; i < orderTaskCount; i++)
    {
        Task.Factory.StartNew((obj) =>
        {
            LoadOrder(obj);
        }, i);
    }

    cde.Wait();

    Console.WriteLine("\nOrder表數(shù)據(jù)全部加載完畢！\n");

    Console.WriteLine("\n(*^__^*) 嘻嘻，恭喜你，數(shù)據(jù)全部加載完畢\n");

    Console.Read();
}

static void LoadUser(object obj)
{
    try
    {
        Console.WriteLine("當前任務:{0}正在加載User部分數(shù)據(jù)！", obj);
    }
    finally
    {
        cde.Signal();
    }
}

static void LoadProduct(object obj)
{
    try
    {
        Console.WriteLine("當前任務:{0}正在加載Product部分數(shù)據(jù)！", obj);
    }
    finally
    {
        cde.Signal();
    }
}

static void LoadOrder(object obj)
{
    try
    {
        Console.WriteLine("當前任務:{0}正在加載Order部分數(shù)據(jù)！", obj);
    }
    finally
    {
        cde.Signal();
    }
}

二：SemaphoreSlim

在.net 4.0之前，framework中有一個重量級的Semaphore，人家可以跨進程同步，咋輕量級不行，msdn對它的解釋為：限制可同時訪問某一資源或資源池的線程數(shù)。關于它的重量級demo，我的上一個系列有演示，你也可以理解為CountdownEvent是 SemaphoreSlim的功能加強版，好了，舉一個輕量級使用的例子。

static SemaphoreSlim slim = new SemaphoreSlim(Environment.ProcessorCount, 12);

static void Main(string[] args)
{
    for (int i = 0; i < 12; i++)
    {
        Task.Factory.StartNew((obj) =>
        {
            Run(obj);
        }, i);
    }

    Console.Read();
}

static void Run(object obj)
{
    slim.Wait();

    Console.WriteLine("當前時間:{0}任務 {1}已經(jīng)進入。", DateTime.Now, obj);

    //這里busy3s中
    Thread.Sleep(3000);

    slim.Release();
}

同樣，防止死鎖的情況，我們需要知道”超時和取消標記“的解決方案，像SemaphoreSlim這種定死的”線程請求范圍“，其實是降低了擴展性，所以說，試水有風險，使用需謹慎，在覺得有必要的時候使用它。

三： ManualResetEventSlim

相信它的重量級別大家都知道是ManualReset，而這個輕量級別采用的是"自旋等待“+”內(nèi)核等待“，也就是說先采用”自旋等待的方式“等待，直到另一個任務調(diào)用set方法來釋放它。如果遲遲等不到釋放，那么任務就會進入基于內(nèi)核的等待，所以說如果我們知道等待的時間比較短，采用輕量級的版本會具有更好的性能，原理大概就這樣，下面舉個小例子。

//2047：自旋的次數(shù)
 static ManualResetEventSlim mrs = new ManualResetEventSlim(false, 2047);

 static void Main(string[] args)
 {

     for (int i = 0; i < 12; i++)
     {
         Task.Factory.StartNew((obj) =>
         {
             Run(obj);
         }, i);
     }

     Console.WriteLine("當前時間:{0}我是主線程{1}，你們這些任務都等2s執(zhí)行吧：\n",
     DateTime.Now,
     Thread.CurrentThread.ManagedThreadId);
     Thread.Sleep(2000);

     mrs.Set();
 }

 static void Run(object obj)
 {
     mrs.Wait();

     Console.WriteLine("當前時間:{0}任務 {1}已經(jīng)進入。", DateTime.Now, obj);
 }