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，而运行在远程服务器上的这些函数，其实是在并行地执行着某种运算，那么这个过程可以认为是并行计算。当这些并行计算，使用的是世界各地的可伸缩计算资源时，那么这个过程其实就是云计算。所以说，写作这个过程还是挺有意思的，对不对？
