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

Posted 2023-03-15

【中文标题】在 c# 中用于繁重 IO 操作的线程类型

【英文标题】：Type of threading to use in c# for heavy IO operations

【发布时间】：2020-12-09 05:22:36

【问题描述】：

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

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

几个要求是：

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

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

【问题讨论】：

能否请您发布当前代码？

@Enigmativity 当前的代码是一个相当大的项目，我什至不知道从哪里开始分割它，即使是一些片段。这就是为什么我要问一个高层次的问题来试图得到一个高层次的答案。对不起。

那么您应该尝试给出一些示例签名，以表明您至少正在拨打的电话。完成这项工作的工具需要匹配。这有点像你问我们如何砍木头——有上百万种工具可以砍木头——但你需要一把拼图还是一把斧头？

当前应用中的工作队列是如何管理的？工作是如何被触发的？结果如何？

【参考方案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 范围是灰色区域。

【参考方案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感兴趣。

【讨论】：

“采用那种处理方式”。用户将上传许多要处理的项目（数据文件），目前该过程是“单线程”的。一个文件被处理，然后是下一个，然后是下一个......如果我们有 15 个用户大约在同一时间上传，每个文件大约需要 1 分钟来处理，那么最后上传的人将等待大约 15 分钟。由于这 1 分钟的大部分时间都在等待各种调用中的 SQL，我们希望同时处理多个文件，以便最后一个上传者更快地完成。

【参考方案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 使用。它有一些学习曲线和一些需要注意的问题，但这并不可怕。

【参考方案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

【讨论】：

我想不出任何方式 ContinueWith 比 await“更好”。