理解并掌握C#的Channel:从使用案例到源码解读(一)

  • 理解并掌握C#的Channel:从使用案例到源码解读(一)已关闭评论
  • 117 次浏览
  • A+
所属分类:.NET技术
摘要

在C#的并发编程中,Channel是一种非常强大的数据结构,用于在生产者和消费者之间进行通信。本文将首先通过一个实际的使用案例,介绍如何在C#中使用Channel,然后深入到Channel的源码中,解析其内部的实现机制。


引言

在C#的并发编程中,Channel是一种非常强大的数据结构,用于在生产者和消费者之间进行通信。本文将首先通过一个实际的使用案例,介绍如何在C#中使用Channel,然后深入到Channel的源码中,解析其内部的实现机制。

使用案例一:文件遍历和过滤

在我们的使用案例中,我们需要遍历一个文件夹及其所有子文件夹,并过滤出具有特定扩展名的文件。在此,我们使用了C#的Channel来实现这个任务。

首先,我们创建了一个名为EnumerateFilesRecursively的方法,这个方法接受一个文件夹路径作为参数,并返回一个ChannelReader。这个方法中,我们创建了一个有界的Channel,然后在一个单独的任务中遍历指定的文件夹及其所有子文件夹,并将找到的每个文件的路径写入Channel。当遍历完成后,我们关闭Channel的写入端。

ChannelReader<string> EnumerateFilesRecursively(string root, int capacity = 100, CancellationToken token = default) {     var output = Channel.CreateBounded<string>(capacity);      async Task WalkDir(string path)     {         IEnumerable<string> files = null, directories = null;         try         {             files = Directory.EnumerateFiles(path);             directories = Directory.EnumerateDirectories(path);         }         catch (Exception ex)         {             Console.WriteLine($"An error occurred: {ex.Message}");         }          if (files != null)         {             foreach (var file in files)             {                 await output.Writer.WriteAsync(file, token);             }         }          if (directories != null)             await Task.WhenAll(directories.Select(WalkDir));     }      Task.Run(async () =>     {         await WalkDir(root);         output.Writer.Complete();     }, token);      return output.Reader; } 

然后,我们创建了一个名为FilterByExtension的方法,这个方法接受一个ChannelReader和一个扩展名集合作为参数,并返回一个ChannelReader。在这个方法中,我们创建了一个无界的Channel,然后在一个单独的任务中从输入的Channel中读取每个文件路径,检查其扩展名,如果满足条件,就将其转换为FileInfo并写入输出的Channel。当所有的文件都被处理后,我们关闭Channel的写入端。

ChannelReader<FileInfo> FilterByExtension(     ChannelReader<string> input, IReadOnlySet<string> exts, CancellationToken token = default) {     var output = Channel.CreateUnbounded<FileInfo>();      Task.Run(async () =>     {         try         {             await foreach (var file in input.ReadAllAsync(token).ConfigureAwait(false))             {                 var fileInfo = new FileInfo(file);                 if (exts.Contains(fileInfo.Extension))                     await output.Writer.WriteAsync(fileInfo, token).ConfigureAwait(false);             }         }         catch (Exception ex)         {             Console.WriteLine($"An error occurred: {ex.Message}");         }         finally         {             output.Writer.Complete();         }     }, token);      return output; } 

最后,在Main方法中,我们首先调用EnumerateFilesRecursively方法,遍历指定的文件夹,并得到一个文件路径的Channel。然后,调用FilterByExtension方法,过滤出具有特定扩展名的文件,并得到一个文件信息的Channel。最后,遍历这个Channel,打印出每个文件的全路径。

var fileSource = EnumerateFilesRecursively("D:\Program Files\.nuget\packages"); var sourceCodeFiles =     FilterByExtension(fileSource, new HashSet<string> { ".json", ".map", ".dll" });  await foreach (var file in sourceCodeFiles.ReadAllAsync().ConfigureAwait(false)) {     Console.WriteLine($"{file.FullName}"); }  Console.ReadKey(); 

在这个例子中,可以看到无论是文件的遍历还是过滤,都是并行进行的,并且这两个任务之间通过Channel进行了解耦,使得代码更加简洁和清晰。此外,由于Channel的异步特性,我们的程序在等待数据的时候不会阻塞,从而大大提高了程序的性能和响应性。

使用案例二:Excel读取与翻译内容

在我们的使用案例中,我们需要读取Excel文件,同时将读取的内容处理,调用对应的翻译服务进行翻译,并将翻译结果打印到控制台并存储到新的Excel文件中。为此,我们定义了一个名为ExcelTranslationProvider的类。

ExcelTranslationProvider类

ExcelTranslationProvider类是一个专门处理Excel文件翻译的工具。它主要使用了.NET的Channel来处理异步数据流,从而提高了翻译的效率。以下是该类的代码:

public class ExcelTranslationProvider : TranslationProvider {     public static Translater Translater { get; set; } = Translater.Azure;      public static II18NTermTranslateService TranslateService => TranslateServiceProvider.GetTranslateService(Translater);      private static ExcelTranslationParameters translationParameters;     public static async Task Translate(TranslationParameters parameters)     {         if (parameters is not ExcelTranslationParameters excelParameters)             throw new ArgumentException("Invalid parameters for Excel translation.");         translationParameters = excelParameters;          var translateText = TranslateText(excelParameters.Path);         var i = 1;         List<TranslationDto> list = new List<TranslationDto>();         await foreach (var text in translateText.ReadAllAsync().ConfigureAwait(false))         {             System.Console.WriteLine($"{i++}、" + text.TranslatText);             list.Add(text);         }         await ExcelUtil.SaveAsAsync(excelParameters.SavePath, list);     }      private static ChannelReader<TranslationDto> TranslateText(string path)     {         var output = Channel.CreateUnbounded<TranslationDto>();          _ = TranslateAndWriteToChannelAsync(path, output.Writer);          return output.Reader;     }      private static async Task TranslateAndWriteToChannelAsync(string path, ChannelWriter<TranslationDto> writer)     {         var query = await ExcelUtil.QueryAsync<TranslationDto>(path, translationParameters.Sheet);         var tasks = query.Select(async item =>         {             try             {                 var res = await TranslateService.TranslateSync(item.Name, "en-US");                 item.TranslatText = res;                 await writer.WriteAsync(item);             }             catch (Exception ex)             {                 System.Console.WriteLine($"An error occurred: {ex.Message}");             }         });          await Task.WhenAll(tasks);         writer.Complete();     } } 
  • Translater和TranslateService:这两个静态属性用于配置和获取翻译服务。Translater是一个枚举类型,表示可用的翻译服务提供者。默认的翻译服务是Azure。TranslateService是一个只读属性,返回一个实现了II18NTermTranslateService接口的翻译服务对象。这个对象是通过TranslateServiceProvider.GetTranslateService(Translater)方法获取的。

  • translationParameters:用于保存翻译参数,这些参数包括源文件的路径、目标文件的路径等。

  • Translate:这个方法首先检查传入的参数是否为ExcelTranslationParameters类型。然后,它调用TranslateText方法开始翻译过程。翻译的结果被保存在一个List列表中,然后写入到Excel文件。

  • TranslateText:它创建了一个无界Channel,并启动了一个异步任务来进行翻译操作并将结果写入到Channel中。无界Channel是一种可以存储任意数量元素的Channel,它是通过Channel.CreateUnbounded()方法创建的。创建Channel后,这个方法返回Channel的读取端,同时启动了一个异步任务TranslateAndWriteToChannelAsync来进行翻译并将结果写入到Channel的中。

  • TranslateAndWriteToChannelAsync:它负责从Excel文件中读取数据,进行翻译,并将翻译结果写入到Channel中。这个方法首先从Excel文件中读取数据,然后为每一条数据创建一个异步翻译任务。所有的翻译任务是并发执行的,使用了Task.WhenAll(tasks)来等待所有的翻译任务完成。完成所有的翻译任务后,这个方法调用writer.Complete()方法来表示没有更多的数据要写入到Channel中。