C＃concurrent：使用很多AutoResetEvent是个好主意吗？

Posted 2021-05-04

假设有许多线程调用Do()，并且只有一个工作线程处理实际的工作。

void Do(Job job)
{
    concurrentQueue.Enqueue(job);
    // wait for job done
}

void workerThread()
{
    while (true)
    {
        Job job;
        if (concurrentQueue.TryDequeue(out job))
        {
            // do job
        }
    }
}

Do（）应该等到作业完成后再返回。所以我写了下面的代码：

class Task 
{
    public Job job;
    public AutoResetEvent ev;
}

void Do(Job job)
{
    using (var ev = new AutoResetEvent(false))
    {
        concurrentQueue.Enqueue(new Task { job = job, ev = ev }));
        ev.WaitOne();
    }
}

void workerThread()
{
    while (true)
    {
        Task task;
        if (concurrentQueue.TryDequeue(out task))
        {
            // do job
            task.ev.Set();
        }
    }
}

经过一些测试后，我发现它按预期工作。但是我不确定这是分配许多AutoResetEvents的好方法，还是有更好的方法来完成？

答案

由于所有客户端都必须等待一个线程来完成工作，因此不需要使用队列。所以我建议使用Monitor类，特别是Wait/Pulse功能。虽然它有点低级且冗长。

class Worker<TResult> : IDisposable
{
    private readonly object _outerLock = new object();
    private readonly object _innerLock = new object();
    private Func<TResult> _currentJob;
    private TResult _currentResult;
    private Exception _currentException;
    private bool _disposed;

    public Worker()
    {
        var thread = new Thread(MainLoop);
        thread.IsBackground = true;
        thread.Start();
    }

    private void MainLoop()
    {
        lock (_innerLock)
        {
            while (true)
            {
                Monitor.Wait(_innerLock); // Wait for client requests
                if (_disposed) break;
                try
                {
                    _currentResult = _currentJob.Invoke();
                    _currentException = null;
                }
                catch (Exception ex)
                {
                    _currentException = ex;
                    _currentResult = default;
                }
                Monitor.Pulse(_innerLock); // Notify the waiting client that the job is done
            }
        } // We are done
    }

    public TResult DoWork(Func<TResult> job)
    {
        TResult result;
        Exception exception;
        lock (_outerLock) // Accept only one client at a time
        {
            lock (_innerLock) // Acquire inner lock
            {
                if (_disposed) throw new InvalidOperationException();
                _currentJob = job;
                Monitor.Pulse(_innerLock); // Notify worker thread about the new job
                Monitor.Wait(_innerLock); // Wait for worker thread to process the job
                result = _currentResult;
                exception = _currentException;
                // Clean up
                _currentJob = null;
                _currentResult = default;
                _currentException = null;
            }
        }
        // Throw the exception, if occurred, preserving the stack trace
        if (exception != null) ExceptionDispatchInfo.Capture(exception).Throw();
        return result;
    }

    public void Dispose()
    {
        lock (_outerLock)
        {
            lock (_innerLock)
            {
                _disposed = true;
                Monitor.Pulse(_innerLock); // Notify worker thread to exit loop
            }
        }
    }
}

用法示例：

var worker = new Worker<int>();
int result = worker.DoWork(() => 1); // Accepts a function as argument
Console.WriteLine($"Result: {result}");
worker.Dispose();

输出：

Result: 1

---

更新：以前的解决方案不是等待友好的，所以这里有一个允许正确等待的解决方案。它为每个作业使用TaskCompletionSource，存储在BlockingCollection中。

class Worker<TResult> : IDisposable
{
    private BlockingCollection<TaskCompletionSource<TResult>> _blockingCollection
        = new BlockingCollection<TaskCompletionSource<TResult>>();

    public Worker()
    {
        var thread = new Thread(MainLoop);
        thread.IsBackground = true;
        thread.Start();
    }

    private void MainLoop()
    {
        foreach (var tcs in _blockingCollection.GetConsumingEnumerable())
        {
            var job = (Func<TResult>)tcs.Task.AsyncState;
            try
            {
                var result = job.Invoke();
                tcs.SetResult(result);
            }
            catch (Exception ex)
            {
                tcs.TrySetException(ex);
            }
        }
    }

    public Task<TResult> DoWorkAsync(Func<TResult> job)
    {
        var tcs = new TaskCompletionSource<TResult>(job,
            TaskCreationOptions.RunContinuationsAsynchronously);
        _blockingCollection.Add(tcs);
        return tcs.Task;
    }

    public TResult DoWork(Func<TResult> job) // Synchronous call
    {
        var task = DoWorkAsync(job);
        try { task.Wait(); } catch { } // Swallow the AggregateException
        // Throw the original exception, if occurred, preserving the stack trace
        if (task.IsFaulted) ExceptionDispatchInfo.Capture(task.Exception.InnerException).Throw();
        return task.Result;
    }

    public void Dispose()
    {
        _blockingCollection.CompleteAdding();
    }
}

用法示例

var worker = new Worker<int>();
int result = await worker.DoWorkAsync(() => 1); // Accepts a function as argument
Console.WriteLine($"Result: {result}");
worker.Dispose();

输出：

Result: 1

另一答案

从同步的角度来看，这很好。

但这样做似乎毫无用处。如果你想一个接一个地执行作业，你可以使用一个锁：

lock (lockObject) {
  RunJob();
}

你对这段代码的意图是什么？

还有一个效率问题，因为每个任务都会创建一个OS事件并等待它。如果您使用更现代的TaskCompletionSource，如果您同步等待该任务，这将在引擎盖下使用相同的东西。您可以使用异步等待（await myTCS.Task;）来提高效率。当然，这会使用async / await感染整个调用堆栈。如果这是一个相当低的音量操作，你将获得不多。

另一答案

一般来说，我认为会起作用，虽然当你说“很多”线程正在调用Do（）时，这可能无法很好地扩展...挂起的线程使用资源。

此代码的另一个问题是，在空闲时，“workerThread”中将出现“硬循环”，这将导致应用程序返回高CPU利用率时间。您可能希望将此代码添加到“workerThread”：

if (concurrentQueue.IsEmpty) Thread.Sleep(1);

您可能还想在WaitOne调用中引入超时以避免日志阻塞。