我正在运行一个多线程循环：
protected ParallelOptions parallelOptions = new ParallelOptions();
parallelOptions.MaxDegreeOfParallelism = 2;
Parallel.ForEach(items, parallelOptions, item =>
{
// Loop code here
});
parallelOptions.MaxDegreeOfParallelism，以减少或增加线程数量。parallelOptions.MaxDegreeOfParallelism = 5;
它似乎无法增加线程。有人有任何想法吗？
我不认为在调用ForEach之后改变并行度是可能的。据我了解，ForEach将确定它可以创建多少线程，创建相应数量的分区，并创建线程来操作这些分区。它没有任何时候可以说：“哦，等等，他改变了我们的资源分配，让我重新对数组进行分区和重新分配线程。”
Parallel.ForEachAsync API，它允许动态配置并行度。它与原生API共享相同的参数和行为，唯一的区别是它接受一个派生版本的ParallelOptions作为参数（DynamicParallelOptions）。这个类有一个额外的属性DegreeOfParallelism。改变这个属性会迅速适应当前活动的并行度。Parallel.ForEachAsync API的源序列进行限流的想法。API本身被配置为DegreeOfParallelism的最大预期值。实际的并行度通过限制循环对源元素的自由访问来有效地进行限制。每当处理另一个元素时，就会向前传播一个元素。限流本身是使用一个无界的SemaphoreSlim来执行的。通过调用信号量的Release/WaitAsync方法来改变最大并行度。/// <summary>
/// Executes a parallel foreach operation on an asynchronous sequence, enforcing
/// a degree of parallelism that can be dynamically changed during the execution.
/// </summary>
public static Task DynamicParallelForEachAsync<TSource>(
IAsyncEnumerable<TSource> source,
DynamicParallelOptions options,
Func<TSource, CancellationToken, ValueTask> body)
{
ArgumentNullException.ThrowIfNull(source);
ArgumentNullException.ThrowIfNull(options);
ArgumentNullException.ThrowIfNull(body);
SemaphoreSlim throttler = new(options.DegreeOfParallelism);
options.DegreeOfParallelismChangedDelta += Options_ChangedDelta;
void Options_ChangedDelta(object sender, int delta)
{
if (delta > 0)
throttler.Release(delta);
else
for (int i = delta; i < 0; i++) throttler.WaitAsync();
}
async IAsyncEnumerable<TSource> GetThrottledSource(
[EnumeratorCancellation] CancellationToken cancellationToken = default)
{
IAsyncEnumerator<TSource> enumerator = source.GetAsyncEnumerator(
cancellationToken);
await using (enumerator.ConfigureAwait(false))
{
while (true)
{
await throttler.WaitAsync().ConfigureAwait(false);
if (!await enumerator.MoveNextAsync().ConfigureAwait(false)) break;
yield return enumerator.Current;
}
}
}
return Parallel.ForEachAsync(GetThrottledSource(), options, async (item, ct) =>
{
try { await body(item, ct).ConfigureAwait(false); }
finally { throttler.Release(); }
}).ContinueWith(t =>
{
options.DegreeOfParallelismChangedDelta -= Options_ChangedDelta;
return t;
}, default, TaskContinuationOptions.DenyChildAttach |
TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default).Unwrap();
}
/// <summary>
/// Stores options that configure the DynamicParallelForEachAsync method.
/// </summary>
public class DynamicParallelOptions : ParallelOptions
{
private int _degreeOfParallelism;
public event EventHandler<int> DegreeOfParallelismChangedDelta;
public DynamicParallelOptions(int maxDegreeOfParallelism)
{
// The native Parallel.ForEachAsync will see the base.MaxDegreeOfParallelism.
base.MaxDegreeOfParallelism = maxDegreeOfParallelism;
_degreeOfParallelism = Environment.ProcessorCount;
}
public int DegreeOfParallelism
{
get { return _degreeOfParallelism; }
set
{
if (value < 1) throw new ArgumentOutOfRangeException();
if (value == _degreeOfParallelism) return;
int delta = value - _degreeOfParallelism;
DegreeOfParallelismChangedDelta?.Invoke(this, delta);
_degreeOfParallelism = value;
}
}
}
DynamicParallelOptions.DegreeOfParallelism属性不是线程安全的。假定控制并行度将由单个线程执行，或者至少操作将被同步。
使用示例，以 Channel<T> 作为并行循环的源：
Channel<int> channel = Channel.CreateUnbounded<int>();
DynamicParallelOptions options = new(maxDegreeOfParallelism: 50)
{
DegreeOfParallelism = 2
};
await DynamicParallelForEachAsync(
channel.Reader.ReadAllAsync(), options, async (item, ct) =>
{
Console.WriteLine($"Processing #{item}");
await Task.Delay(1000, ct); // Simulate an I/O-bound operation
});
// Push values to the channel from any thread
channel.Writer.TryWrite(1);
channel.Writer.TryWrite(2);
channel.Writer.TryWrite(3);
channel.Writer.Complete();
// Set the DegreeOfParallelism to a positive value at any time from a single thread
options.DegreeOfParallelism = 5;
public static Task DynamicParallelForEachAsync<TSource>(
IEnumerable<TSource> source,
DynamicParallelOptions options,
Func<TSource, CancellationToken, ValueTask> body)
{
ArgumentNullException.ThrowIfNull(source);
#pragma warning disable CS1998
async IAsyncEnumerable<TSource> GetSource()
{ foreach (TSource item in source) yield return item; }
#pragma warning restore CS1998
return DynamicParallelForEachAsync(GetSource(), options, body);
}
public static void DynamicParallelForEach<TSource>(
IEnumerable<TSource> source,
DynamicParallelOptions options,
Action<TSource, CancellationToken> body)
{
ArgumentNullException.ThrowIfNull(body);
DynamicParallelForEachAsync(source, options, (item, ct) =>
{
body(item, ct); return ValueTask.CompletedTask;
}).Wait();
}
DegreeOfParallelism的默认值是Environment.ProcessorCount。允许将DegreeOfParallelism设置为大于MaxDegreeOfParallelism的值，尽管它没有任何效果。 MaxDegreeOfParallelism表示有效并行度的硬上限。建议使用合理的值配置MaxDegreeOfParallelism。如果将其设置得太高，例如Int32.MaxValue，可能会增加整个操作的内存和CPU开销，特别是如果source包含大量元素。