如何在 C# 8 中使用 Channels

本文转载自微信公众号「码农读书」，中使作者码农读书。中使转载本文请联系码农读书公众号。中使

 在面对 生产者-消费者 的中使场景下， netcore 提供了一个新的中使命名空间 System.Threading.Channels 来帮助我们更高效的处理此类问题，有了这个 Channels 存在,中使 生产者 和 消费者 可以各自处理自己的任务而不相互干扰，有利于两方的中使并发处理，这篇文章我们就来讨论下如何使用 System.Threading.Channels。中使

Dataflow vs Channel

在 System.Threading.Tasks.Dataflow 命名空间下提供了一个数据流库，中使主要封装了 存储 和 处理 两大块,中使该库专注于 pipeline 处理，而 System.Threading.Tasks.Channels 主要专注于 存储 这块，中使从单一职责上来说，中使在 生产者-消费者 场景下，中使Channels 比 Dataflow 性能要高得多。中使

为什么要使用 Channels

可以利用 Channels 来实现 生产者和消费者 之间的中使解耦，大体上有两个好处：

生产者 和 消费者 是相互独立的，两者可以并行执行。 如果生产者不给力，可以创建多个的生产者，如果消费者不给力，可以创建更多的消费者。

总的来说，云服务器提供商在 生产者-消费者 模式下可以帮助我们提高应用程序的吞吐率。

安装 System.Threading.Channels

要想使用 Channel，需要用 nuget 引用 System.Threading.Channels 包，还可以通过 Visual Studio 2019 的 NuGet package manager 可视化界面安装 或者 通过 NuGet package manager 命令行工具输入以下命令：

dotnet add package System.Threading.Channels 

创建 channel

本质上来说，你可以创建两种类型的 channel，一种是有限容量的 bound channel，一种是无限容量的 unbound channel，接下来的问题是，如何创建呢?Channels 提供了两种 工厂方法 用于创建，如下代码所示：

CreateBounded 创建的 channel 是一个有消息上限的通道。 CreateUnbounded 创建的 channel 是一个无消息上限的通道。

下面的代码片段展示了如何创建 unbounded channel，并且只能存放 string 类型。

static void Main(string[] args)        {             var channel = Channel.CreateUnbounded<string>();        } 

对了，Bounded channel 还提供了一个 FullMode 属性，用于指定当 channel 已满时该如何对插入的服务器托管 message 进行处理，通常有四种做法。

Wait DropWrite DropNewest DropOldest

下面的代码片段展示了如何在 Bounded channel 上使用 FullMode。

static void Main(string[] args)       {            var channel = Channel.CreateBounded<string>(new BoundedChannelOptions(1000)           {                FullMode = BoundedChannelFullMode.Wait           });       } 

将消息写入到 channel

要想将 message 写入到 channel，可以使用 WriteAsync() 方法，如下代码所示:

static async Task Main(string[] args)        {             var channel = Channel.CreateBounded<string>(new BoundedChannelOptions(1000)            {                 FullMode = BoundedChannelFullMode.Wait            });            await channel.Writer.WriteAsync("Hello World!");        } 

从 channel 中读取消息

要想从 channel 中读取 message，可以使用 ReadAsync(),如下代码所示：

static async Task Main(string[] args)        {             var channel = Channel.CreateBounded<string>(new BoundedChannelOptions(1000)            {                 FullMode = BoundedChannelFullMode.Wait            });            while (await channel.Reader.WaitToReadAsync())            {                 if (channel.Reader.TryRead(out var message))                {                     Console.WriteLine(message);                }            }        } 

System.Threading.Channels 例子

下面是完整的代码清单，展示了如何从 channel 中读写 message。

class Program    {         static async Task Main(string[] args)        {             await SingleProducerSingleConsumer();            Console.ReadKey();        }        public static async Task SingleProducerSingleConsumer()        {             var channel = Channel.CreateUnbounded<int>();            var reader = channel.Reader;            for (int i = 0; i < 10; i++)            {                 await channel.Writer.WriteAsync(i + 1);            }            while (await reader.WaitToReadAsync())            {                 if (reader.TryRead(out var number))                {                     Console.WriteLine(number);                }            }        }    } 

可以看到，控制台中输出了数字 1-10,这些数字正是 Writer 写入到 channel 中的，对吧。

总的来说，要想使用 生产者-消费者 场景，有几种实现途径，比如：BlockingCollection 和 TPL Dataflow，但本篇介绍的 Channels 要比前面的两种性能更高，关于 Channels 更多的细节，我会在未来的文章中进行讨论，如果您现在想急于了解的话，可以参考MSDN：https://docs.microsoft.com/en-us/dotnet/api/system.threading.channels?view=netcore-3.0

译文链接：https://www.infoworld.com/article/3445156/how-to-use-systemthreadingchannels-in-net-core.html

很赞哦!（63）