.net - 学习实现线程池-使用autoresetevent时，信号事件丢失

Question

我坚信通过创新进行学习。考虑到这一点，我着手实现自定义线程池。我为自己设定的目标如下:

能够将工作项在线程池上排队。

能够处理具有固定线程数的工作项-所有这些线程都是同时创建的。

通用工作线程函数应该只知道如何进行双端队列操作，而不应该处理其他函数/属性，例如
IsEmpty或Count。

我成功实现了上述目标，但想验证我与Stackoverflow专家所采用的方法。另外，想了解是否有更好的方法，或者多线程专家将如何解决此问题。以下几段提到了我面临的挑战以及如何解决它。

我创建的线程池在内部维护了一个工作项队列，所有工作线程都从该队列中选择该项，然后对其进行处理。每当有新项目排队时，它将发出事件信号，以便任何空闲线程都可以选择并执行它。

我从autoresetevent开始，用信号通知队列中任何新工作项的等待线程，但是我遇到了丢失信号事件的问题。当一个以上的项目排队并且没有空闲线程来处理该项目时，就会发生这种情况。未处理的总项目与由于重叠的设置(信号)事件而丢失的总信号事件相同。

为了解决丢失的信号事件的问题，我在autoresetevent之上创建了一个包装器，并使用它来代替autoresetevent。它解决了这个问题。这是相同代码的 list :

public static class CustomThreadPool
{
    static CustomThreadPool()
    {
        for (int i = 0; i < minThreads; i++)
            _threads.Add(
                new Thread(ThreadFunc) { IsBackground = true }
                );

        _threads.ForEach((t) => t.Start());
    }

    public static void EnqueWork(Action action)
    {
        _concurrentQueue.Enqueue(action);
        _enqueEvent.Set();
    }

    private static void ThreadFunc()
    {
        Action action = null;
        while (true)
        {
            _enqueEvent.WaitOne();
            _concurrentQueue.TryDequeue(out action);
            action();
        }
    }

    private static ConcurrentQueue<Action> _concurrentQueue = new ConcurrentQueue<Action>();
    private static List<Thread> _threads = new List<Thread>();
    private static CountAutoResentEvent _enqueEvent = new CountAutoResentEvent();
    private static object _syncObject = new object();
    private const int minThreads = 4;
    private const int maxThreads = 10;

    public static void Test()
    {
        CustomThreadPool.EnqueWork(() => {

            for (int i = 0; i < 10; i++) Console.WriteLine(i);
            Console.WriteLine("****First*****");
        });

        CustomThreadPool.EnqueWork(() =>
        {
            for (int i = 0; i < 10; i++) Console.WriteLine(i);
            Console.WriteLine("****Second*****");
        });

        CustomThreadPool.EnqueWork(() =>
        {
            for (int i = 0; i < 10; i++) Console.WriteLine(i);
            Console.WriteLine("****Third*****");
        });

        CustomThreadPool.EnqueWork(() =>
        {
            for (int i = 0; i < 10; i++) Console.WriteLine(i);
            Console.WriteLine("****Fourth*****");
        });

        CustomThreadPool.EnqueWork(() =>
        {
            for (int i = 0; i < 10; i++) Console.WriteLine(i);
            Console.WriteLine("****Fifth*****");
        });
    }
}

public class CountAutoResentEvent
{
    public void Set()
    {
        _event.Set();
        lock (_sync)
            _countOfSet++;
    }

    public void WaitOne()
    {
        _event.WaitOne();
        lock (_sync)
        {
            _countOfSet--;
            if (_countOfSet > 0)
                _event.Set();
        }
    }

    private AutoResetEvent _event = new AutoResetEvent(false);
    private int _countOfSet = 0;
    private object _sync = new object();
}

现在，我有几个问题:

我的方法充分证明了吗？

什么同步机制最适合此问题，为什么？

多线程专家将如何处理此问题？

谢谢。

Answer 1

从我所见，我会说这是正确的。我喜欢您使用了ConcurrentQueue并且没有打算实现自己的同步队列。那是一团糟，很可能不会像现有的那样快。

我只想指出，您的自定义“信号机制”实际上非常类似于信号量:一种锁，该锁允许多个线程进入关键部分。该功能已经存在于Semaphore类中。