任务调度程序：在 Task.Factory.StartNew 中等待时，线程是否返回到池中？

Question

我正在实现一个具有并发上限的工作引擎。我正在使用信号量等待并发量降至最大值以下，然后使用Task.Factory.StartNew 将异步处理程序包装在try/catch 中，并使用finally 释放信号量。

我意识到这会在线程池上创建线程 - 但我的问题是，当这些任务运行线程之一实际等待时（在真正的 IO 调用或等待句柄上），线程是否返回到池中，就像我一样希望是吗？

如果有更好的方法来实现具有有限并发的任务调度程序，其中工作处理程序是异步方法（返回 Task），我也很想听听。或者，假设理想情况下，如果有一种方法可以让异步方法排队（同样，它是一个 Task-returning 异步方法），感觉不如将它包装在同步委托中并将其传递给 Task.Factory.StartNew，那将看起来很完美..？

（这也让我认为这里有两种并行性：总共处理了多少个任务，以及同时在不同线程上运行了多少个延续。为两者提供可配置选项可能很酷，但不是一个固定的要求..）

编辑：sn-p：

                    concurrencySemaphore.Wait(cancelToken);
                    deferRelease = false;
                    try
                    {
                        var result = GetWorkItem();
                        if (result == null)
                        { // no work, wait for new work or exit signal
                            signal = WaitHandle.WaitAny(signals);
                            continue;
                        }

                        deferRelease = true;
                        tasks.Add(Task.Factory.StartNew(() =>
                        {
                            try
                            {
                                DoWorkHereAsync(result); // guess I'd think to .GetAwaiter().GetResult() here.. not run this yet
                            }
                            finally
                            {
                                concurrencySemaphore.Release();
                            }
                        }, cancelToken));
                    }
                    finally
                    {
                        if (!deferRelease)
                        {
                            concurrencySemaphore.Release();
                        }
                    }

Answer 1

您可以认为线程返回到ThreadPool，即使它实际上不是返回。当异步操作开始时，线程只会选择下一个排队的项目。

我建议您查看Task.Run 而不是Task.Factory.StartNew Task.Run vs Task.Factory.StartNew。

还可以查看TPL DataFlow。我认为它会符合您的要求。

Answer 2

这里是一个 TaskWorker 的例子，它不会产生无数的工作线程。

魔术是通过等待SemaphoreSlim.WaitAsync() 完成的，这是一个 IO 任务（并且没有线程）。

class TaskWorker
{
    private readonly SemaphoreSlim _semaphore;

    public TaskWorker(int maxDegreeOfParallelism)
    {
        if (maxDegreeOfParallelism <= 0)
        {
            throw new ArgumentOutOfRangeException(nameof(maxDegreeOfParallelism));
        }

        _semaphore = new SemaphoreSlim(maxDegreeOfParallelism, maxDegreeOfParallelism);
    }

    public async Task RunAsync(Func<Task> taskFactory, CancellationToken cancellationToken = default(CancellationToken))
    {
        // No ConfigureAwait(false) here to keep the SyncContext if any
        // for the real task
        await _semaphore.WaitAsync(cancellationToken);
        try
        {
            await taskFactory().ConfigureAwait(false);
        }
        finally
        {
            _semaphore.Release(1);
        }
    }

    public async Task<T> RunAsync<T>(Func<Task<T>> taskFactory, CancellationToken cancellationToken = default(CancellationToken))
    {
        await _semaphore.WaitAsync(cancellationToken);
        try
        {
            return await taskFactory().ConfigureAwait(false);
        }
        finally
        {
            _semaphore.Release(1);
        }
    }
}

和一个简单的控制台应用程序进行测试

class Program
{
    static void Main(string[] args)
    {
        var worker = new TaskWorker(1);
        var cts = new CancellationTokenSource();
        var token = cts.Token;

        var tasks = Enumerable.Range(1, 10)
            .Select(e => worker.RunAsync(() => SomeWorkAsync(e, token), token))
            .ToArray();

        Task.WhenAll(tasks).GetAwaiter().GetResult();
    }

    static async Task SomeWorkAsync(int id, CancellationToken cancellationToken)
    {
        Console.WriteLine($"Some Started {id}");
        await Task.Delay(2000, cancellationToken).ConfigureAwait(false);
        Console.WriteLine($"Some Finished {id}");
    }
}

更新

TaskWorker 实现IDisposable

class TaskWorker : IDisposable
{
    private readonly CancellationTokenSource _cts = new CancellationTokenSource();
    private readonly SemaphoreSlim _semaphore;
    private readonly int _maxDegreeOfParallelism;

    public TaskWorker(int maxDegreeOfParallelism)
    {
        if (maxDegreeOfParallelism <= 0)
        {
            throw new ArgumentOutOfRangeException(nameof(maxDegreeOfParallelism));
        }

        _maxDegreeOfParallelism = maxDegreeOfParallelism;
        _semaphore = new SemaphoreSlim(maxDegreeOfParallelism, maxDegreeOfParallelism);
    }

    public async Task RunAsync(Func<Task> taskFactory, CancellationToken cancellationToken = default(CancellationToken))
    {
        ThrowIfDisposed();

        using (var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _cts.Token))
        {
            // No ConfigureAwait(false) here to keep the SyncContext if any
            // for the real task
            await _semaphore.WaitAsync(cts.Token);
            try
            {
                await taskFactory().ConfigureAwait(false);
            }
            finally
            {
                _semaphore.Release(1);
            }
        }
    }

    public async Task<T> RunAsync<T>(Func<Task<T>> taskFactory, CancellationToken cancellationToken = default(CancellationToken))
    {
        ThrowIfDisposed();

        using (var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, _cts.Token))
        {
            await _semaphore.WaitAsync(cts.Token);
            try
            {
                return await taskFactory().ConfigureAwait(false);
            }
            finally
            {
                _semaphore.Release(1);
            }
        }
    }

    private void ThrowIfDisposed()
    {
        if (disposedValue)
        {
            throw new ObjectDisposedException(this.GetType().FullName);
        }
    }

    #region IDisposable Support
    private bool disposedValue = false;

    protected virtual void Dispose(bool disposing)
    {
        if (!disposedValue)
        {
            if (disposing)
            {
                _cts.Cancel();
                // consume all semaphore slots
                for (int i = 0; i < _maxDegreeOfParallelism; i++)
                {
                    _semaphore.WaitAsync().GetAwaiter().GetResult();
                }
                _semaphore.Dispose();
                _cts.Dispose();
            }
            disposedValue = true;
        }
    }

    public void Dispose()
    {
        Dispose(true);
    }
    #endregion
}