您没有在任何代码中利用任何异步 I/O API。你所做的一切都受 CPU 限制，所有 I/O 操作都会阻塞地浪费 CPU 资源。AsParallel适用于计算密集型任务，如果您想利用异步 I/O，则需要在 BeginXXX/EndXXX您正在使用的基于 I/O 的类上的方法，并在可用时利用这些方法。
初学者请阅读这篇文章：TPL TaskFactory.FromAsync 与具有阻塞方法的任务 https://stackoverflow.com/questions/5018897/tpl-taskfactory-fromasync-vs-tasks-with-blocking-methods/5073816#5073816
接下来，你不想使用AsParallel无论如何在这种情况下。AsParallel启用流式传输，这将导致立即为每个项目安排一个新任务，但您在这里不需要/不希望这样做。通过使用分区工作，您会得到更好的服务Parallel::ForEach.
让我们看看如何利用这些知识在您的特定情况下实现最大并发性：
var refs = GetReferencesFromDB();
// Using Parallel::ForEach here will partition and process your data on separate worker threads
Parallel.ForEach(
refs,
ref =>
{
string filePath = GetFilePath(ref);
byte[] fileDataBuffer = new byte[1048576];
// Need to use FileStream API directly so we can enable async I/O
FileStream sourceFileStream = new FileStream(
filePath,
FileMode.Open,
FileAccess.Read,
FileShare.Read,
8192,
true);
// Use FromAsync to read the data from the file
Task<int> readSourceFileStreamTask = Task.Factory.FromAsync(
sourceFileStream.BeginRead
sourceFileStream.EndRead
fileDataBuffer,
fileDataBuffer.Length,
null);
// Add a continuation that will fire when the async read is completed
readSourceFileStreamTask.ContinueWith(readSourceFileStreamAntecedent =>
{
int soureFileStreamBytesRead;
try
{
// Determine exactly how many bytes were read
// NOTE: this will propagate any potential exception that may have occurred in EndRead
sourceFileStreamBytesRead = readSourceFileStreamAntecedent.Result;
}
finally
{
// Always clean up the source stream
sourceFileStream.Close();
sourceFileStream = null;
}
// This is here to make sure you don't end up trying to read files larger than this sample code can handle
if(sourceFileStreamBytesRead == fileDataBuffer.Length)
{
throw new NotSupportedException("You need to implement reading files larger than 1MB. :P");
}
// Convert the file data to a string
string html = Encoding.UTF8.GetString(fileDataBuffer, 0, sourceFileStreamBytesRead);
// Parse the HTML
string convertedHtml = ParseHtml(html);
// This is here to make sure you don't end up trying to write files larger than this sample code can handle
if(Encoding.UTF8.GetByteCount > fileDataBuffer.Length)
{
throw new NotSupportedException("You need to implement writing files larger than 1MB. :P");
}
// Convert the file data back to bytes for writing
Encoding.UTF8.GetBytes(convertedHtml, 0, convertedHtml.Length, fileDataBuffer, 0);
// Need to use FileStream API directly so we can enable async I/O
FileStream destinationFileStream = new FileStream(
destinationFilePath,
FileMode.OpenOrCreate,
FileAccess.Write,
FileShare.None,
8192,
true);
// Use FromAsync to read the data from the file
Task destinationFileStreamWriteTask = Task.Factory.FromAsync(
destinationFileStream.BeginWrite,
destinationFileStream.EndWrite,
fileDataBuffer,
0,
fileDataBuffer.Length,
null);
// Add a continuation that will fire when the async write is completed
destinationFileStreamWriteTask.ContinueWith(destinationFileStreamWriteAntecedent =>
{
try
{
// NOTE: we call wait here to observe any potential exceptions that might have occurred in EndWrite
destinationFileStreamWriteAntecedent.Wait();
}
finally
{
// Always close the destination file stream
destinationFileStream.Close();
destinationFileStream = null;
}
},
TaskContinuationOptions.AttachedToParent);
// Send to external system **concurrent** to writing to destination file system above
SendToWs(ref, convertedHtml);
},
TaskContinuationOptions.AttachedToParent);
});
现在，这里有一些注意事项：
- 这是示例代码，因此我使用 1MB 缓冲区来读/写文件。这对于 HTML 文件来说是过多的并且浪费系统资源。您可以降低它以满足您的最大需求，也可以将链式读/写实现到 StringBuilder 中，这是我留给您的练习，因为我要编写大约 500 行以上的代码来执行异步链式读/写。 ：P
- 您会注意到，在读/写任务的延续中，我有
TaskContinuationOptions.AttachedToParent。这非常重要，因为它将阻止工作线程Parallel::ForEach从完成开始工作，直到所有底层异步调用完成。如果不是这样，您将同时启动所有 5000 个项目的工作，这将导致数千个计划任务污染 TPL 子系统，并且根本无法正常扩展。
- 我在这里调用 SendToWs 并发将文件写入文件共享。我不知道 SendToWs 实现的基础是什么，但它听起来也像是进行异步的一个很好的候选者。现在假设它是纯粹的计算工作，因此在执行时会消耗 CPU 线程。我将其作为练习留给您，以了解如何最好地利用我向您展示的内容来提高吞吐量。
- 这都是自由输入的形式，我的大脑是这里唯一的编译器，SO 的语法高亮是我用来确保语法良好的全部。所以，请原谅任何语法错误，如果我搞砸了任何事情，以至于你无法理解它，请告诉我，我会跟进。
