在面对 生产者-消费者 的场景下， 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<T> 创立的 channel 是一个有音讯下限的通道。
  • CreateUnbounded<T> 创立的 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...

译文链接：https://www.infoworld.com/art...

更多高质量干货：参见我的 GitHub: csharptranslate