代码有几个问题，包括忘记枚举producers和consumers枚举。IEnumerable被懒惰地评估，所以在你用例如foreachor实际枚举它之前ToList，什么都不会生成。
ContinueWith使用得当也没什么问题。这绝对比使用异常作为控制流更好、更便宜。
通过使用一些常见的 Channel 编码模式，可以对代码进行很多改进。
- 生产者拥有并封装通道
- 生产者只公开读者
另外，ContinueWith这是一个很好的选择来表示 ChannelWriter 的完成，因为我们根本不关心哪个线程会这样做。如果有的话，我们宁愿使用“工作者”线程之一来避免线程切换。
假设生产者函数是：
async Task Produce(ChannelWriter<string> writer, int producerNumber)
{
return Task.Run(async () =>
{
var rnd = new Random();
for (var i = 0; i < 10; i++)
{
var t = $"Message {i}";
Console.WriteLine($"Producing {t} on producer {producerNumber}");
await channel.Writer.WriteAsync(t);
await Task.Delay(TimeSpan.FromSeconds(rnd.Next(3)));
}
}
}
制片人
生产者可以是：
ChannelReader<string> ProduceData(int dop)
{
var channel=Channel.CreateUnbounded<string>();
var writer=channel.Writer;
var tasks=Enumerable.Range(0,dop)
.Select(producerNumber => Produce(producerNumber))
.ToList();
_ =Task.WhenAll(tasks).ContinueWith(t=>writer.TryComplete(t.Exception));
.
return channel.Reader;
}
完成和错误传播
注意这一行：
_ =Task.WhenAll(tasks).ContinueWith(t=>writer.TryComplete(t.Exception));
这表示一旦生产者完成，编写者本身就应该完成，但可能会引发任何异常。延续运行在哪个线程上并不重要，因为它除了 call 之外什么都不做TryComplete。
更重要的是，t=>writer.TryComplete(t.Exception)将工作异常传播给下游消费者。否则消费者永远不会知道出了什么问题。如果您有一个数据库使用者，您希望它避免在源中止时完成任何更改。
消费者
消费者方法可以是：
async Task Consume(ChannelReader<string> reader,int dop,CancellationToken token=default)
{
var tasks= Enumerable
.Range(1, dop)
.Select(consumerNumber =>
Task.Run(async () =>
{
await foreach(var item in reader.ReadAllAsync(token))
{
Console.WriteLine($"Consuming {item} on consumer {consumerNumber}");
}
}));
await Task.WhenAll(tasks);
}
在这种情况下await Task.WhenAll(tasks);，枚举工作任务，从而启动它们。
生成所有生成的消息不需要其他任何东西。当所有生产者完成时，Channel.Reader就完成了。发生这种情况时，ReadAllAsync将继续向消费者提供所有剩余的消息并退出。
作品
结合这两种方法很简单：
var reader=Produce(10);
await Consume(reader);
一般模式
这是使用 Channels 的管道阶段的一般模式 - 从 ChannelReader 读取输入，将其写入内部 Channel 并仅返回拥有的通道的 Reader。这样舞台就拥有了通道，这使得完成和错误处理变得更加容易：
static ChannelReader<TOut> Crunch<Tin,TOut>(this ChannelReader<Tin>,int dop=1,CancellationToken token=default)
{
var channel=Channel.CreateUnbounded<TOut>();
var writer=channel.Writer;
var tasks=Enumerable.Range(0,dop)
.Select(async i=>Task.Run(async ()=>
{
await(var item in reader.ReadAllAsync(token))
{
try
{
...
await writer.WriteAsync(msg);
}
catch(Exception exc)
{
//Handle the exception and keep processing messages
}
}
},token));
_ =Task.WhenAll(tasks)
.ContinueWith(t=>writer.TryComplete(t.Exception));
return channel.Reader;
}
这允许将多个“阶段”链接在一起以形成管道：
var finalReader=Producer(...)
.Crunch1()
.Crunch2(10)
.Crunch3();
await foreach(var result in finalReader.ReadAllAsync())
{
...
}
生产者和消费者方法可以以相同的方式编写，例如允许创建数据导入管道：
var importTask = ReadFiles<string>(somePath)
.ParseCsv<string,Record[]>(10)
.ImportToDb<Record>(connectionString);
await importTask;
和ReadFiles
static ChannelReader<string> ReadFiles(string folder)
{
var channel=Channel.CreateUnbounded<string>();
var writer=channel.Writer;
var task=Task.Run(async ()=>{
foreach(var path in Directory.EnumerateFiles(folder,"*.csv"))
{
await writer.WriteAsync(path);
}
});
task.ContinueWith(t=>writer.TryComplete(t.Exception));
return channel.Reader;
}
.NET 6 Parallel.ForEachAsync 的更新
现在生产中支持 .NET 6，可以使用Parallel.ForEachAsync将并发使用者简化为：
static ChannelReader<TOut> Crunch<Tin,TOut>(this ChannelReader<Tin>,
int dop=1,CancellationToken token=default)
{
var channel=Channel.CreateUnbounded<TOut>();
var writer=channel.Writer;
var dop=new ParallelOptions {
MaxDegreeOfParallelism = dop,
CancellationToken = token
};
var task=Parallel.ForEachAsync(
reader.ReadAllAsync(token),
dop,
async item =>{
try
{
...
await writer.WriteAsync(msg);
}
catch(Exception exc)
{
//Handle the exception and keep processing messages
}
});
task.ContinueWith(t=>writer.TryComplete(t.Exception));
return channel.Reader;
}