如何使用Monitor.Wait与Monitor.Pulse实现自定义Semaphore信号量

核心实现逻辑 #

• 维护线程安全的可用资源计数、最大资源上限
• 使用Monitor.Wait()实现无可用资源时的线程阻塞
• 使用Monitor.Pulse()实现资源释放后唤醒等待线程

自定义Semaphore实现代码 #

public class CustomSemaphore
{
    private readonly object _lockObj = new object();
    private int _currentCount;
    private readonly int _maxCount;

    public CustomSemaphore(int initialCount, int maxCount)
    {
        if (initialCount < 0 || initialCount > maxCount)
            throw new ArgumentOutOfRangeException(nameof(initialCount), "初始计数不能小于0，也不能大于最大计数");
        if (maxCount < 1)
            throw new ArgumentOutOfRangeException(nameof(maxCount), "最大计数不能小于1");
        _currentCount = initialCount;
        _maxCount = maxCount;
    }

    /// <summary>
    /// 阻塞等待直到有资源被释放
    /// </summary>
    public void Request()
    {
        lock (_lockObj)
        {
            // 循环判断防止虚假唤醒
            while (_currentCount == 0)
            {
                Monitor.Wait(_lockObj);
            }
            _currentCount--;
        }
    }

    /// <summary>
    /// 释放占用的资源
    /// </summary>
    public void Release()
    {
        lock (_lockObj)
        {
            if (_currentCount >= _maxCount)
                throw new SemaphoreFullException("释放资源后超过信号量最大容量");
            _currentCount++;
            // 唤醒一个等待的线程
            Monitor.Pulse(_lockObj);
        }
    }

    /// <summary>
    /// 返回当前可用资源的数量
    /// </summary>
    public int NumberAvailable()
    {
        lock (_lockObj)
        {
            return _currentCount;
        }
    }
}

调整后的业务代码 #

替换原有SemaphoreSlim为自定义CustomSemaphore即可，改动量极小：

class Program
{
    private const int InitialCount = 2;
    private const int MaximumCount = 8;

    static void Main()
    {
        var semaphoreObject = new CustomSemaphore(InitialCount, MaximumCount);
        var printerObject = new Printer();

        for (int i = 0; i < 20; ++i)
        {
            int j = i;
            Task.Factory.StartNew(() =>
            {
                Request(semaphoreObject);
                printerObject.Print(j);
                Release(semaphoreObject);
                NumberAvailable(semaphoreObject);
            });
        }
        Console.ReadLine();
    }

    private static void Request(CustomSemaphore s)
    {
        s.Request();
    }

    private static void Release(CustomSemaphore s)
    {
        s.Release();
    }

    private static void NumberAvailable(CustomSemaphore s)
    {
        Console.WriteLine($"Available resource: {s.NumberAvailable()}");
    }
}

class Printer
{
    public void Print(int documentToPrint)
    {
        Console.WriteLine("Printing document: " + documentToPrint);
        Thread.Sleep(TimeSpan.FromSeconds(5));
    }
}

注意事项 #

• Request方法中使用while循环判断可用资源数，是为了避免线程的虚假唤醒问题
• 所有涉及计数读写的操作都加了锁，保证多线程场景下计数的准确性
• 增加了参数合法性校验，避免非法输入导致的逻辑错误

内容的提问来源于stack exchange，提问作者user122222