异步任务处理器

Question

我正在开发一个异步任务处理器。我需要高性能处理器，所以使用的同步原语应该尽可能低级。处理器应该持有一个线程，在没有任务时休眠，在任务出现时唤醒。任务处理和任务添加应该在不同的线程中执行。

我尝试使用AutoResetEvent 实现，但它有竞争条件：

public class Processor
{
    ConcurrentQueue<Action> _workItemQueue = new ConcurrentQueue<Action>();
    AutoResetEvent _newWorkItemAutoResetEvent = new AutoResetEvent(false);
    private bool _disposed;
    Thread _thread;

    public void Do(Action action)
    {
        _workItemQueue.Enqueue(action);
        _newWorkItemAutoResetEvent.Set();
    }

    public Processor()
    {
        _workerThread = new Thread(() =>
        {
            while (!_disposed)
            {
                _newWorkItemAutoResetEvent.WaitOne(); // 
                while (_workItemQueue.TryDequeue(out Action action))
                {
                    action();
                }
                // at this "bad" moment another thread calls Do method. 
                // New action has been enqueued, but when we call
                // _newWorkIteManualAutoEvent.WaitOne() we fall asleep.
            }
        });
        _thread.Start();
    }
}

然后我尝试使用ManualResetEvent：

public class Processor
{
    ConcurrentQueue<Action> _workItemQueue = new ConcurrentQueue<Action>();
    ManualResetEventSlim _newWorkItemManualResetEvent = new ManualResetEventSlim(false);
    private bool _disposed;
    Thread _thread;

    public void Do(Action action)
    {
        _workItemQueue.Enqueue(action);
        _newWorkItemManualResetEvent.Set();
    }

    public Processor()
    {
        _workerThread = new Thread(() =>
        {
            while (!_disposed)
            {
                _newWorkItemManualResetEvent.WaitOne();
                _newWorkItemManualResetEvent.Reset();

                while (_workItemQueue.TryDequeue(out Action action))
                {
                    action();
                }
            }
        });
        _thread.Start();
    }
}

我没有看到使用 ManualResetEvent 实现的任何竞争条件。

问题：我说的对吗？或者我需要另一个同步原语？我正在考虑 CountupEvent（反向 CountdownEvent）。它在其计数大于零时发出信号，而在其计数等于零时不发出信号。 Countup事件的计数对应于要执行的任务数。

Answer 1

方便的BlockingCollection 将为您处理大部分问题。

类似：

public sealed class Processor : IDisposable
{
    //set a max queue depth to provide back pressure to the request rate
    BlockingCollection<Action> _workItemQueue = new BlockingCollection<Action>(32);
    private bool _disposed = false;
    private Thread _workerThread;
    private CancellationTokenSource _cancelTokenSource = new CancellationTokenSource();

    public void Do(Action action)
    {
        _workItemQueue.Add(action);
    }

    public void Dispose()
    {
        if (!_disposed)
        {
            _workItemQueue.CompleteAdding();
            _cancelTokenSource.Cancel();
            _disposed = true;
        }
    }

    public Processor()
    {
        _workerThread = new Thread(() =>
        {
            while (!_workItemQueue.IsCompleted)
            {
                if (_workItemQueue.TryTake(out Action action, 1000*2,_cancelTokenSource.Token))
                {
                    action();
                }
            }

        });

        _workerThread.Start();
    }
}