仅使用一小部分 CPU 在多个任务中运行 CPU 密集型方法？

Answer 1

这段代码有几个问题限制了它的可扩展性。

• 它在同一任务中执行 IO 和 CPU 工作，而不使用异步方法。您可以同时执行的 CPU 密集型任务的数量受内核数量的限制。您可以执行比这更多的异步任务。
• IO 是阻塞的，不是异步的。这意味着任务（或者更确切地说它的线程）在等待操作系统检索数据时什么也不做。
• 代码读取了太多数据，生成了太多临时对象。 ReadAllText 在只需要 1000 个字符时读取整个文件。字符串是不可变的，所以html.Substring(1,1000) 生成一个新的子串。所有这些都会占用内存，并且必须在某个时候进行垃圾收集。
• ConcurrentBag 不是像 ConcurrentQueue 或 ConcurrentDictionary 这样的通用并发集合。它使用线程本地存储来确保创建项目的线程可以比其他线程更快地检索它。

.NET 提供了几个高级类，可用于构建解析管道，这比加载、解析和导入文件要复杂得多。这些包括Dataflow blocks、Channels 和Async Streams/IAsyncEnumerable。

改进问题代码的一种方法是使用 Dataflow 块来枚举根文件夹，加载文件内容并将其解析为具有不同并行度的不同块。

首先，可以将爬取、加载和解析代码提取到单独的方法中：

record Search(DirectoryInfo root,string pattern);
record Sample(FileInfo file,string sample);
record SampleHtml(FileInfo file,HtmlDocument html);

IEnumerable<FileInfo> Crawl(Search search)
{
    var (root,pattern)=search;
    var searchOptions=new EnumerationOptions { 
        RecurseSubdirectories=true,
        IgnoreInaccessible=true
    };
    return root.EnumerateFiles(pattern,searchOptions);
}

async Task<Sample> ReadSample(FileInfo file, int length)
{
    var buffer=new char[1001];
    using var reader=file.OpenText();
    await reader.ReadAsync(buffer,0,1001);
    var count=await reader.ReadBlockAsync(buffer,0,1000);
    var length=Math.Min(count,1000);
    var html= new String(buffer,1,length);
    return new Sample(file,html);
}

SampleHtml ParseSample(Sample sample)
{
    var html=new HtmlDocument();
    html.LoadHtml(sample.sample);
    return new SampleHtml(sample.file,html);
}

数据流块可用于创建以下管道：

1. 单线程文件搜索块
2. 加载程序块一次加载 2 个文件
3. 解析器块一次解析 4 个样本
4. 一个结果BufferBlock收集解析器的输出

   var loadOptions=new ExecutionDataflowBlockOptions{ 
       MaxDegreeOfParallelism=2,
       BoundedCapacity=1
   };
   var parseOptions=new ExecutionDataflowBlockOptions{ 
       MaxDegreeOfParallelism=4,
       BoundedCapacity=1
   };
   
   var crawler=new TransformManyBlock<Search,FileInfo>(search=>
       Crawl(search);
   
   var loader =  new TransformBlock<FileInfo,Sample>(file=> 
       ReadSample(file),loadOptions);
   
   var parserBlock=new TransformBlock<Sample,SampleHtml>(sample=>
       ParseHtml(sample),parseOptions);
   
   var results=new BufferBlock<SampleHtml>();
   
   var linkOptions=new DataflowLinkOptions {
       PropagateCompletion = true
   };
   crawler.LinkTo(loader,linkOptions);
   loader.LinkTo(parser,linkOptions);
   //Don't propagate completion, we just cache results here
   parser.Linkto(results);
   

   要使用管道，我们将搜索规范发布到头块crawler 并等待直到最后一个块解析器完成所有处理

   var root=new DirectoryInfo(path_to_root);
   var pattern="*.html";
   await crawler.SendAsync(new Search(root,pattern));
   crawler.Complete();
   await parser.Completion;
   

   A this point results 包含所有结果。我们可以使用TryReceive 一个一个地弹出项目或TryReceiveAll 将所有内容读入容器：

   if(results.TryReceiveAll(out var docs)
   {
       var last=docs[^1];
   }
   

   loader 和 parser 块的 BoundedCapacity 为 1。这意味着它们的输入缓冲区将只接受正在处理的项目之外的单个项目。在发布新项目之前，任何上游块都必须等待，一直到爬虫。这可以防止无法足够快地处理的对象填充内存。

   重用缓冲区

   ArrayPool 类可以提供可重用的缓冲区，从而避免为每个文件创建一个新的char[1001] 缓冲区。加载器 DOP 为 4，这意味着我们只需要 4 个缓冲区而不是 3000 个缓冲区：

   async Task<Sample> ReadSample(FileInfo file, int length)
   {
       var buffer=ArrayPool<char>.Shared.Rent(1001);
       try
       {
           using var reader=file.OpenText();
           await reader.ReadAsync(buffer,0,1001);
           var count=await reader.ReadBlockAsync(buffer,0,1000);
           var length=Math.Min(count,1000);
           var html= new String(buffer,1,length);
           return new Sample(file,html);
       }
       finally
       {
           ArrayPool<char>.Shared.Return(buffer);
       }
   }