在 F# 中异步操作来自 streamReader 的数据

Question

在Read large txt file multithreaded?这行，我怀疑它是否相当于向每个线程传递一个 Seq 的切片块，以及它是否会安全地处理并行性；它是 StreamReader 线程安全的吗？

这是我用来测试的代码（欢迎对使用的模式提出任何建议或批评：)）

nthreads = 4    

let Data = seq {
        use sr = new System.IO.StreamReader (filePath)
        while not sr.EndOfStream do
            yield sr.ReadLine ()
        }

let length = (Data |> Seq.length)

let packSize = length / nthreads

let groups =
     [ for i in 0..(nthreads - 1) -> if i < nthreads - 1  then Data |> Seq.skip( packSize * i )
                                                                    |> Seq.take( packSize )
                                                          else Data |> Seq.skip( packSize * i ) ]

let f = some_complex_function_modifiying_data

seq{ for a in groups -> f a }
        |> Async.Parallel
        |> Async.RunSynchronously

Answer 1

您的Data 值具有seq<string> 类型，这意味着它是惰性的。这意味着当您执行一些访问它的计算时，惰性序列将创建一个 StreamReader 的新实例并独立于其他计算读取数据。

当您向seq { .. } 块添加一些打印时，您可以很容易地看到这一点：

let Data = seq {
    printfn "reading"
    use sr = new System.IO.StreamReader (filePath)
    while not sr.EndOfStream do
        yield sr.ReadLine ()  }

因此，您的并行处理实际上很好。它将为每个单独的并行线程创建一个新的计算，因此StreamReader 实例永远不会共享。

另一个问题是这是否真的有用 - 从磁盘读取数据通常是一个瓶颈，因此在一个循环中执行操作可能会更快。即使这可行，使用Seq.length 也是一种获取长度的缓慢方法（因为它需要读取整个文件），skip 也是如此。更好（但更复杂）的解决方案可能是使用流Seek。