关于Parallel.For/ForEach按顺序获取枚举元素的技术问询

问题根源 #

Parallel.For/Parallel.ForEach默认使用**分区器（Partitioner）**预先将集合分割成多个块分配给线程，而非按需逐个获取元素。当任务耗时差异极大时，快的线程会提前干完自己的任务块，慢的线程还在处理耗时任务，导致后期活跃线程数锐减，CPU利用率下降。

你之前尝试的lock抢索引方式存在逻辑问题：Parallel.For(0, items.Length, ...)会启动与集合长度相等的并行任务（200个），即使设置了MaxDegreeOfParallelism=8，调度器只是控制同时运行的任务数，而非保持8个线程持续循环取任务，本质还是任务级调度，无法解决后期线程闲置的问题。

解决方案：固定线程数+线程安全队列 #

核心思路是：启动固定数量的工作线程（8个），让每个线程从线程安全的FIFO队列中按需获取下一个任务，直到队列为空。这种方式能严格保持8个线程运行，直到所有任务被领取处理。

实现方式1：手动创建队列+Task.Run #

using System.Collections.Concurrent;
using System.Threading;

// 1. 按任务复杂度排序（替换为你的实际排序逻辑）
var sortedItems = Enumerable.Range(0, 200)
    .OrderBy(item => /* 例如：任务耗时预估，复杂度高的排前面 */)
    .ToList();

// 2. 初始化线程安全队列
var taskQueue = new ConcurrentQueue<int>(sortedItems);

// 3. 启动8个工作线程
var workerTasks = new Task[8];
for (int i = 0; i < 8; i++)
{
    workerTasks[i] = Task.Run(() =>
    {
        // 循环取任务直到队列空
        while (taskQueue.TryDequeue(out var item))
        {
            // 替换为你的任务处理逻辑
            Console.WriteLine($"线程{Thread.CurrentThread.ManagedThreadId}处理：{item}");
            // 模拟任务耗时差异
            Thread.Sleep(new Random().Next(1, 1000));
        }
    });
}

// 4. 等待所有线程完成
Task.WaitAll(workerTasks);

实现方式2：TPL Dataflow（更优雅的并发控制） #

如果你需要更灵活的并发管理（比如取消、进度跟踪、批量处理），可以使用TPL Dataflow的ActionBlock，它内置了并发度控制：

using System.Threading.Tasks.Dataflow;
using System.Threading;

// 1. 配置ActionBlock，设置并发度为8
var processor = new ActionBlock<int>(
    item =>
    {
        // 任务处理逻辑
        Console.WriteLine($"线程{Thread.CurrentThread.ManagedThreadId}处理：{item}");
        Thread.Sleep(new Random().Next(1, 1000));
    },
    new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 8 }
);

// 2. 按顺序发送排序后的任务
var sortedItems = Enumerable.Range(0, 200)
    .OrderBy(item => /* 复杂度排序逻辑 */);
foreach (var item in sortedItems)
{
    processor.Post(item);
}

// 3. 标记任务发送完成
processor.Complete();

// 4. 等待所有任务处理完毕
await processor.Completion;

方案优势 #

• 严格保持8个线程运行：直到队列中所有任务被领取，避免后期线程闲置。
• 任务按顺序领取：队列是FIFO结构，排序后的任务会按顺序被线程获取，满足你“按顺序从枚举中获取元素”的需求（任务完成顺序不影响）。
• 适配耗时差异大的任务：快线程干完立刻取下一个任务，不会因为预分区导致资源浪费。

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