c# – 如何聚合来自异步生成器的数据并将其写入文件？

AsyncProducerConsumerQueue<byte[]> _queue;
Stream _stream;

// producer
async Task ReceiveAsync(CancellationToken token)
{
    while (true)
    {
       var list = new List<byte>();
       while (true)
       {
           token.ThrowIfCancellationRequested(token);
           var packet = await _device.ReadAsync(token);
           list.Add(packet);
           if (list.Count == 1000)
               break;
       }
       // push next batch
       await _queue.EnqueueAsync(list.ToArray(),token);
    }
}

// consumer
async Task LogAsync(CancellationToken token)
{
    Task previousFlush = Task.FromResult(0); 
    CancellationTokenSource cts = null;
    while (true)
    {
       token.ThrowIfCancellationRequested(token);
       // get next batch
       var nextBatch = await _queue.DequeueAsync(token);
       if (!previousFlush.IsCompleted)
       {
           cts.Cancel(); // cancel the previous flush if not ready
           throw new Exception("failed to flush on time.");
       }
       await previousFlush; // it's completed,observe for any errors
       // start flushing
       cts = CancellationTokenSource.CreateLinkedTokenSource(token);
       previousFlush = _stream.WriteAsync(nextBatch,nextBatch.Count,cts.Token);
    }
}

如果您不想让记录器失败,而是希望取消刷新并继续下一批,则可以对此代码进行最小的更改.

回应@ l3arnon评论：

1. A packet is not a byte,it’s byte[]. 2. You haven’t used the OP’s ToHexString. 3. AsyncProducerConsumerQueue is much less robust and
   tested than .Net’s TPL Dataflow. 4. You await previousFlush for errors
   just after you throw an exception which makes that line redundant.
   etc. In short: I think the possible added value doesn’t justify this
   very complicated solution.

>“数据包不是字节,它是字节[]” – 数据包是一个字节,这从OP的代码中很明显：buffer [i] = await device.ReadAsync().然后,一批数据包是byte [].
>“你还没有使用OP的ToHexString.” – 目标是展示如何使用Stream.WriteAsync本身接受取消令牌,而不是WriteLineAsync,它不允许取消.将ToHexString与Stream.WriteAsync一起使用并仍然利用取消支持是微不足道的：

var hexBytes = Encoding.ASCII.GetBytes(ToHexString(nextBatch) + 
    Environment.NewLine);
_stream.WriteAsync(hexBytes,hexBytes.Length,token);

>“AsyncProducerConsumerQueue的稳健性和测试性都不如.Net的TPL数据流” – 我不认为这是一个确定的事实.但是,如果OP关心它,他可以使用常规的BlockingCollection,它不会阻塞生产者线程.在等待下一批时阻止使用者线程是可以的,因为写入是并行完成的.与此相反,您的TPL Dataflow版本带有一个冗余CPU和锁密集操作：使用logAction.Post(数据包)将数据从生产者管道移动到写入器pipleline,逐字节.我的代码不这样做.>“在抛出导致该行冗余的异常之后,您等待previousFlush出错.” – 这条线不是多余的.也许,你错过了这一点：当previousFlush.IsFaulted或previousFlush.IsCancelled也为true时,previousFlush.IsCompleted可能为true.因此,等待previousFlush与之相关,以观察已完成任务上的任何错误(例如,写入失败),否则将丢失.