假设有许多线程调用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调用中引入超时以避免日志阻塞。