在 c# 中用于繁重 IO 操作的线程类型

Question

我的任务是更新一个在其操作中非常单线程的 c# 应用程序（非 gui），并向其添加多线程以使其更快地转换工作队列。

每个线程将需要执行非常少量的计算，但大部分工作将调用并等待 SQL Server 请求。因此，与 CPU 时间相比，等待时间更长。

几个要求是：

• 在一些有限的硬件上运行（也就是说，只有几个内核）。当前系统，当它被“推送”时，只需要大约 25% 的 CPU。但是，由于它主要是等待 SQL Server 响应（不同的服务器），因此我们希望能够拥有比内核更多的线程。
• 能够限制线程数。我也不能只拥有无限数量的线程。我不介意通过数组、列表等来限制自己。
• 能够跟踪这些线程何时完成，以便我可以进行一些后期处理。

在我看来，.NET Framework 有很多不同的线程处理方式，我不确定其中一种是否比另一种更好。我不确定我是否应该使用Task、Thread、ThreadPool 或其他东西... appers 我认为async \ await 模型会不不适合这种情况，因为它会等待一项特定任务完成。

Answer 1

我不确定我是否应该使用 Task、Thread、ThreadPool 等等……

在你的情况下，它没有你想象的那么重要。您可以专注于最适合您（现有）代码样式和数据流的内容。

因为它主要是在等待 SQL Server 响应

您的主要目标是让尽可能多的 SQL 查询并行运行。

能够限制线程数。

别担心太多。在 4 个核心，25% 的 CPU 上，您可以轻松拥有 100 个线程。更多关于 64 位的信息。但是你不想要1000个线程。一个 .net 线程至少使用 1MB，估计您可以节省多少 RAM。

所以这取决于您的应用程序，您可以同时运行多少个查询。首先担心线程安全。

当并行查询数 > 1000 时，您将需要 async/await 以在更少的线程上运行。

只要小于 100，就让线程阻塞在 I/O 上。 Parallel.ForEach()、Parallel.Invoke() 等看起来是不错的工具。

100 - 1000 范围是灰色区域。

Answer 2

为其添加多线程以使其更快地完成工作队列。

每个线程将需要执行非常少量的计算，但大部分工作将调用并等待 SQL Server 请求。因此，与 CPU 时间相比，等待时间更长。

对于这种处理方式，多线程对您有什么好处尚不清楚。多线程是并发的一种形式，由于您的工作负载主要受 I/O 限制，因此首先要考虑异步（而不是多线程）。

在我看来，.NET Framework 有很多不同的线程处理方式，我不确定其中一种是否比另一种更好。

确实如此。作为参考，Thread 和 ThreadPool 这些天几乎是遗产；有更好的更高级别的 API。 Task 在用作委托任务时也应该很少见（例如，Task.Factory.StartNew）。

在我看来，async \ await 模型在这种情况下并不适合，尽管它会等待一项特定任务完成。

await 将一次等待一个任务，是的。 Task.WhenAll 可以用来组合 多个任务，然后您可以await 组合任务。

让它更快地完成工作队列。

能够限制线程数。

能够跟踪这些线程何时完成，以便我可以进行一些后期处理。

在我看来TPL Dataflow 将是您系统的最佳方法。数据流允许您定义数据流经的“管道”，其中一些步骤是异步的（例如，查询 SQL Server），而其他步骤是并行的（例如，数据处理）。

我问了一个高层次的问题，试图得到一个高层次的答案。

您可能对my book感兴趣。

Answer 3

TPL Dataflow 库可能是这项工作的最佳选择之一。以下是构建由两个块组成的简单数据流管道的方法。第一个块接受文件路径并生成一些中间数据，稍后可以将其插入数据库。第二个块通过将来自第一个块的数据发送到数据库来使用它们。

var inputBlock = new TransformBlock<string, IntermediateData>(filePath =>
{
    return GetIntermediateDataFromFilePath(filePath);
}, new ExecutionDataflowBlockOptions()
{
    MaxDegreeOfParallelism = Environment.ProcessorCount // What the local machine can handle
});

var databaseBlock = new ActionBlock<IntermediateData>(item =>
{
    SaveItemToDatabase(item);
}, new ExecutionDataflowBlockOptions()
{
    MaxDegreeOfParallelism = 20 // What the database server can handle
});

inputBlock.LinkTo(databaseBlock);

现在每次用户上传文件时，您只需将文件保存在临时路径中，并将路径发布到第一个块：

inputBlock.Post(filePath);

就是这样。数据会自动从流水线的第一个块流到最后一个块，并根据每个块的配置进行转换和处理。

这是一个有意简化的示例，用于演示基本功能。生产就绪的实现可能会定义更多选项，例如CancellationToken 和BoundedCapacity，将观察inputBlock.Post 的返回值以在块无法接受工作的情况下做出反应，可能有completion propagation，查看databaseBlock.Completion 属性是否有错误等。

如果您有兴趣遵循这条路线，最好稍微研究一下库，以便熟悉可用的选项。例如，有一个TransformManyBlock 可用，适用于从单个输入产生多个输出。 BatchBlock 在某些情况下也可能有用。

TPL 数据流内置在 .NET Core 中，可作为 .NET Framework 的 package 使用。它有一些学习曲线和一些需要注意的问题，但这并不可怕。

Answer 4

在我看来，async \ await 模型在这种情况下并不适合，尽管它会等待一项特定任务完成。

这是错误的。 Async/await 只是一种简化异步代码状态机机制的语法。它等待而不消耗任何线程。换句话说，async 关键字不会创建线程，await 不会阻止任何线程。

能够限制线程数

见How to limit the amount of concurrent async I/O operations?

能够跟踪这些线程何时完成，以便我可以进行一些后期处理。

如果您不使用“即发即弃”模式，那么您只需编写 await task 即可跟踪任务及其异常情况

var task = MethodAsync();
await task;
PostProcessing();

async Task MethodAsync(){ ... }

或者对于类似的方法，您可以使用ContinueWith：

var task = MethodAsync();
await task.ContinueWith(() => PostProcessing());

async Task MethodAsync(){ ... }

阅读更多：

Releasing threads during async tasks

https://docs.microsoft.com/en-us/dotnet/standard/asynchronous-programming-patterns/?redirectedfrom=MSDN