同时从不同的客户端流式传输大数据

Question

这是一个体系结构和代码问题。我有很多源URL，其中包含来自许多不同客户端的巨大文件，我必须下载并保存在文件系统上。

我对RAM有硬件限制。因此，我想以字节为单位缓冲每个流，并且我认为为每次下载流启动一个线程是个好主意。

我已经添加了使用诸如此类的任务并行库来启动线程/任务的编码：

public Task RunTask(Action action)
{
    Task task = Task.Run(action);

    return task;
}

并且我为action参数传递以下方法：

public void DownloadFileThroughWebStream(WebClient webClient, Uri src, string dest, long buffersize)
{
    Stream stream = webClient.OpenRead(src);

    byte[] buffer = new byte[buffersize];
    int len;
    using (BufferedStream bufferedStream = new BufferedStream(stream))
    {
        using (FileStream fileStream = new FileStream(Path.GetFullPath(dest), FileMode.Create, FileAccess.Write))
        {
            while ((len = stream.Read(buffer, 0, buffer.Length)) > 0)
            {
                fileStream.Write(buffer, 0, len);
                fileStream.Flush();
            }
        }

    }
}

并且出于测试目的，我尝试通过为每个特定的下载启动线程/任务来从http uri下载一些资源：

[Test]
public async Task DownloadSomeStream()
{
    Uri uri = new Uri("http://mirrors.standaloneinstaller.com/video-sample/metaxas-keller-Bell.mpeg");

    List<Uri> streams = new List<Uri> { uri, uri, uri};

    List<Task> tasks = new List<Task>();

    var path = "C:\\TMP\\";
    //Create task for each of the streams from uri
    int c = 1;
    foreach (var uri in streams)
    {
        WebClient webClient = new WebClient();
        Task task = taskInitiator.RunTask(() => DownloadFileThroughWebStream(webClient, uri, Path.Combine(path,"File"+c), 8192));
        tasks.Add(task);
        c++;
    }
    Task allTasksHaveCompleted = Task.WhenAll(tasks);
    await allTasksHaveCompleted;
}

我收到以下异常：

System.IO.IOException: 'The process cannot access the file 'D:\TMP\File4' because it is being used by another process'

在线：

using (FileStream fileStream = new FileStream(Path.GetFullPath(dest), FileMode.Create, FileAccess.Write))

因此，有两个我无法理解的例外情况：

1. 为什么不允许写？以及另一个进程如何分配文件？

2. 当我仅添加3个url时为什么要保存file4，所以我只应该有文件：file1, file2, and file3吗？

此外，其他一些可能会引起思考的问题：

1. 关于要实现的目标，我正在采取的正确方法吗？我使用Task Parallel Library正确执行Task初始化了吗？

2. 任何技巧和窍门，最佳做法等？

Answer 1

我们可以创建可以执行下载的下载方法：

async Task DownloadFile(string url, string location, string fileName)
{
    using (var client = new WebClient())
    {
        await client.DownloadFileTaskAsync(url, $"{location}{fileName}");
    }
}

并且Task.Run()可以调用上述方法来执行文件的同时下载：

IList<string> urls = new List<string>()
{
    @"http://mirrors.standaloneinstaller.com/video-sample/metaxas-keller-Bell.mpeg",
    @"https://...",
    @"https://..."
};

string location = "D:";
Directory.CreateDirectory(location);

Task.Run(async () =>
{
    var tasks = urls.Select(url => 
    {
        var fileName = url.Substring(url.LastIndexOf('/'));
        return DownloadFile(url, location, fileName);
    }).ToArray();
    await Task.WhenAll(tasks);
}).GetAwaiter().GetResult();