快捷導(dǎo)航

C#并行編程之信號(hào)量

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

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

一：CountdownEvent

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

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

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

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

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

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

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

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

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

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

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

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

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

    //加載product需要8個(gè)任務(wù)
    var productTaskCount = 8;

    //重置信號(hào)
    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個(gè)任務(wù)
    var orderTaskCount = 12;

    //重置信號(hào)
    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("當(dāng)前任務(wù):{0}正在加載User部分?jǐn)?shù)據(jù)！", obj);
    }
    finally
    {
        cde.Signal();
    }
}

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

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

二：SemaphoreSlim

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

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("當(dāng)前時(shí)間:{0}任務(wù) {1}已經(jīng)進(jìn)入。", DateTime.Now, obj);

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

    slim.Release();
}

同樣，防止死鎖的情況，我們需要知道”超時(shí)和取消標(biāo)記“的解決方案，像SemaphoreSlim這種定死的”線程請(qǐng)求范圍“，其實(shí)是降低了擴(kuò)展性，所以說，試水有風(fēng)險(xiǎn)，使用需謹(jǐn)慎，在覺得有必要的時(shí)候使用它。

三： ManualResetEventSlim

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

//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("當(dāng)前時(shí)間:{0}我是主線程{1}，你們這些任務(wù)都等2s執(zhí)行吧：\n",
     DateTime.Now,
     Thread.CurrentThread.ManagedThreadId);
     Thread.Sleep(2000);

     mrs.Set();
 }

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

     Console.WriteLine("當(dāng)前時(shí)間:{0}任務(wù) {1}已經(jīng)進(jìn)入。", DateTime.Now, obj);
 }