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任務時，它會等待，等待，等待最後一個由於大量網路延遲而堅持阻塞的任務。不是很好！

那麼，你將如何解決這個問題呢？我很樂意查看不同的解決方案，包括非同步工作流（以及在這種情況下如何調整我的非非同步函式）、並行擴展、奇怪的並行模式等。

謝謝。

ParallelOptions.MaxDegreeOfParallelism 限制並行方法呼叫執行的並發操作數

使用“非同步”將使您能夠在各種 I/O 呼叫處於“海上”狀態時執行 I/O 綁定工作而不會燒毀執行緒，因此這是我的第一個建議。將程式碼轉換為非同步應該很簡單，通常按照

• 將每個函式體包裝在 中，在必要時async{...}添加return
• 通過以下方式創建尚未在庫中的任何 I/O 原語的非同步版本Async.FromBeginEnd
• 將表單的呼叫切換let r = Foo()為let! r = AsyncFoo()
• 用於Async.Parallel將 5000 個非同步對象轉換為並行執行的單個非同步

有各種教程可以做到這一點；一個這樣的網路廣播在這裡。

引用自：https://stackoverflow.com/questions/2002864