如何在PLINQ查询中动态调整DegreeOfParallelism并行度？

原生PLINQ并没有提供WithDynamicDegreeOfParallelism这类原生方法来动态调整并行度——WithDegreeOfParallelism设置的是固定值，一旦查询启动，并行度就会固定下来，无法在执行过程中修改。不过可以通过替代方案实现类似的动态控制效果。

替代实现方案 #

1. 手动任务调度+信号量动态限流 #

通过SemaphoreSlim控制并发任务数量，配合后台线程定期更新信号量的可用许可数，实现动态调整并行度的目的。这种方式能更精细地控制每一个任务的并发数，同时保证输出顺序（因为按索引处理）。

示例代码：

// 示例回调函数，动态返回当前线程限制
int CurrentThreadLimit() => DateTime.Now.Second / 4 + 1;

var semaphore = new SemaphoreSlim(CurrentThreadLimit());
var outputs = new object[inputs.Length]; // 替换为你的实际输出类型
var tasks = new List<Task>();

// 后台线程定期调整信号量许可数
var adjustTask = Task.Run(async () =>
{
    while (tasks.Any(t => !t.IsCompleted))
    {
        int newLimit = CurrentThreadLimit();
        int currentCount = semaphore.CurrentCount;
        
        // 仅处理许可数增加的情况，减少的情况可等现有任务完成后自然限制
        if (newLimit > currentCount)
        {
            semaphore.Release(newLimit - currentCount);
        }
        await Task.Delay(1000); // 每秒检查调整一次
    }
});

// 提交所有转换任务
for (int i = 0; i < inputs.Length; i++)
{
    int index = i;
    tasks.Add(Task.Run(async () =>
    {
        await semaphore.WaitAsync();
        try
        {
            outputs[index] = MyConverter(inputs[index]);
        }
        finally
        {
            semaphore.Release();
        }
    }));
}

// 等待所有任务完成
await Task.WhenAll(tasks);
await adjustTask;

2. 分批执行PLINQ查询 #

将输入数组拆分为多个批次，每处理完一个批次后，重新获取最新的线程限制，再启动下一批PLINQ查询。这种方式实现简单，且保留了PLINQ的原生优化。

示例代码：

// 示例回调函数
int CurrentThreadLimit() => DateTime.Now.Second / 4 + 1;

// 自定义Batch扩展方法，用于拆分输入为批次
public static IEnumerable<IEnumerable<T>> Batch<T>(this IEnumerable<T> source, int batchSize)
{
    using var enumerator = source.GetEnumerator();
    while (enumerator.MoveNext())
    {
        var batch = new List<T> { enumerator.Current };
        while (batch.Count < batchSize && enumerator.MoveNext())
        {
            batch.Add(enumerator.Current);
        }
        yield return batch;
    }
}

var outputs = new List<object>(); // 替换为你的实际输出类型
// 按每批100个元素拆分，可根据实际情况调整批次大小
var inputBatches = inputs.Batch(100);

foreach (var batch in inputBatches)
{
    int currentLimit = CurrentThreadLimit();
    // 对当前批次使用最新的并行度执行PLINQ转换
    var batchOutput = batch
        .AsParallel()
        .AsOrdered()
        .WithDegreeOfParallelism(currentLimit)
        .Select(MyConverter)
        .ToArray();
    outputs.AddRange(batchOutput);
}

var finalOutputs = outputs.ToArray();

注意事项 #

• 动态调整并行度会带来额外的调度开销，如果回调返回的线程限制波动过于频繁，可能导致性能不稳定，建议给限制值设置合理的上下边界（比如最小1，最大不超过Environment.ProcessorCount * 2）。
• 若需要严格保证输出顺序，两种方案都需要注意顺序控制：手动调度方案通过索引直接对应输出位置，分批方案依赖AsOrdered确保批次内的顺序，且批次按输入顺序处理。

内容的提问来源于stack exchange，提问作者stb