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

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:_*)

我想下载所有通过下载方法的varargs进入的源，并将结果组合到一个Seq/List/Set中，无论如何，在一起。当一个Future失败时，假设因为服务器无法访问，它应该接受所有其他的并继续前进并返回结果unetheless。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]]。

一个Future[Iterator[String]]是我想要的。（我想我返回一个Iterator，因为我需要稍后在Iterator上的重置方法中重用它。）

Answer 1

您可以使用并行集合:

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:_*)