C# Channel的入门与应用

1. 入门

Channel 是微软在 .NET Core 3.0 以后推出的新的集合类型，该类型位于 System.Threading.Channels 命名空间下，具有异步 API 、高性能、线程安全等等的特点。目前，Channel 最主要的应用场景是生产者-消费者模型。如下图所示，生产者负责向队列中写入数据，消费者负责从队列中读出数据。在此基础上，通过增加生产者或者消费者的数目，对这个模型做进一步的扩展。我们平时使用到的 RabbitMQ 或者 Kafka，都可以认为是生产者-消费者模型在特定领域内的一种应用，甚至于我们还能从中读出一点广义上的读写分离的味道。

class Producer<T>
{
    private readonly Queue<T> _queue;
    public Producer(Queue<T> queue) { _queue = queue; }
}

class Consumer<T>
{
    private readonly Queue<T> _queue;
    public Consumer(Queue<T> queue) { _queue = queue; }
}

这个思路理论上是没有问题的，可惜实际操作起来槽点满满。譬如，生产者应该只负责写，消费者应该只负责读，可当你亲手把一个队列传递给它们的时候，想要保持这种职责上的纯粹属实是件困难的事情，更不必说，在使用队列的过程中，生产者会有队列“满”的忧虑，消费者会有队列“空”的烦恼，如果再考虑多个生产者、多个消费者、多线程/锁等等的因素，显然，这并不是一个简单的问题。为了解决这个问题，微软先后增加了 BlockingCollection 和 BufferBlock 两种数据结构，这里以前者为例，下面是一个典型的生产者-消费者模型：

var bc = new BlockingCollection<int>();

// 生产者
var producer = Task.Run(() => {
    for (var i = 0; i < Count; i++) {
        bc.Add(i);
        Console.WriteLine("Producer Write Item: {0}", i);
    }
    bc.CompleteAdding();
});

// 消费者
var consumer = Task.Run(() => {
    while (!bc.IsCompleted) {
        if (bc.TryTake(out var item)) {
            Console.WriteLine("Consumer Read Item: {0}", item);
        }
    }
});

await Task.WhenAll(producer, consumer);

测试了读写 10000 条数据的场景下，三种数据结构各自的性能表现，显而易见 Channel 的性能是最好的.

2. 应用

// 创建一个有限容量的 Channel
var boundedChannel = Channel.CreateBounded<int>(100);

// 创建一个无限容量的 Channel
var unboundedChannel = Channel.CreateUnbounded<string>();

在生产者-消费者模型中，一个容量有限的固定，一定会无可避免地出现队列“满”的情形，此时，我们就需要制定某种策略或者机制来完善整个模型。对于这个问题，Channel 的解决方案是 BoundedChannelFullMode ：

var boundedChannel = Channel.CreateBounded<string>(
    new BoundedChannelOptions(100) {FullMode = BoundedChannelFullMode.Wait});

这是一个枚举类型，事实上，它共有 Wait、DropNewest、DropOldest、DropWrite 四个取值，默认为 Wait。其中：

• Wait：当队列已满时，写入数据时会返回 false，直到队列内有空间时可以继续写入。
• DropNewest：移除最新的数据，即从队列尾部开始移除元素。
• DropOldest：移除最旧的数据，即从队列头部开始移除元素。
• DropWrite：可以写入数据，但是数据会被立即丢弃。

除了队列“满”或者队列“空”的问题，我们还考虑过多线程环境下的生产者-消费者模型可能会遇到的问题。值得庆幸的是， Channel 天生就支持多线程，我们可以通过 ChannelOptions 的 SingleWriter 和 SingleReader 来指定 Channel 是否是单一的消费者或者生产者，默认情况下，这两个值都是 false :

var boundedChannel = Channel.CreateBounded<string>(
    new BoundedChannelOptions(100) {
        SingleWriter = true,
        SingleReader = false,
        FullMode = BoundedChannelFullMode.Wait
});

通过以上代码片段，我们就可以创建出一个单生产者、多消费者的 Channel ，对于 Channel 而言，其最重要的两个成员分别是 Writer 和 Reader , 前者对应生产者，类型定义为：ChannelWriter；后者对应消费者，类型定义为：ChannelReader，这一次，我们做到了真正意义上的读写分离：

// 生产者生产数据
channel.Writer.TryWrite("大漠孤烟直，长河落日圆。");

// 消费者消费数据
// 模式一：一次读一个
while (await channel.Reader.WaitToReadAsync())
{
    while (channel.Reader.TryRead(out var item))
    {
        // 在这里写具体的处理逻辑
    }
}

// 模式二：一次全部读出来
while (await channel.Reader.WaitToReadAsync())
{
    await foreach (var item in channel.Reader.ReadAllAsync())
    {
        // 在这里写具体的处理逻辑
    }
}

下面这三个方法做了一件什么样的事情呢？我个人以为，这其实就是我们上面提到的数据流，首先，我们通过 GetFiles() 方法获得指定目录内的文件信息；然后，这些信息交给 Analyse() 方法去做处理，这里做的事情是统计出 markdown 格式文件的字符串，以及筛选出那些非 markdown 格式的文件或者子目录；最后，通过 Merge() 函数，我们将上一步的结果进行汇总输出。

// GetFiles
Task<Channel<string>> GetFiles(string root) {
    var filePathChannel = Channel.CreateUnbounded<string>();
    var directoryInfo = new DirectoryInfo(root);

    foreach (var file in directoryInfo.EnumerateFileSystemInfos()) {
        filePathChannel.Writer.TryWrite(file.FullName);
    }

    filePathChannel.Writer.Complete();
    return Task.FromResult(filePathChannel);
}

// Analyse
async Task<Channel<string>[]> Analyse(Channel<string> rootChannel) {
    var counterChannel = Channel.CreateUnbounded<string>();
    var errorsChannel = Channel.CreateUnbounded<string>();

    while (await rootChannel.Reader.WaitToReadAsync()) {
        await foreach (var filePath in rootChannel.Reader.ReadAllAsync()) {
            var fileInfo = new FileInfo(filePath);
            if (fileInfo.Extension == ".md") {
                var totalWords = File.ReadAllText(filePath).Length;
                counterChannel.Writer.TryWrite($"文章 [{fileInfo.Name}] 共 {totalWords} 个字符.");
            } else {
                errorsChannel.Writer.TryWrite($"路径 [{filePath}] 是文件夹或者格式不正确.");
            }
        }
    }

    counterChannel.Writer.Complete();
    errorsChannel.Writer.Complete();

    return new Channel<string>[] { counterChannel, errorsChannel };
}

// Merge
async Task<Channel<string>> Merge(params Channel<string>[] channels) {
    var mergeTasks = new List<Task>();
    var outputChannel = Channel.CreateUnbounded<string>();

    foreach (var channel in channels) {
        var thisChannel = channel;
        var mergeTask = Task.Run(async () => {
            while (await thisChannel.Reader.WaitToReadAsync()) {
                await foreach (var item in thisChannel.Reader.ReadAllAsync()) {
                    outputChannel.Writer.TryWrite(item);
                }
            }
        });

        mergeTasks.Add(mergeTask);
    }

    await Task.WhenAll(mergeTasks);
    outputChannel.Writer.Complete();

    return outputChannel;
}

// Run
var filePathChannel = await GetFiles(@"/hugo-blog/content/posts/");
var analysedChannels = await Analyse(filePathChannel);
var mergedChannel = await Merge(analysedChannels);
while (await mergedChannel.Reader.WaitToReadAsync()) {
    await foreach (var item in mergedChannel.Reader.ReadAllAsync()) {
        Console.WriteLine(item);
    }
}

从某种意义上来讲，这是一种“分治”策略，即：把一个大任务分解为若干个小任务，再将这些小任务的结果合并起来。很多年前，我曾在一本讲并行编程的书上见过类似的代码片段，那个时候我已经对 Google 的 MapReduce 略有耳闻，后来又接触到了 Parallel ，我突然意识到，如果 Map() 和 Reduce() 两个函数运行在一台远程服务器上，那么这个过程可以认为是 RPC，而运行在远程服务器上的这些函数，其实是在并行地执行着某种运算，那么这个过程可以认为是并行计算。当这些并行计算，使用的是世界各地的可伸缩计算资源时，那么这个过程其实就是云计算。所以说，写作这个过程还是挺有意思的，对不对？

原文链接

---