parallel.foreach生成方式线程过多

问题

虽然我将在这里讨论的代码是我用f编写的,但它是基于.NET 4框架,而不是具体取决于f的任何特殊性(至少看起来是这样!).

我的磁盘上有一些数据,我应该从网络上更新,将最新版本保存到磁盘上:

type MyData =
    { field1 : int;
      field2 : float }

type MyDataGroup =
    { Data : MyData[];
      Id : int }

// load : int -> MyDataGroup
let load dataId =
    let data = ... // reads from disk
    { Data = data;
      Id = dataId }

// update : MyDataGroup -> MyDataGroup
let update dg =
    let newData = ... // reads from the network and process
                      // newData : MyData[]

    { dg with Data = dg.Data
                     |> Seq.ofArray
                     |> Seq.append newData
                     |> processDataSomehow
                     |> Seq.toArray }

// save : MyDataGroup -> unit
let save dg = ... // writes to the disk

let loadAndSaveAndUpdate = load >> update >> save

问题在于 loadAndSaveAndUpdate 所有的数据,我必须执行函数 许多的 时代:

{1 .. 5000} |> loadAndSaveAndUpdate

每一步都可以

• 一些磁盘IO,
• 一些数据处理,
• 一些网络IO(可能存在大量延迟)
• 更多的数据处理,
• 还有一些磁盘IO。

在某种程度上,并行完成这项工作不是很好吗?不幸的是,我的读取和解析函数都不是“异步工作流就绪”。

我提出的第一个(不是很好)解决方案

任务

我做的第一件事就是 Task[] 然后全部开始:

let createTask id = new Task(fun _ -> loadAndUpdateAndSave id)
let tasks = {1 .. 5000}
            |> Seq.map createTask
            |> Seq.toArray

tasks |> Array.iter (fun x -> x.Start())
Task.WaitAll(tasks)

然后按ctrl+esc查看它使用了多少线程。15,17,…,35,…,170,……直到终止应用程序!出了点问题。

平行

我做了几乎相同的事情,但使用 Parallel.ForEach(...) 结果是一样的:很多,很多,很多线。

有效的解决方案…有点

然后我决定只开始 n 线程, Task.WaitAll(of them) 然后其他 n ,直到没有其他任务可用。

这是可行的,但问题是当它完成处理后,比如, n-1 任务,它将等待,等待,等待该死的最后一个任务,因为许多网络延迟坚持阻塞。这不好!

所以, 你将如何解决这个问题 是吗?我很高兴能看到不同的解决方案,包括异步工作流(在本例中是如何调整非异步函数)、并行扩展、奇怪的并行模式等。

谢谢。