C#控制台应用：如何让100线程持续运行优化批量并行循环？

这个场景太常见了——当你用Parallel.ForEach按批次处理任务时，一旦批次里混着极耗时的任务，就会出现大部分线程闲等的尴尬情况。要实现始终保持100个线程满负载运行，我推荐用TPL Dataflow里的ActionBlock<T>，它完美适配这种“持续喂任务、自动维持并发数”的需求。

为什么原方案效率低？ #

你原来的代码是批次式执行：必须等当前整批myItems的所有任务（包括那个10分钟的慢任务）全部完成，才会去取下一批数据。这就导致99个线程早早干完活，却要空等10分钟才能处理新数据，完全浪费了系统资源。

最优实现方案：用ActionBlock持续处理 #

ActionBlock<T>是TPL Dataflow组件的核心之一，它可以：

• 固定最大并发数（比如你要的100）
• 只要有线程空闲，就自动从待处理队列里取新任务执行
• 支持持续异步喂入任务，不用等整批结束

步骤1：引入依赖 #

首先需要安装System.Threading.Tasks.Dataflow NuGet包（这是官方的组件，放心用）。

步骤2：完整代码示例 #

using System;
using System.Collections.Generic;
using System.Threading.Tasks.Dataflow;

class Program
{
    static void Main(string[] args)
    {
        // 初始化ActionBlock，设置最大并发数为100
        var updateBlock = new ActionBlock<int>(
            itemId => CallUpdate(itemId),
            new ExecutionDataflowBlockOptions
            {
                MaxDegreeOfParallelism = 100,
                // 可选：如果需要支持取消，可以传入CancellationToken
                // CancellationToken = cancellationToken.Token
            });

        // 启动后台任务，持续获取数据并喂入ActionBlock
        Task.Run(async () =>
        {
            List<Items> myItems;
            do
            {
                myItems = GetItems();
                foreach (var item in myItems)
                {
                    // 异步发送任务到块，不会阻塞当前线程
                    await updateBlock.SendAsync(item.ID);
                }
            } while (myItems.Count > 0);

            // 所有数据都喂完了，通知块不再接收新任务
            updateBlock.Complete();
        });

        // 等待所有任务处理完成
        updateBlock.Completion.Wait();

        Console.WriteLine("所有更新任务已完成");
    }

    // 模拟你的获取数据方法
    static List<Items> GetItems()
    {
        // 这里替换成你的实际逻辑
        return new List<Items>();
    }

    // 模拟你的更新方法
    static void CallUpdate(int itemId)
    {
        // 这里替换成你的实际逻辑
        // 比如那个耗时10分钟的任务和1秒的任务
    }

    class Items
    {
        public int ID { get; set; }
    }
}

关键优势说明 #

• 持续满负载：只要有线程完成任务，就立刻接手新的任务（来自下一批获取的数据），不会出现99个线程闲等的情况。
• 异步喂入：SendAsync是异步的，获取数据的线程不会被阻塞，可以持续拉取新数据喂给处理块。
• 灵活可控：可以轻松添加取消令牌、错误处理（比如通过Completion捕获异常）、限制队列长度等。

额外注意事项 #

1. 线程安全：确保CallUpdate方法本身是线程安全的——如果多个线程同时操作共享资源，需要加锁或者用线程安全的数据结构。
2. 队列压力：如果GetItems拉取数据的速度远快于处理速度，ActionBlock的内部队列会越来越长，可能占用过多内存。可以通过设置BoundedCapacity来限制队列长度，当队列满时SendAsync会自动等待，避免内存溢出。
3. 错误处理：如果CallUpdate抛出异常，ActionBlock会进入故障状态，不再处理后续任务。你可以通过updateBlock.Completion.ContinueWith来捕获异常并处理，比如记录日志、重试等。

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