即使一个或多个失败，如何产生未知数量的期货并组合结果？

Question

我想使用 Futures 将以下顺序代码转换为并发代码，并且需要有关如何构建它的建议。

顺序：

import java.net.URL

val providers = List(
  new URL("http://www.cnn.com"),
  new URL("http://www.bbc.co.uk"),
  new URL("http://www.othersite.com")
)

def download(urls: URL*) = urls.flatMap(url => io.Source.fromURL(url).getLines).distinct

val res = download(providers:_*)

我想下载通过下载方法的可变参数传入的所有源，并将结果组合到一个 Seq/List/Set 中，无论如何，一起。当一个 Future 失败时，假设因为服务器无法访问，它应该接受所有其他的并继续前进并返回结果。 firstCompletedOf 不起作用，因为我需要所有结果，除了一个因错误而失败的结果。我想过像下面那样使用 Future.sequence ，但我无法让它工作。这是我尝试过的...

def download(urls: URL*) = Future.sequence {
  urls.map { url =>
    Future {
      io.Source.fromURL(url).getLines
    }
  }
} 

这会产生一个与 M_[Future[A_]] 不兼容的 Seq[Future[Iterator[String]]]。

A Future[Iterator[String]] 是我想要的。 （我以为我返回了一个迭代器，因为我需要稍后在迭代器上使用 reset 方法重用它。）

Answer 1

你可以使用parallel collections:

import java.net.URL

val providers = List(
  new URL("http://www.cnn.com"),
  new URL("http://www.bbc.co.uk"),
  new URL("http://www.othersite.com")
)

def download(urls: URL*) = urls.par.flatMap(url => {
  Try {
    io.Source.fromURL(url).getLines
  } match {
    case Success(e) => e
    case Failure(_) => Seq()
  }
}).toSeq

val res: Seq[String] = download(providers:_*)

或者如果您想要带有Future 的非阻塞版本：

def download(urls: URL*) = Future {
  blocking {
    urls.par.flatMap(url => {
      Try {
        io.Source.fromURL(url).getLines
      } match {
        case Success(e) => e
        case Failure(_) => Seq()
      }
    })
  }
}

val res: Future[Seq[String]] = download(providers:_*)