Parallel.Foreach产生太多线程的方式

Question

问题

虽然我将在这里谈论的代码我在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.

在某种程度上并行完成并不是件好事吗？不幸的是,我的阅读和解析功能都不是"async-workflows-ready".

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

任务

我做的第一件事就是设置一个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任务时,它会等待,等待,等待由于大量网络延迟而坚持阻止的该死的最后一个任务.这个不好!

那么,你会如何解决这个问题呢？我很欣赏能够查看不同的解决方案,包括异步工作流(在这种情况下,如何调整我的非异步函数),并行扩展,奇怪的并行模式等.

谢谢.

Answer 1

ParallelOptions.MaxDegreeOfParallelism 限制并行方法调用运行的并发操作数

Answer 2

使用'async'可以让你在没有刻录线程的情况下进行I/O绑定工作,而各种I/O调用是'海上',所以这将是我的第一个建议.将代码转换为异步应该是直截了当的,通常是沿着

• 将每个函数体包裹起来async{...},return在必要时添加
• 创建通过库中尚未存在的任何I/O原语的异步版本 Async.FromBeginEnd
• 形式的切换通话let r = Foo()至let! r = AsyncFoo()
• 使用Async.Parallel到5000名异步对象转换成并行运行的一个异步

这样做有各种各样的教程; 一个这样的网络直播在这里.

Answer 3

您确定您的个人任务是否及时完成？我认为,无论Parallel.ForEach和Task类已经使用.NET线程池.任务通常应该是短期工作项,在这种情况下,线程池只会产生少量实际线程,但是如果你的任务没有进展并且还有其他任务排队,那么使用的线程数将稳步增加到最大值(默认值为250 /处理器在.NET 2.0 SP1中,但在不同版本的框架下有所不同).值得注意的是(至少在.NET 2.0 SP1中)新的线程创建被限制为每秒2个新线程,因此达到您所看到的线程数量表明任务未在短时间内完成时间(因此可能不完全准确地指责Parallel.ForEach).

我认为Brian建议使用async工作流是一个很好的建议,特别是如果长期任务的来源是IO,因为async在IO完成之前会将你的线程返回到线程池.另一种选择是简单地接受你的任务没有快速完成并允许产生许多线程(可以通过使用System.Threading.ThreadPool.SetMaxThreads在某种程度上控制) - 根据你的情况,你可能没有什么大不了的.很多线程.