如何使用 Channel 类来创建一个生产者-消费者模型

.NET 中 Channel 类简单使用

Channel 是干什么的

The System.Threading.Channels namespace provides a set of synchronization data structures for passing data between producers and consumers asynchronously. The library targets .NET Standard and works on all .NET implementations.
Channels are an implementation of the producer/consumer conceptual programming model.
以上是微软官方的解释 channels。用中文说的话就是这个类提供了在生产者跟消费者之间异步传统数据的能力，简单来说可以认为是一个内存消息队列。

示例 1

下面是一个简单的示例，说明如何使用 Channel 类来创建一个生产者-消费者模型：

    static async Task Main(string[] args)
    {
        var channel = Channel.CreateUnbounded<int>();

        var producer = Task.Run(async () =>
        {
            for (int i = 0; i < 10; i++)
            {
                await channel.Writer.WriteAsync(i);
                await Task.Delay(1000); // 模拟生产者需要一些时间来生成数据
            }

            channel.Writer.Complete();
        });

        var consumer = Task.Run(async () =>
        {
            await foreach (var item in channel.Reader.ReadAllAsync())
            {
                Console.WriteLine($"消费者接收到: {item}");
            }
        });

        await Task.WhenAll(producer, consumer);
    }

在这个例子中，我们创建了一个无界的通道，然后创建了两个任务，一个是生产者，一个是消费者。生产者每秒生成一个数字，然后写入通道。消费者从通道中读取数据并打印出来。当生产者完成写入后，它会调用 channel.Writer.Complete() 来通知消费者没有更多的数据可以读取。

示例 2

你可以使用 Channel.CreateBounded(capacity) 方法来创建一个有界的通道，其中 capacity 参数指定了通道的容量。当通道满时，尝试写入的操作将会阻塞，直到有空间可用。

    static async Task Main(string[] args)
    {
        var channel = Channel.CreateBounded<int>(5); // 创建一个容量为5的有界通道

        var producer = Task.Run(async () =>
        {
            for (int i = 0; i < 10; i++)
            {
                await channel.Writer.WriteAsync(i);
                Console.WriteLine($"生产者生成了: {i}");
                await Task.Delay(1000); // 模拟生产者需要一些时间来生成数据
            }

            channel.Writer.Complete();
        });

        var consumer = Task.Run(async () =>
        {
            await foreach (var item in channel.Reader.ReadAllAsync())
            {
                Console.WriteLine($"消费者接收到: {item}");
                await Task.Delay(2000); // 模拟消费者需要一些时间来处理数据
            }
        });

        await Task.WhenAll(producer, consumer);
    }

在这个例子中，我们创建了一个容量为5的有界通道。生产者每秒生成一个数字，然后写入通道。消费者从通道中读取数据并打印出来，但消费者处理数据的速度比生产者慢，所以当通道满时，生产者的 WriteAsync 操作将会阻塞，直到消费者读取了一些数据，使得通道有空间可用。

示例 3

下面是一个示例，展示了如何在多个生产者和消费者之间共享一个通道：

    static async Task Main(string[] args)
    {
        var channel = Channel.CreateUnbounded<int>();

        // 创建两个生产者
        var producer1 = Produce(channel.Writer, id: 1);
        var producer2 = Produce(channel.Writer, id: 2);

        // 创建两个消费者
        var consumer1 = Consume(channel.Reader, id: 1);
        var consumer2 = Consume(channel.Reader, id: 2);

        // 等待所有生产者和消费者完成
        await Task.WhenAll(producer1, producer2, consumer1, consumer2);
    }

    static async Task Produce(ChannelWriter<int> writer, int id)
    {
        for (int i = 0; i < 10; i++)
        {
            await writer.WriteAsync(i);
            Console.WriteLine($"生产者{id}生成了: {i}");
            await Task.Delay(1000); // 模拟生产者需要一些时间来生成数据
        }

        writer.Complete();
    }

    static async Task Consume(ChannelReader<int> reader, int id)
    {
        await foreach (var item in reader.ReadAllAsync())
        {
            Console.WriteLine($"消费者{id}接收到: {item}");
            await Task.Delay(2000); // 模拟消费者需要一些时间来处理数据
        }
    }

在这个例子中，我们创建了两个生产者和两个消费者，它们都共享同一个通道。这是一个非常重要使用模式。因为当我们使用消息队列的时候往往会有多个生产者跟多个消费者。我们可以通过控制生产者生产的速度来控制推入队列的数据量。我们还可以通过控制消费者的数量来控制消费数据的速度，从而来调节系统的流量，达到消峰填谷的作用。

总结

Channel 类是 .NET CORE 3.0 后新加入的类。为我们提供了便利的生产者/消费者模式实现方案。相当于是一个进程内的内存队列，而且它没有持久化，纯内存操作，性能是非常非常高的。当我们面对真正的高并发的时候可以为我们的系统提供吞吐量。当然代价是内存跟放弃一些实时性。