c# - 如何使用线程处理多个任务

Question

我有一个单独处理“大量”(可能 > 100,000 条)记录的 C# 要求。事实证明，按顺序运行此过程非常慢，每条记录都需要一秒钟左右的时间才能完成(超时错误设置为 5 秒)。

我想尝试通过使用一定数量的工作“线程”异步运行这些任务(我在这里谨慎使用术语“线程”，因为我不确定我是否应该查看线程，任务或别的东西)。

我看过ThreadPool，但我无法想象它可以对所需的请求量进行排队。我理想的伪代码看起来像这样......

public void ProcessRecords() {
    SetMaxNumberOfThreads(20);
    MyRecord rec;
    while ((rec = GetNextRecord()) != null) {
        var task = WaitForNextAvailableThreadFromPool(ProcessRecord(rec));
        task.Start()
    }
}

我还需要一种机制，处理方法可以向父类/调用类报告。

谁能用一些示例代码为我指出正确的方向？

Answer 1

一个可能的简单解决方案是使用 TPL 数据流 block ，它比 TPL 具有更高的抽象度，具有并行度等配置。您只需创建 block (在本例中为 ActionBlock)，将所有内容Post，异步等待完成，TPL Dataflow 会为您处理所有剩余的事情:

var block = new ActionBlock<MyRecord>(
    rec => ProcessRecord(rec), 
    new ExecutionDataflowBlockOptions{MaxDegreeOfParallelism = 20});

MyRecord rec;
while ((rec = GetNextRecord()) != null)
{
     block.Post(rec);
}

block.Complete();
await block.Completion

另一个好处是，该 block 在第一条记录到达后立即开始工作，而不是仅在收到所有记录后才开始工作。

如果您需要报告每条记录，您可以使用 TransformBlock 进行实际处理并将 ActionBlock 链接到它以进行更新:

var transform = new TransfromBlock<MyRecord, Report>(rec =>
{
    ProcessRecord(rec);
    return GenerateReport(rec);
}, new ExecutionDataflowBlockOptions{MaxDegreeOfParallelism = 20});

var reporter = new ActionBlock<Report>(report =>
{
    RaiseEvent(report) // Or any other mechanism...
});

transform.LinkTo(reporter, new DataflowLinkOptions { PropagateCompletion = true });

MyRecord rec;
while ((rec = GetNextRecord()) != null)
{
     transform.Post(rec);
}

transform.Complete();
await transform.Completion