引导多个生产者和消费者

Question

我有以下代码：

var channel = Channel.CreateUnbounded<string>();

var consumers = Enumerable
    .Range(1, 5)   
    .Select(consumerNumber =>
        Task.Run(async () =>
        {
            var rnd = new Random();
            while (await channel.Reader.WaitToReadAsync())
            {
                if (channel.Reader.TryRead(out var item))
                {
                    Console.WriteLine($"Consuming {item} on consumer {consumerNumber}");
                }
            }
        }));

var producers = Enumerable
    .Range(1, 5)    
    .Select(producerNumber =>
        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)));
            }
        }));

await Task.WhenAll(producers)
    .ContinueWith(_ => channel.Writer.Complete());

await Task.WhenAll(consumers);

其工作原理应该是这样的，但是我希望它在生产的同时进行消费。然而

await Task.WhenAll(producers)
    .ContinueWith(_ => channel.Writer.Complete());

阻止消费者运行直到完成，我想不出一种方法让它们都运行？

Answer 1

代码存在一些问题，包括忘记枚举producers和consumers可枚举项。IEnumerable是惰性求值的，因此在您实际使用 egforeach或枚举它之前ToList，不会生成任何内容。

ContinueWith只要使用得当，也没有什么问题。它绝对比使用异常作为控制流更好、更便宜。

通过使用一些常见的通道编码模式可以大大改进代码。

1. 生产者拥有并封装通道
2. 生产者仅公开读者

另外，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);

一般模式

这是使用 Channel 的管道阶段的一般模式 - 从 ChannelReader 读取输入，将其写入内部 Channel 并仅返回拥有的 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;
}