响应流的昂贵异步读取

Question

在过去的几天里，我一直在尝试学习 F#，但一直遇到一些令我困惑的事情。我的“学习项目”是一些我有兴趣处理的数据的屏幕抓取工具。

在 F# PowerPack 中有一个调用 Stream.AsyncReadToEnd。我不想只为那一次通话使用 PowerPack，所以我看看他们是如何做到的。

module Downloader =
    open System
    open System.IO
    open System.Net
    open System.Collections

    type public BulkDownload(uriList : IEnumerable) =
        member this.UriList with get() = uriList

        member this.ParalellDownload() =
            let Download (uri : Uri) = async {
                let UnblockViaNewThread f = async {
                    do! Async.SwitchToNewThread()
                    let res = f()
                    do! Async.SwitchToThreadPool()
                    return res }

                let request = HttpWebRequest.Create(uri)
                let! response = request.AsyncGetResponse()
                use responseStream = response.GetResponseStream()
                use reader = new StreamReader(responseStream)
                let! contents = UnblockViaNewThread (fun() -> reader.ReadToEnd())
                return uri, contents.ToString().Length }

            this.UriList
            |> Seq.cast
            |> Seq.map Download
            |> Async.Parallel
            |> Async.RunSynchronously

他们有这个功能 UnblockViaNewThread。这真的是异步读取响应流的唯一方法吗？创建一个新线程不是真的很昂贵（我已经看到到处都是“〜1mb的内存”）。有一个更好的方法吗？这是每次Async* 调用（我可以let!）中真正发生的事情吗？

编辑：我遵循 Tomas 的建议，实际上想出了一些独立于 F# PowerTools 的东西。这里是。这确实需要错误处理，但它异步请求并将 url 下载到字节数组。

namespace Downloader
open System
open System.IO
open System.Net
open System.Collections

type public BulkDownload(uriList : IEnumerable) =
    member this.UriList with get() = uriList

    member this.ParalellDownload() =                
        let Download (uri : Uri) = async {
            let processStreamAsync (stream : Stream) = async { 
                let outputStream = new MemoryStream()
                let buffer = Array.zeroCreate<byte> 0x1000
                let completed = ref false
                while not (!completed) do
                    let! bytesRead = stream.AsyncRead(buffer, 0, 0x1000)
                    if bytesRead = 0 then
                        completed := true
                    else
                        outputStream.Write(buffer, 0, bytesRead)
                stream.Close()
                return outputStream.ToArray() }

            let request = HttpWebRequest.Create(uri)
            let! response = request.AsyncGetResponse()
            use responseStream = response.GetResponseStream()
            let! contents = processStreamAsync responseStream
            return uri, contents.Length }

        this.UriList
        |> Seq.cast
        |> Seq.map Download
        |> Async.Parallel
        |> Async.RunSynchronously

    override this.ToString() = String.Join(", ", this.UriList)

Answer 1

我认为 AsyncReadToEnd 在单独的线程上同步调用 ReadToEnd 是错误的。

F# PowerPack 还包含一个类型AsyncStreamReader，它包含流读取的正确异步实现。它有一个ReadLine 方法（异步）返回下一行，并且只从源流中下载几个块（使用异步ReadAsync，而不是在后台线程上运行）。

let processStreamAsync stream = async { 
  use asyncReader = new AsyncStreamReader(stream)
  let completed = ref false
  while not (!completed) do 
    // Asynchrnously get the next line
    let! nextLine = asyncReader.ReadLine()
    if nextLine = null then completed := true
    else
       (* process the next line *)  }

如果您想将整个内容下载为一个字符串（而不是逐行处理），那么您可以使用ReadToEnd 的AsyncStreamReader 方法。这是一个适当的异步实现，它开始（异步）下载数据块并在不阻塞的情况下重复此操作。

async { 
  use asyncReader = new AsyncStreamReader(stream)
  return! asyncReader.ReadToEnd() }

此外，F# PowerPack 是开源的并且具有许可许可证，因此使用它的最佳方法通常是将您需要的几个文件复制到您的项目中。