未来在并行搜索和结果返回中的应用

Question

我正在创建一个服务，它遍历一个单词列表，并对每个单词检查数据库中是否有相关的操作(执行)。

我试图同时使用Futures.com来做这件事，我不确定我是否在使用最好的方法。

```javascript

类CommandDiscoveryService (commandsDAO: CommandsDAO，text: String) {

val words = text.split("\s+")

var结果=新的ListBufferOption[Execution]

//临时处理数据库上的并发搜索

// TODO将所有命令加载到内存中并检查列表?？memcache或其他缓存服务

if (words.size <= 6) {

Logger.debug("Searching for executions with text " + text )

findExecution()

}

def findExecution() ={

val lb = new ListBuffer[Future[Seq[Execution]]]()

for (word <- words) {

  lb += commandsDAO.findExecutionByName(word)

}

lb.foreach(Await.result(_, 1 seconds))

import scala.concurrent.ExecutionContext.Implicits.global // FIXME LATER

val res = lb.map {

  ftr => ftr.map{

    res => {

      if (res.size > 0 ) {

        Logger.debug("RES SIZE:" + res.size)

        res.map{ ex => results += Some(ex) }

      }

    }

  }

}

}

def getExecution()：OptionExecution ={

if (results.size > 1 ) {

  Logger.debug("ERROR_TOMANYEXECS: Found more than one execution " + results.head)

  results.foreach{

    execs => Logger.debug("ERROR_TOMANYEXECS: " + execs)

  }

  None

} else {

  if (results.size == 0 ) {

    Logger.debug("NOTHING FOUND IN RES")

    None

  } else {

    Logger.debug("FOUND RES " + results.head)

    results.head

  }

}

}

}

当我调用getExecution时，我需要已经获得搜索完成的值。我不确定对这个结果变量执行锁定是否是一个解决方案，Future[SeqExecution]上的等待已经没有恢复。

PS:我使用的是PlayFramework2.6.x和Slick来运行它。

Answer 1

您的results ListBuffer只填充了Some[Execution]，而没有使用None，因此在那里使用Option是没有意义的。我建议使用不可变集合，并重新定义findExecution方法以返回Future[List[Execution]]。

val words = text.split("\\s+").toList

def findExecution: Future[List[Execution]] = {
  val executions = words.map(commandsDAO.findExecutionByName(_)) // List[Future[Seq[Execution]]]
  val filtered = executions.map(_.filter(_.nonEmpty)) // List[Future[Seq[Execution]]
  val flattened = Future.sequence(filtered).map(_.flatten) // Future[List[Execution]]
  flattened
}

findExecution现在返回所有单词或名称的所有Executions的单个Future，但没有任何Executions的单词除外。

当我调用getExecution时，我需要已经获得搜索完成的值

更好的方法是让getExecution也返回一个Future：

def getExecution: Future[Option[Exception]] = {
  val executions = findExecution // Future[List[Execution]]
  executions.map { e =>
    if (e.size > 1) {
      // ...
      None
    } else if (e.isEmpty) {
      // ...
      None
    } else { // e.size is one
      // ...
      Some(e.head)
    }
  }
}

上述方法避免了阻塞Await调用，适合于异步播放和灵活的API。