如何确保按启动顺序保存多异步下载的数据？

Question

我正在编写一个基本的Http (HLS)下载程序，在这里，我将按照“#”指定的时间间隔重新下载m3u8媒体播放列表，然后在*.ts段可用时下载它们。

这就是第一次下载m3u8媒体播放列表时的样子。

#EXTM3U
#EXT-X-VERSION:3
#EXT-X-TARGETDURATION:12
#EXT-X-MEDIA-SEQUENCE:1
#EXTINF:7.975,
http://website.com/segment_1.ts
#EXTINF:7.941,
http://website.com/segment_2.ts
#EXTINF:7.975,
http://website.com/segment_3.ts

我想同时下载这些*.ts片段，同时使用HttpClient异步/等待。这些段没有相同的大小，因此即使"segment_1.ts“的下载是先启动的，但它可能会在其他两个段之后完成。

这些片段都是一个大视频的一部分，所以重要的是下载片段的数据是按照开始的顺序写的，而不是按照它们完成的顺序写的。

如果分段是一个接一个地下载的话，下面的代码工作得非常好，但是当多个段同时下载时就不行了，因为有时它们没有按照启动的顺序完成。

我考虑使用Task.WhenAll，它保证了正确的顺序，但我不想不必要地将下载的段保存在内存中，因为它们可以有几兆字节的大小。如果"segment_1.ts“的下载首先完成，那么应该立即将其写入磁盘，而不必等待其他段完成。将所有*.ts段写入单独的文件并最终加入它们也不是一种选择，因为它需要双磁盘空间，总视频可以是几千兆字节大小。

我不知道该怎么做，我想知道是否有人能帮我。我正在寻找一种不需要手动创建线程或长期阻塞ThreadPool线程的方法。

一些代码和异常处理已经被删除，以便更容易地看到正在发生的事情。

// Async BlockingCollection from the AsyncEx library
private AsyncCollection<byte[]> segmentDataQueue = new AsyncCollection<byte[]>();

public void Start()
{
    RunConsumer();
    RunProducer();
}

private async void RunProducer()
{
    while (!_isCancelled)
    {
        var response = await _client.GetAsync(_playlistBaseUri + _playlistFilename, _cts.Token).ConfigureAwait(false);
        var data = await response.Content.ReadAsStringAsync().ConfigureAwait(false);

        string[] lines = data.Split(new string[] { "\n" }, StringSplitOptions.RemoveEmptyEntries);
        if (!lines.Any() || lines[0] != "#EXTM3U")
            throw new Exception("Invalid m3u8 media playlist.");

        for (var i = 1; i < lines.Length; i++)
        {
           var line = lines[i];
           if (line.StartsWith("#EXT-X-TARGETDURATION"))
           {
               ParseTargetDuration(line);
           }
           else if (line.StartsWith("#EXT-X-MEDIA-SEQUENCE"))
           {
               ParseMediaSequence(line);
           }
           else if (!line.StartsWith("#"))
           {
               if (_isNewSegment)
               {

                   // Fire and forget
                   DownloadTsSegment(line);

               }
           }
        }

        // Wait until it's time to reload the m3u8 media playlist again
        await Task.Delay(_targetDuration * 1000, _cts.Token).ConfigureAwait(false);
    }
}

// async void. We never await this method, so we can download multiple segments at once
private async void DownloadTsSegment(string tsUrl)
{
    var response = await _client.GetAsync(tsUrl, _cts.Token).ConfigureAwait(false);
    var data = await response.Content.ReadAsByteArrayAsync().ConfigureAwait(false);

    // Add the downloaded segment data to the AsyncCollection
    await segmentDataQueue.AddAsync(data, _cts.Token).ConfigureAwait(false);
}

private async void RunConsumer()
{
    using (FileStream fs = new FileStream(_filePath, FileMode.Create, FileAccess.Write, FileShare.Read))
    {
        while (!_isCancelled)
        {
            // Wait until new segment data is added to the AsyncCollection and write it to disk
            var data = await segmentDataQueue.TakeAsync(_cts.Token).ConfigureAwait(false);
            await fs.WriteAsync(data, 0, data.Length).ConfigureAwait(false);
        }
    }
}

Answer 1

我认为你根本不需要生产者/消费者排队。不过，我认为你应该避免“失火而忘”。

您可以同时启动它们，并在它们完成时对它们进行处理。

首先，定义如何下载单个段：

private async Task<byte[]> DownloadTsSegmentAsync(string tsUrl)
{
  var response = await _client.GetAsync(tsUrl, _cts.Token).ConfigureAwait(false);
  return await response.Content.ReadAsByteArrayAsync().ConfigureAwait(false);
}

然后添加对播放列表的解析，从而生成段下载列表(所有这些都在进行中)：

private List<Task<byte[]>> DownloadTasks(string data)
{
  var result = new List<Task<byte[]>>();
  string[] lines = data.Split(new string[] { "\n" }, StringSplitOptions.RemoveEmptyEntries);
  if (!lines.Any() || lines[0] != "#EXTM3U")
    throw new Exception("Invalid m3u8 media playlist.");
  ...
           if (_isNewSegment)
           {
             result.Add(DownloadTsSegmentAsync(line));
           }
  ...
  return result;
}

通过将此列表写入文件，一次(按顺序)使用该列表：

private async Task RunConsumerAsync(List<Task<byte[]>> downloads)
{
  using (FileStream fs = new FileStream(_filePath, FileMode.Create, FileAccess.Write, FileShare.Read))
  {
    for (var task in downloads)
    {
      var data = await task.ConfigureAwait(false);
      await fs.WriteAsync(data, 0, data.Length).ConfigureAwait(false);
    }
  }
}

然后和一个制片人一起开场：

public async Task RunAsync()
{
  // TODO: consider CancellationToken instead of a boolean.
  while (!_isCancelled)
  {
    var response = await _client.GetAsync(_playlistBaseUri + _playlistFilename, _cts.Token).ConfigureAwait(false);
    var data = await response.Content.ReadAsStringAsync().ConfigureAwait(false);

    var tasks = DownloadTasks(data);
    await RunConsumerAsync(tasks);

    await Task.Delay(_targetDuration * 1000, _cts.Token).ConfigureAwait(false);
  }
}

请注意，此解决方案确实同时运行所有下载，这可能会导致内存压力。如果这是一个问题，我建议您重组使用TPL数据流，它已经内置支持节流。

Answer 2

为每个下载分配一个序列号。将结果放入Dictionary<int, byte[]>中。每次下载完成后，它都会添加自己的结果。

然后，它检查是否有要写入磁盘的段：

while (dict.ContainsKey(lowestWrittenSegmentNumber + 1)) {
 WriteSegment(dict[lowestWrittenSegmentNumber + 1]);
 lowestWrittenSegmentNumber++;
}

这样，所有的段都按照顺序和缓冲结束在磁盘上。

RunConsumer();
RunProducer();

确保使用async Task，以便可以等待使用await Task.WhenAll(RunConsumer(), RunProducer());完成。但是你不应该再需要RunConsumer了。