Zukunft (<nicht abgeschlossen>), wenn Future.map Combinator

Question

Gebrauchte

Nach einem Future.map mit den ersten Future, die zweiten abhängigen Future zu verarbeiten ist immer Future(<not completed>) einmal Future.onComplete wieder aufgerufen wird. Andere Konstruktionen, die zwei Future s verwenden, zeigen dieses Verhalten niemals.

Kann jemand erklären, warum Future.onComplete aufgerufen wird, obwohl die Zukunft anscheinend nicht abgeschlossen ist?

Scala 2.12.3 wird verwendet. Ausschneiden und einfügen, um dieses Problem bei Bedarf zu bewerten.

import scala.collection.mutable.ListBuffer 
import scala.concurrent.ExecutionContext.Implicits.global 
import scala.concurrent.Future 
import scala.util.{Failure, Success} 


object FutureNotCompleted { 

    def threadNumber: String = f"${Thread.currentThread().getId.toInt}%2d" 

    ///////////////////////////////////////////////////// 
    // Future completion control 
    ///////////////////////////////////////////////////// 

    var futures: ListBuffer[Future[Any]] = ListBuffer() 

    def add(future: Future[Any]): Unit = synchronized(futures += future) 

    def remove(future: Future[Any]): Unit = synchronized(futures = futures.filter(_ != future)) 

    def loopTillCompleted: Unit = { 
    var futuresOnList = true; 

    while (futuresOnList) { 
     Thread.sleep(100) 
     for (future <- futures) { 
     if (future.isCompleted) { 
      future.value.get match { 
      case Success(v) => println(s"${threadNumber} Success: ${v}") 
      case Failure(e) => println(s"${threadNumber} Error: ${e}") 
      } 
      remove(future) 
     } 
     } 
     if (futures.size == 0) futuresOnList = false 
    } 
    } 

    ///////////////////////////////////////////////////// 
    // Future factory 
    ///////////////////////////////////////////////////// 

    def createRegisteredFuture: Future[Int] = { 
    val future = createFuture 
    add(future) 
    future 
    } 

    def createFuture: Future[Int] = Future { 
    val i = (Math.random() * 1000).toInt 
    println(s"${threadNumber} Future work start: ${i}") 
    Thread.sleep((Math.random() * 1000).toLong) 
    println(s"${threadNumber} Future work stop: ${i}") 
    if (Math.random > 0.7) throw new RuntimeException(s"${threadNumber} Error for ${i}") 
    i 
    } 

    ///////////////////////////////////////////////////// 
    // Functions exhibiting Future use conditions 
    ///////////////////////////////////////////////////// 

    def futureDoesNotComplete: Unit = { 

    val f1 = createRegisteredFuture 
    val f2 = f1.map { 
     i => createRegisteredFuture 
    } 
    // This is never completed at the time the 'onComplete' callback is called 
    f2.onComplete({ 
     case Success(j) => println(s"${threadNumber} j: ${j} ") 
     case Failure(e) => println(s"${threadNumber} f2 Failure: ${e}") 
    }) 

    loopTillCompleted 
    println(s"${threadNumber} All done.") 
    } 

    def futureCompletes: Unit = { 

    val f1 = createRegisteredFuture 

    f1.onComplete({ 
     case Success(i) => { 
     val f2 = createRegisteredFuture 
     f2.onComplete({ 
      case Success(j) => println(s"${threadNumber} i: ${i} j: ${j} ${i}+${j}=${i + j}") 
      case Failure(e) => println(s"${threadNumber} f2 Failure: ${e}") 
     }) 
     } 
     case Failure(e) => println(s"${threadNumber} f1 Failure: ${e}") 
    }) 

    loopTillCompleted 
    println(s"${threadNumber} All done.") 
    } 

    def futureCompletesFor: Unit = { 

    for { 
     f1 <- createRegisteredFuture 
     f2 <- createRegisteredFuture 
    } yield { 
     println(s"f1: ${f1} f2: ${f2}: f1+f2=${f1+f2}") 
    } 

    loopTillCompleted 
    println(s"${threadNumber} All done.") 
    } 

    def main(a: Array[String]): Unit = { 
    // futureCompletes 
    futureDoesNotComplete 
    // futureCompletesFor 
    } 

} 

Answer 1

Ich bin nicht 100% sicher, was das hier gewünschte Verhalten genau ist, aber ich würde versuchen flatMap hier, Sie wirklich wollen, die ganze Zukunft Kette Ich denke an onComplete. Wenn dies das Problem ist, dann schließen Sie gerade die äußere Zukunft ab, und das ist es, was ich vermute, seit wann Sie for (was funktioniert) es tut dasselbe wie flatMap tut.

10 Future work start: 791 
10 Future work stop: 791 
11 Future work start: 819 
11 Future work stop: 819 
11 j: 819 
1 Success: 791 
1 Success: 819 
1 All done. 

Ich würde Karte ändern flatMap, da createRegisteredFuture eine Zukunft schafft sich.

val f2 = f1.flatMap { 
     i => createRegisteredFuture 
}