代码之家 › 专栏 › 技术社区 › Sun

Scala在for循环中创建一个带有Future的列表,并检查每个Future的结果

future scala

Sun · 技术社区 · 8 年前

我需要并行处理条目列表并验证结果。有谁能帮我在Scala做到这一点吗

就像下面这样,

FileList = List { "/temp/File1", "/temp/File2","/temp/File3","/temp/File4"........}

for( file <- FileList){
  Future { getRecCount(file) } // Need to create a List with all Futures
}

现在我需要检查所有期货的结果,直到它完成

getRecCount-将计算记录计数并写入文件

2 回复 | 直到 8 年前

Artavazd Balayan 8 年前

如果你只是对例外感兴趣,你可以用 Try 和使用 Future.traverse 正如@adrice727提到的paralles文件处理:

// Sequence of files for processing
val files = Seq[String]("/tmp/1.txt", "/tmp/2.txt", "/tmp/3.txt", "/tmp/4.txt",
  "/tmp/5.txt", "/tmp/6.txt")

def fileHandler(path: String): String = {
  val idx = files.indexOf(path)
  // Throw exception for each second file
  if (idx % 2 == 0) throw new Exception(s"Unable to hande $path")
  else path
}

val futures: Future[Seq[Try[String]]] = Future.traverse(files){ path =>
  Future {
    // Wrap it by Try
    Try {
      // Each file handler
      fileHandler(path)
    }
  }
}

futures.map { seq =>
  seq.map {
    case Success(s) => println(s"Result: $s")
    case Failure(ex) => println(s"Exception: ${ex.getMessage}")
  }
}

Sun 8 年前

谢谢-我就是这样实现的

val tobeprocessed:Seq[String]=Seq(“FileName1”,“FileName2”,“FileName3,”FileName4)//创建了一个具有文件名的Seq

    val latch = new CDL(tobeProcesssed.size)
      val trackFutures: Future[Seq[Try[Boolean]]] = Future.traverse(tobeProcesssed.seq) { fileNm =>
        Future {
          LOG.info("In the Futures Loop for {}", fileNm)
          Try {
                    doProcessing(fileNm)
          }
        }
      }

      trackFutures.map { seq =>
        seq.map {
          case Success(state) => {
            if (state) {
              LOG.info("On Sucess - {} - {}", latch.getCount, state)
              return = 0
              latch.countDown()

            } else {
              LOG.error("Failed - {} - {}", latch.getCount, state)
              while (latch.getCount > 0) {
                return  = 1
                latch.countDown()
              }
            }

          }
          case Failure(exception) => {
            LOG.error("Versioning Failed -{} - {}", latch.countDown(), exception)
            while (latch.getCount > 0) {
              return = = 1
              latch.countDown()
            }
          }
        }
      }
      latch.await()