C# 並行編程 之 限制資源的並發訪問 使用SemaphoreSlim

C# 並行編程 之 限制資源的並發訪問 使用SemaphoreSlim

概要

當多個任務或線程並行運行時，難以避免的對某些有限的資源進行並發的訪問。可以考慮使用信號量來進行這方面的控制（System.Threading.Semaphore）是表示一個Windows內核的信號量對象。如果預計等待的時間較短，可以考慮使用SemaphoreSlim，它則帶來的開銷更小。

.NetFrameWork中的信號量通過跟蹤進入和離開的任務或線程來協調對資源的訪問。信號量需要知道資源的最大數量，當一個任務進入時，資源計數器會被減1，當計數器為0時，如果有任務訪問資源，它會被阻塞，直到有任務離開為止。

示例程序： 10個任務並行訪問3個資源

using System;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Diagnostics;

namespace Sample5_8_semaphoreslim
{
    class Program
    {
        private static int _TaskNum = 10;
        private static Task[] _Tasks;
        private const int MAX_RESOURCE = 3;
        private const int RUN_LOOP = 10;

        private static SemaphoreSlim m_Semaphore;

        private static void Work1(int TaskID)
        {
            int i = 0;
            var sw = Stopwatch.StartNew();
            var rnd = new Random();

            while (i < RUN_LOOP)
            {
                Thread.Sleep(rnd.Next(200, 500));

                Console.WriteLine("TASK " + TaskID + " REQUESTing {");
                m_Semaphore.Wait();

                try
                {
                    Console.WriteLine("TASK " + TaskID + " WOrking  ...  ..." + i);
                    sw.Restart();
                    Thread.Sleep(rnd.Next(200, 500));
                }
                finally
                {
                    Console.WriteLine("TASK " + TaskID + " REQUESTing }");
                    m_Semaphore.Release();
                    i++;
                }
            }
        }

        static void Main(string[] args)
        {
            _Tasks = new Task[_TaskNum];
            m_Semaphore = new SemaphoreSlim(MAX_RESOURCE);
            int i = 0;

            for (i = 0; i < _TaskNum; i++)
            {
                _Tasks[i] = Task.Factory.StartNew((num) =>
                {
                    var taskid = (int)num;
                    Work1(taskid);
                }, i);
            }

            var finalTask = Task.Factory.ContinueWhenAll(_Tasks, (tasks) =>
            {
                Task.WaitAll(_Tasks);
                Console.WriteLine("==========================================================");
                Console.WriteLine("All Phase is completed");
                Console.WriteLine("==========================================================");
            });

            try
            {
                finalTask.Wait();
            }
            catch (AggregateException aex)
            {
                Console.WriteLine("Task failed And Canceled" + aex.ToString());
            }
            finally
            {
                m_Semaphore.Dispose();
            }
            Console.ReadLine();
        }
    }
}

使用超時和取消

信號量當然不可能永久的阻塞在那裡。信號量也提供了超時處理機制。方法是在Wait函數中傳入一個超時等待時間 - Wait(int TIMEOUT)。當Wait返回值為false時表明它超時了。如果傳入了 -1，則表示無限期的等待。

程序示例：注意其中的m_Semaphore.Release();已經被注釋掉了，任務會等待1秒鐘然後超時。

using System;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Diagnostics;

namespace Sample5_8_semaphoreslim
{
    class Program
    {
        private static int _TaskNum = 10;
        private static Task[] _Tasks;
        private const int MAX_RESOURCE = 3;
        private const int RUN_LOOP = 10;

        private static SemaphoreSlim m_Semaphore;

        private static void Work1(int TaskID)
        {
            int i = 0;
            var sw = Stopwatch.StartNew();
            var rnd = new Random();

            while (i < RUN_LOOP)
            {
                Thread.Sleep(rnd.Next(200, 500));

                Console.WriteLine("TASK " + TaskID + " REQUESTing {");
                if (!m_Semaphore.Wait(1000))
                {
                    Console.WriteLine("TASK " + TaskID + " TIMEOUT!!!");
                    return;
                }

                try
                {
                    Console.WriteLine("TASK " + TaskID + " WOrking  ...  ..." + i);
                    sw.Restart();
                    Thread.Sleep(rnd.Next(2000, 5000));
                }
                finally
                {
                    Console.WriteLine("TASK " + TaskID + " REQUESTing }");
                    //m_Semaphore.Release();
                    i++;
                }
            }
        }


        static void Main(string[] args)
        {
            _Tasks = new Task[_TaskNum];
            m_Semaphore = new SemaphoreSlim(MAX_RESOURCE);
            int i = 0;

            for (i = 0; i < _TaskNum; i++)
            {
                _Tasks[i] = Task.Factory.StartNew((num) =>
                {
                    var taskid = (int)num;
                    Work1(taskid);
                }, i);
            }

            var finalTask = Task.Factory.ContinueWhenAll(_Tasks, (tasks) =>
            {
                Task.WaitAll(_Tasks);
                Console.WriteLine("==========================================================");
                Console.WriteLine("All Phase is completed");
                Console.WriteLine("==========================================================");
            });

            try
            {
                finalTask.Wait();
            }
            catch (AggregateException aex)
            {
                Console.WriteLine("Task failed And Canceled" + aex.ToString());
            }
            finally
            {
                m_Semaphore.Dispose();
            }
            Console.ReadLine();
        }
    }
}

跨進程或AppDomain的同步

如果需要有跨進程或AppDomain的同步時，可以考慮使用Semaphore。Semaphore是取得的Windows 內核的信號量，所以在整個系統中是有效的。
它主要的接口時 Release和WaitOne，使用的方式和SemaphoreSlim是一致的。