並行執行期貨清單

Question

我對Scala的Future和Akka非常陌生，目前，我正試圖執行一個應用程序，它執行一系列獨立任務並將結果收集在一起。

例如，我想要一個由多個任務組成的應用程序，每個任務接收一個數字，休眠幾秒鐘，然後返回「Hello」消息。

演員被實現如下：

class HelloActor extends Actor { 
    def receive = { 
    case name:Int => { 
     println("%s will sleep for %s seconds".format(name, name % 4)) 
     Thread.sleep(name % 4 * 1000) 
     sender ! "Hello %s".format(name) 
    } 
    } 
} 

主要目的是這樣實現的：

object HelloAkka extends App { 
    val system = ActorSystem("HelloSystem") 

    val helloActor = system.actorOf(Props[HelloActor], name = "helloactor") 

    implicit val timeout = Timeout(20, TimeUnit.SECONDS) 

    val futures = (1 to 10).map(num => { 
    helloActor ? num 
    }) 

    val future = Future.sequence(futures) 

    val results = Await.result(future, timeout.duration) 

    println(results) 

    system.shutdown 
} 

由於每個任務會睡眠0，1，2或3秒，我期望睡眠時間較短的任務要先執行。但是，結果是：

1 will sleep for 1 seconds 
2 will sleep for 2 seconds 
3 will sleep for 3 seconds 
4 will sleep for 0 seconds 
5 will sleep for 1 seconds 
6 will sleep for 2 seconds 
7 will sleep for 3 seconds 
8 will sleep for 0 seconds 
9 will sleep for 1 seconds 
10 will sleep for 2 seconds 
Vector(Hello 1, Hello 2, Hello 3, Hello 4, Hello 5, Hello 6, Hello 7, Hello 8, Hello 9, Hello 10) 

換句話說，所有任務都按順序執行。我想知道是否有任何方法讓我平行執行所有任務。

Answer 1

正如在評論中提到的，您將所有任務/消息發送給一個參與者，並且確保所有這些任務/消息都將按順序處理。

要並行處理任務，您需要有多個處理器參與者實例，在這種情況下，您的情況爲HelloActor。

當然，您可以創建HelloActor的多個實例，但這絕對不是最佳實踐。

對於這種類型的任務，您應該使用內置路由功能，它允許您管理員工/處理程序池並通過一個router角色與其交互，例如。

val router: ActorRef = 
    context.actorOf(RoundRobinPool(10).props(Props[HelloActor]), "router") 

... 
router ? num 
... 

請按照Akka Routing文件，以獲得更多的細節。

Answer 2

而不是按照評論和答案中的建議啓動多個參與者，我會建議執行Future中的實際任務。所以你的演員會更像是任務的協調員。例如： -

//...  

// import pipe pattern to get access to `pipeTo` method 
import akka.pattern.pipe 
import scala.concurrent.Future 

// the `Future`s will be executed on this dispatcher 
// depending on your needs, you may want to create a 
// dedicated executor for this 
class TaskCoordinatorActor extends Actor { 
    import context.dispatcher 

    def receive = { 
    case name: Int => 
     Future { 
     Thread.sleep(name % 4 * 1000) 
     "Hello %s".format(name) 
     } pipeTo sender() 
    } 
} 

上面的代碼在scala.concurrent.Future和管道執行你的任務結果到原始發件人。這樣，演員在任務完成之前不會阻止，但一旦創建了Future，則準備好接收下一條消息。

P.S .:不是發送普通整數，而是創建消息類型，這些消息類型可以明確你想讓演員做什麼。在你的情況下，它可能是例如：

case class Sleep(duration: Duration) 

Answer 3

從同一個演員發送給同一個演員的消息將按順序執行。

你有兩個選擇。

要麼創建HelloActor的新副本的每個消息，讓他們都並行執行，或修改您的HelloActor要的東西如下（可能是錯誤的進口，通過記憶去）：

import akka.pattern.pipe._ 

class HelloActor extends Actor { 
    def receive = { 
    case name:Int => { 
     println("%s will sleep for %s seconds".format(name, name % 4)) 
     Future(sleepAndRespond(name)) pipeTo sender 
    } 
} 

def sleepAndRespond(name:String) = { 
    Thread.sleep(name % 4 * 1000) 
    "Hello %s".format(innerName) 
} 
} 

這樣，執行的順序部分就是未來的管道，然後對十條消息中的每條消息進行異步執行。