C# Channels

JusterZhu

发布于 2023-11-27 14:44:12

2210

发布于 2023-11-27 14:44:12

文章被收录于专栏：JusterZhuJusterZhu

通过使用异步编程，我们可以提高应用程序的响应性和吞吐量。C# 提供了一些内置的方案来处理异步编程，例如 async/await 关键字和 Task 类。然而，有时候我们需要处理更复杂的场景，比如处理流式数据或者实现生产者/消费者模型。这就是为什么 .NET Core 3.0 引入了 System.Threading.Channels 的地方。

在本文中，我们将详细介绍如何使用 C# Channels 进行异步编程。让我们先看看 Channels 是什么。

Channels 简介

Channels 提供了一种通信机制，允许生产者和消费者之间安全、可靠地交换信息，即使它们在不同的执行线程上运行。并且，Channels 已经完全集成到 .NET 的异步模型中，支持 async/await 关键字。

创建和使用 Channel

使用C# Channels演示生产者/消费者模式。

using System;
using System.Threading.Channels;
using System.Threading.Tasks;

class Program
{
    static async Task Main(string[] args)
    {
        var channel = Channel.CreateUnbounded<int>();
        
        var producer = Task.Run(async () =>
        {
            for (var i = 0; i < 10; i++)
            {
                Console.WriteLine($"Producing: {i}");
                await channel.Writer.WriteAsync(i);
            }

            // 编写完成后调用Complete()
            channel.Writer.Complete();
        });

        var consumer = Task.Run(async () =>
        {
            // 使用await foreach读取channel中的所有数据
            await foreach (var item in channel.Reader.ReadAllAsync())
            {
                Console.WriteLine($"Consuming: {item}");
            }
        });

        // 等待任务完成
        await Task.WhenAll(producer, consumer);
    }
}

在此示例中，创建了一个unbounded channel来传输int类型的数据。然后创建了两个任务：一个生产者和一个消费者。生产者每次写入一个数字到channel，消费者则从channel中读取这些数字并打印出来。

在生产者将所有数据写入channel后，必须调用channel.Writer.Complete()来表示不再有更多的数据要写入。消费者可以通过channel.Reader.ReadAllAsync()读取channel中的所有数据，直到channel.Writer.Complete()被调用。

其他使用方式

除了前述的基础使用方式，C# Channels 也支持更复杂和高级的用法。下面是一些示例：

有界和无界通道: 在创建 Channel 时，你可以选择创建无界（unbounded）或有界（bounded）通道。无界通道可以接收无限数量的数据，而有界通道则只能接受指定数量的数据。

var unboundedChannel = Channel.CreateUnbounded<int>();
var boundedChannel = Channel.CreateBounded<int>(5); // 最多可接收5个元素

异步等待通道空闲: 如果通道已满（对于有界通道），使用 WriteAsync 方法将使当前线程异步等待，直到有足够的空间可供写入新的元素。

await boundedChannel.Writer.WriteAsync(42);

批量处理: 你可以实现一种模式，让消费者一次处理多个元素，而不是单独处理每个元素。这样做可能会提高效率和性能。

IAsyncEnumerable<int> batch = channel.Reader.ReadBatchAsync(10); // 读取10个元素以进行批处理

配合 CancellationToken 使用: 你可以使用 CancellationToken 来取消读取或写入操作。

var cts = new CancellationTokenSource();
await channel.Reader.WaitToReadAsync(cts.Token); // 使用 CancellationToken 取消读取操作

结合 Dataflow 库使用: 这个库提供了与 Channels 类似的功能，但它提供了更加复杂和强大的数据处理能力，比如数据并行、流分割等。Channels 可以很好地与 Dataflow 库一起工作，以处理更复杂的场景。

其他实现

Channels 非常适合实现一些特定的设计模式，尤其是与并发和异步编程相关的设计模式。以下是其中的一部分：

生产者消费者模式: 这是 Channels 最直接且显而易见的用途。Channel 提供了一种机制，允许一个或多个生产者线程生成数据，并由一个或多个消费者线程进行处理。
流水线模式: 在此模式中，每个步骤都在单独的线程（可能是在不同的物理设备上）上进行，并且使用 Channel 建立数据传输线，从一个步骤传递到下一个步骤。
发布/订阅模式: 通过使用 Channel，可以创建一个消息主题，生产者将消息发布到主题中，然后任何感兴趣的消费者都可以订阅该主题并接收消息。
工作队列模式: 这个模式涉及到一个任务队列和多个工作线程。工作线程从队列中取出任务并执行，而队列则通过 Channel 实现。